把 Litestar 的 psycopg 后端当消息 BrokerPostgreSQL LISTEN/NOTIFY 事件广播实战指南【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar多进程部署的 WebSocket 应用里最常见的痛点是进程 A 收到的消息进程 B 的客户端看不见。标准答案是引入 Redis 做 broker但如果你的数据库里本来就有 PostgreSQLLitestar Channels 的PsycoPgChannelsBackend提供了一条更省的路——直接把 PostgreSQL 原生的LISTEN/NOTIFY机制封装成跨进程事件广播后端约 120 行代码无额外中间件。这篇指南拆解它的连接模型、并发设计和硬边界帮你判断它值不值得进生产。场景先行不想为广播多养一个 Redis假设你有一个管理后台管理员改了配置需要通知所有在线实例刷新缓存。单体部署时用内存广播即可一旦gunicorn -w 4或上了容器编排进程间就隔断了。传统方案是加 Redis Pub/Sub——多一个要部署、监控、设密码的组件。PsycoPgChannelsBackend的思路是数据库即 brokerPostgreSQL 天然支持LISTEN/NOTIFY任何连到同一数据库的连接执行LISTEN后都能收到任意进程NOTIFY发出的消息跨进程一致性由数据库保证。适合低频到中等频率的广播场景配置变更、通知提醒、列表刷新前提是项目已经深度使用 psycopg3 异步驱动。它要求psycopg 包 3.2.4 及以上版本见 官方文档 Backends 一节。一分钟看懂一条专线加一个信箱 这个后端的运行模型可以压缩成一句话监听走专线发布走快线中间隔着一个信箱。监听连接专线on_startup时开一条长期驻留的AsyncConnection专门用来跑LISTEN并接收NOTIFY用AsyncExitStack登记关闭时统一回收发布连接快线每次publish都现开一条短连接NOTIFY 完即弃。绝不让发布流量挤占监听专线——就像客服专线不会用来打营销电话事件队列信箱监听协程收到通知后put_nowait进asyncio.Queuestream_events()只负责取件。网络 IO 和消费速率就此解耦消费慢不会阻塞收包。所有连接都以autocommitTrue打开因为LISTEN/NOTIFY本就不需要事务包裹。from litestar import Litestar from litestar.channels import ChannelsPlugin from litestar.channels.backends.psycopg import PsycoPgChannelsBackend backend PsycoPgChannelsBackend(pg_dsnpostgresql://user:passlocalhost:5432/mydb) app Litestar( plugins[ ChannelsPlugin( backendbackend, channels[general, notifications], # 频道白名单 create_ws_route_handlersTrue, # 为每个频道自动生成 WS 路由 ) ] )生命周期不用自己管ChannelsPlugin 通过 lifespan 在启动时调backend.on_startup()、关闭时调on_shutdown()。上手路径两个容易踩的参数差异接入时真正需要留意的只有两点。频道必须白名单制。未开启arbitrary_channels_allowedTrue时向未声明的频道发布或订阅会抛ChannelsException开启后插件会动态建频道create_ws_route_handlers也会退化为单个{channel_name:str}通配路由。publish是投递不是送达。插件层的publish()是同步非阻塞方法数据先进内部队列由后台 worker 异步写库返回时不保证已发布。需要立即确认落库时用await channels.wait_published(data, channels)直连backend.publish。这条约定与后端选型无关是插件层的行为但用NOTIFY这类发完即忘机制时尤其容易误判。实现拆解锁、停表与异常信箱这部分挑三个最值得看的细节完整逻辑见 源码。为什么改一次订阅要停表再重启LISTEN/UNLISTEN和接收通知的notifies()迭代器共用同一条监听连接。一边执行 SQL 变更、一边迭代收包状态就可能不一致。所以subscribe的策略是拿asyncio.Lock→ 先_stop_listener()停掉消费循环 → 执行LISTEN→finally里重启除非正在关闭。async with self._listener_lock: # 所有订阅变更在此串行化 to_listen requested - self._subscribed_channels # 差集幂等 if not to_listen: return await self._stop_listener() # 变更期间先停消费 try: for ch in to_listen: await self._listener_conn.execute( SQL(LISTEN {c}).format(cIdentifier(ch))) self._subscribed_channels.add(ch) finally: if not self._shutting_down: self._start_listener() # 无论如何恢复监听两处理得讲究一是差集计算让重复订阅幂等二是_stop_listener采用先礼后兵——置_stop_listening标志后最多等 5 秒_STOP_LISTENER_TIMEOUT让循环自然退出超时才cancel()。单测test_subscription_mutations_are_serializedtests/unit/test_channels/test_psycopg_backend.py用假连接注入并发subscribe断言活跃操作数峰值恒为 1验证了锁的串行化约束。监听挂了为什么消费者能感知_listen()循环里CancelledError原样重抛其余任何异常连接断开等不做静默吞掉而是把异常对象本身塞进事件队列。stream_events取到Exception直接raiseasync def stream_events(self): while True: event await self._event_queue.get() if isinstance(event, Exception): raise event # 监听故障直传消费者 if event[0] in self._subscribed_channels: # 退订竞态二次校验 yield event这套设计把故障也当成一种事件走同一条通道传递插件的订阅 worker 一抛就感知到后端已坏不会傻等一条永远不会来的消息。对应的test_stream_events_propagates_listener_failures用一个必然抛RuntimeError的假notifies()迭代器验证了这一点——测试全程不碰真实数据库。发布端如何防注入、保证 fanoutNOTIFY载荷是文本所以 bytes 先decode(utf-8)接收端再encode还原守住事件一律 bytes的接口约定。频道名则通过psycopg.sql.Identifier转义后拼入NOTIFY {channel}, {data}动态 SQL 也逃不出注入风险。对每个目标频道各发一条NOTIFYfanout 由数据库天然完成——所有LISTEN该频道的连接无论属于哪个进程都会收到。边界与替代它不支持什么什么时候该换掉它 最硬的一条边界get_history()直接raise NotImplementedError()。连锁反应是——channels.subscribe(..., historyN)、put_subscriber_history()、生成路由时的ws_handler_send_history全会炸。需要客户端连上后补发最近 N 条这类历史回放的它不是选项。同源实现的AsyncPgChannelsBackendasyncpg 后端行为一致二者唯一实质差异是构造参数pg_dsnvsdsn/make_connection。五个后端横向对比依据 docs/usage/channels.rst后端Broker历史回放适用场景MemoryChannelsBackend进程内存支持history参数测试、本地开发、单进程RedisChannelsPubSubBackendRedis Pub/Sub否低延迟跨进程广播RedisChannelsStreamBackendRedis Streams支持要历史 高吞吐AsyncPgChannelsBackendPostgreSQLasyncpg否已用 asyncpg 的项目PsycoPgChannelsBackendPostgreSQLpsycopg3否已用 psycopg3 的项目可执行的判断项目里已经在用 psycopg3、广播频率中等偏低、不需要历史回放——用它少一个组件广播是核心路径、吞吐和延迟是硬指标——换 Redis 系需要补发历史——RedisChannelsStreamBackend或MemoryChannelsBackend(history20)。避坑清单接入前逐条确认psycopg 版本 ≥ 3.2.4且用的是异步驱动psycopg.AsyncConnectionDSN 在构造时一次给死后端内部所有连接监听 每次发布都复用这个 DSN不要配置任何历史相关参数history、ws_handler_send_history、put_subscriber_history会触发NotImplementedError动态频道记得开arbitrary_channels_allowedTrue否则ChannelsExceptionpublish返回 ≠ 已落库需要确认就用wait_published监听连接源码中没有自动重连逻辑长时运行建议对连接健康做外部监控——它的异常传播机制保证你会知道它挂了但恢复要靠你自己。单测的三个场景串行化、错误传播、退订过滤全部基于注入假连接完成是理解这个后端内部状态最快的阅读入口。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考