Litestar PsycoPgChannelsBackend 源码走读：LISTEN/NOTIFY 如何撑起跨进程事件广播

Litestar PsycoPgChannelsBackend 源码走读LISTEN/NOTIFY 如何撑起跨进程事件广播【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestarLitestar 的 Channels 子系统里PsycoPgChannelsBackend把 PostgreSQL 原生的LISTEN/NOTIFY机制封装成符合 ChannelsBackend 契约的事件广播后端。它解决的是一个很具体的问题应用已经跑在 PostgreSQL 上不想再引入 Redis 这类独立 broker但需要多进程之间互相广播事件比如推给 WebSocket 客户端。全部实现约 120 行位于 litestar/channels/backends/psycopg.py要求 psycopg3 ≥ 3.2.4 的异步驱动。为什么是一条长连接 一堆短连接先看架构里各角色的分工这决定了后端的职责边界BackendChannelsBackend抽象类litestar/channels/backends/base.py的实现只负责与 broker 通信——发布、接收、查询历史PluginChannelsPlugin 管理订阅者路由应用启动时调用backend.on_startup()关闭时调用backend.on_shutdown()后端无需自己管生命周期Subscriber包装单个事件流通过iter_events()或run_in_background()消费。PsycoPgChannelsBackend内部维护两类连接这是理解全文的钥匙# litestar/channels/backends/psycopg.py self._listener_conn await AsyncConnection[Any].connect(self._pg_dsn, autocommitTrue) # 启动时建立启动时开一条长期驻留的监听连接专门挂LISTEN收NOTIFY通知而发布消息时psycopg.pyasync def publish(self, data: bytes, channels: Iterable[str]) - None: dec_data data.decode(utf-8) async with await AsyncConnection[Any].connect(self._pg_dsn, autocommitTrue) as conn: # 每次新建短连接 for channel in channels: await conn.execute(SQL(NOTIFY {channel}, {data}).format(channelIdentifier(channel), datadec_data))为什么发布不复用监听连接因为notifies()迭代与execute()共享同一条连接的状态机发布流量会打断监听循环而NOTIFY本身是无状态指令用短连接发完即弃最干净。代价是每次发布多一次 TCP 握手——对低频广播场景可以接受这也是该后端的适用前提。两个细节值得注意data: bytes先decode(utf-8)是因为NOTIFY的载荷本质是文本频道名经Identifier()转义后拼入 SQL避免注入。所有连接都带autocommitTrue因为LISTEN/NOTIFY不需要事务包裹。一条事件从 NOTIFY 到订阅者的旅程以某进程向notifications频道发布一条消息为主线走一遍发布端。ChannelsPlugin.publish()是同步非阻塞的——消息先丢进内部队列由后台_pub_worker异步调backend.publish()plugin.py。需要立即确认落地时用await channels.wait_published(...)绕过队列。数据库 fanout。PostgreSQL 把NOTIFY投递给所有LISTEN该频道的连接跨进程广播由数据库保证后端自己不做任何路由。监听端。每个进程的后端各有一条监听连接。后台监听协程轮询接收async def _listen(self) - None: while not self._stop_listening: async for notify in self._listener_conn.notifies(timeout_LISTEN_POLL_INTERVAL): self._event_queue.put_nowait((notify.channel, notify.payload.encode(utf-8)))notifies(timeout0.1)每 0.1 秒回查一次_stop_listening标志这既是收通知的循环也是可被礼貌停止的循环。收到通知后立即入队payload 重新encode(utf-8)回到接口约定的 bytes 形态。 4.消费端。插件的_sub_worker调backend.stream_events()按频道分发给订阅者async def stream_events(self) - AsyncGenerator[tuple[str, bytes], None]: while True: event await self._event_queue.get() if isinstance(event, Exception): raise event if event[0] in self._subscribed_channels: # 二次校验频道仍被订阅 yield event队列把网络 IO 与业务消费解耦消费方慢不会反压数据库连接。订阅变更为什么必须先停监听subscribe/unsubscribe是整个后端并发最敏感的地方因为它要改监听连接本身的状态async def subscribe(self, channels: Iterable[str]) - None: requested_channels set(channels) async with self._listener_lock: channels_to_subscribe requested_channels - self._subscribed_channels if not channels_to_subscribe: return await self._stop_listener() try: for channel in channels_to_subscribe: await self._listener_conn.execute(SQL(LISTEN {channel}).format(channelIdentifier(channel))) self._subscribed_channels.add(channel) finally: if not self._shutting_down: self._start_listener()四个动作环环相扣差集计算只对请求集 − 已订阅集执行LISTEN天然幂等锁串行化所有订阅变更在_listener_lock内完成并发请求排队先停后改LISTEN执行期间监听循环还在迭代同一条连接的notifies()状态会互相干扰所以先_stop_listener()暂停finally里除非正在关闭再重启退订对称unsubscribe用请求集 ∩ 已订阅集算UNLISTEN目标流程完全镜像。这里有个容易踩的坑退订后队列里可能还躺着旧频道的事件notifies()与stream_events之间存在竞态。所以stream_events取出事件后要再校验一次event[0] in self._subscribed_channels才放行——双重校验消除了这个窗口。设计决策剖析队列、锁、超时与异常传播把主线里出现的机制集中过一遍看每个决策点的取舍机制现状如果不这么写事件队列中转监听任务只负责入队stream_events只负责出队直接 yield 会让消费方阻塞反压到数据库连接且监听与消费共享异常域_listener_lock订阅变更串行执行并发subscribe会交错执行LISTEN与notifies()连接状态不可预测_STOP_LISTENER_TIMEOUT 5.0停监听先等 5 秒自然退出超时才cancel()直接 cancel 可能在execute()中途打断连接留下半完成的订阅变更异常入队而非重抛_listen捕获异常后put_nowait(exc)stream_events取出即raise监听协程静默死亡故障无人知晓直到某次连接操作偶发报错停止监听的先礼后兵逻辑psycopg.pyasync def _stop_listener(self) - None: self._stop_listening True try: async with asyncio.timeout(_STOP_LISTENER_TIMEOUT): await self._listener_task except TimeoutError: self._listener_task.cancel() # 5 秒内没退出强制取消 with suppress(asyncio.CancelledError, TimeoutError): ...而异常传播链psycopg.py把监听故障显式送进队列except asyncio.CancelledError: raise except Exception as exc: # 监听故障转发给流消费者 self._event_queue.put_nowait(exc)CancelledError原样重抛保证取消语义不被吞掉其他异常如连接中断则变成队列里的特殊事件让插件的订阅 worker 第一个感知到故障。生命周期收尾同样讲究监听连接登记在AsyncExitStack里on_shutdown统一aclose()异常路径下连接也不会泄漏。能力边界历史回放在这条路上走不通get_history()的实现只有一行psycopg.pyasync def get_history(self, channel: str, limit: int | None None) - list[bytes]: raise NotImplementedError()❌ 因此以下用法全部会失败ChannelsPlugin.subscribe(..., historyN)、put_subscriber_history()、以及create_ws_route_handlersTrue时设置ws_handler_send_history0plugin.py 的自动 WebSocket 路由会调put_subscriber_history。原因很直接LISTEN/NOTIFY是纯广播语义PostgreSQL 不为没人监听的通知留存任何记录数据库里没有历史可查。需要历史能力时的替代路径换RedisChannelsStreamBackend基于 Redis Streams支持历史与持久化或MemoryChannelsBackend(history20)进程内历史仅单进程可用依据见 docs/usage/channels.rst 的 Backends 一节。另外两点边界发布端每次新建短连接高吞吐下握手开销会显现插件层未开arbitrary_channels_allowedTrue时向未声明频道发布/订阅会抛ChannelsException。横向对比与选型结合 docs/usage/channels.rst 的说明五类后端的核心差异如下后端Broker历史回放依赖适用场景MemoryChannelsBackend进程内存✅需history参数无测试、本地开发、单进程RedisChannelsPubSubBackendRedis Pub/Sub无redis低延迟跨进程广播不需要历史RedisChannelsStreamBackendRedis Streams支持redis需要历史与持久化的跨进程广播AsyncPgChannelsBackendPostgreSQLasyncpg无asyncpg已深度使用 asyncpg 的项目PsycoPgChannelsBackendPostgreSQLpsycopg3无psycopg ≥ 3.2.4已深度使用 psycopg3 的项目两条 PostgreSQL 后端功能对齐实现细节略有差异asyncpg 版用参数化查询SELECT pg_notify($1, $2);且监听靠add_listener回调psycopg 版用SQL/Identifier构建NOTIFY语句监听靠notifies()迭代因此多了停监听/重启监听这一整套逻辑。选型建议项目已用 psycopg3 且广播频率低通知、管理端刷新这类选PsycoPgChannelsBackend复用现有数据库基础设施少一个中间件广播是核心链路、吞吐与延迟是硬指标优先 Redis 系后端Pub/Sub 覆盖无历史场景需要历史则上 Streams需要给 WebSocket 客户端回放最近 N 条PostgreSQL 两个后端都出局换支持get_history的后端。验证与延伸阅读tests/unit/test_channels/test_psycopg_backend.py 用假连接对象注入_listener_conn与伪造的notifies()在无真实数据库下验证了三条关键行为契约test_subscription_mutations_are_serializedasyncio.gather并发两次subscribe断言连接上活跃操作数峰值恒为 1验证锁的串行化约束test_stream_events_propagates_listener_failures监听器持续抛RuntimeError(listener failed)断言stream_events会把该异常抛给消费者test_stream_events_filters_queued_events_after_unsubscribe队列里预置已退订频道和保留频道的事件断言前者被丢弃、后者正常放行。结合 litestar/channels/plugin.py 可看插件如何消费stream_events并分发到订阅者litestar/channels/backends/asyncpg.py 是同一契约的另一驱动实现适合对照阅读。小结PsycoPgChannelsBackend的设计精髓在于用最少的活动部件承载最脆的并发一条长监听连接、一堆短发布连接、一个事件队列、一把锁。订阅变更串行化 先停后改保证连接状态始终一致双重频道校验 异常入队分别消除了退订竞态和监听静默死亡。它适合数据库即 broker的轻量跨进程广播前提是接受无历史回放、发布端按次建连接的代价超出这个边界就该转向 Redis 系后端。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考