如何在SQLAlchemy+asyncpg中实现Postgres NOTIFY/LISTEN？

核心问题在于：asyncpg的add_listener需要绑定到原生的、持续活跃的asyncpg连接，而SQLAlchemy的异步连接是经过封装的，直接调用asyncpg的方法如果没拿到正确的连接实例，就会监听失败。下面是具体实现步骤：

1. 正确获取原生asyncpg连接 #

SQLAlchemy的AsyncConnection对象可以通过.connection.driver_connection属性拿到底层的asyncpg连接。但要注意：

• 不能用会话（AsyncSession）里的连接，因为会话结束后连接会被放回连接池，监听会失效
• 需要单独从连接池获取一个长期持有的连接，专门用于监听通知

2. 实现监听初始化与回调 #

首先定义一个异步回调函数，用于处理收到的Postgres通知：

async def handle_notification(conn, pid, channel, payload):
    # 处理收到的通知，比如解析触发器发送的payload
    print(f"收到通知: 频道={channel}, 内容={payload}")
    # 推送给所有在线WebSocket客户端
    for websocket in active_websockets:
        await websocket.send_text(payload)

然后在FastAPI启动时，初始化监听：

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from sqlalchemy.ext.asyncio import create_async_engine, AsyncConnection

app = FastAPI()

# 维护活跃的WebSocket连接集合
active_websockets = set()

# SQLAlchemy异步引擎
engine = create_async_engine("postgresql+asyncpg://user:password@localhost/dbname")

# 用于监听的长期连接
listen_conn: AsyncConnection | None = None

@app.on_event("startup")
async def startup_event():
    global listen_conn
    # 直接从引擎获取连接，避免被会话回收
    listen_conn = await engine.connect()
    # 获取原生asyncpg连接实例
    raw_conn = listen_conn.connection.driver_connection
    # 监听触发器指定的频道（和NOTIFY命令的频道一致）
    await raw_conn.add_listener("your_notify_channel", handle_notification)

@app.on_event("shutdown")
async def shutdown_event():
    global listen_conn
    if listen_conn:
        raw_conn = listen_conn.connection.driver_connection
        await raw_conn.remove_listener("your_notify_channel", handle_notification)
        await listen_conn.close()

3. WebSocket端点实现 #

添加WebSocket端点，维护活跃连接列表：

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    active_websockets.add(websocket)
    try:
        while True:
            # 保持连接，不需要接收前端消息可忽略此步骤
            await websocket.receive_text()
    except WebSocketDisconnect:
        active_websockets.remove(websocket)

关键注意事项 #

• 频道名称一致性：确保add_listener的频道和Postgres触发器里NOTIFY的频道完全一致（Postgres频道名大小写敏感）
• 连接长期持有：用于监听的连接不能被放回连接池，所以直接用engine.connect()获取，而非通过会话
• 重连机制：如果Postgres连接断开，监听会失效，可以在回调中或启动定时任务检查连接状态，实现自动重连
• Payload格式化：触发器建议发送JSON格式的内容，方便后端解析后推送，示例触发器代码：

  CREATE OR REPLACE FUNCTION notify_insert()
  RETURNS TRIGGER AS $$
  BEGIN
      PERFORM pg_notify('your_notify_channel', row_to_json(NEW)::text);
      RETURN NEW;
  END;
  $$ LANGUAGE plpgsql;
  

内容的提问来源于stack exchange，提问作者Aleksei Khatkevich