不引 Redis 行不行？Litestar 的 PsycoPgChannelsBackend 把 PostgreSQL 变成广播中枢-平芜编程栈

不引 Redis 行不行？Litestar 的 PsycoPgChannelsBackend 把 PostgreSQL 变成广播中枢

【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar

Litestar 的 Channels 子系统里藏着一个"数据库即 broker"的后端：PsycoPgChannelsBackend。它把 PostgreSQL 原生的LISTEN/NOTIFY封装成标准ChannelsBackend，如果你的项目本来就在用 psycopg3，又多进程应用需要互相同步消息，一个 DSN 就能拿到跨进程事件广播，不用额外跑 Redis。这篇文章从启动到关闭、从故障到选型，把整个实现讲透。

它适合谁用？

先想一个具体场景：管理后台改了一条配置，希望所有在线的 WebSocket 客户端都收到"配置已更新"。消息不大、频率不高，但进程可能部署了好几份。

这时你有两条路：

引入 Redis，搭一套 Pub/Sub；
或者把已经在用的 PostgreSQL 直接当消息中转站。

第二条路就是PsycoPgChannelsBackend的定位。它和 Channels 架构里其他角色（抽象基类 ChannelsBackend、管理订阅与路由的ChannelsPlugin、包装单条事件流的Subscriber）的关系，见 docs/usage/channels.rst 里的术语表和流程图，概念本身不复杂，这里不再展开。

两个前提要确认：

你的 psycopg3 版本 ≥ 3.2.4（文档中 Backends 一节有明确版本要求）；
你能接受没有历史回放——这点后文单独说，它是这个后端最大的限制。

三步接入 🔌

整个接入就是"建后端 → 挂插件 → 用插件"，三步：

from litestar.channels import ChannelsPlugin from litestar.channels.backends.psycopg import PsycoPgChannelsBackend backend = PsycoPgChannelsBackend(pg_dsn="postgresql://user:pass@localhost:5432/mydb") channels = ChannelsPlugin( backend=backend, channels=["general", "notifications"], create_ws_route_handlers=True, )

三步拆开看：

建后端：构造函数只收一个pg_dsn。你不用操心连接细节——源码里所有连接都是AsyncConnection.connect(self._pg_dsn, autocommit=True)建出来的，LISTEN/NOTIFY不需要事务包裹，autocommit 是刻意为之；
挂插件：ChannelsPlugin会把自己注册为应用依赖（注入键名就是channels），并在应用启动/关闭时替你调用backend.on_startup()/on_shutdown()。生命周期你完全不用管；
用插件：在 handler 里直接注入channels，调channels.publish(data, "general")发消息；create_ws_route_handlers=True会为每个声明的频道自动生成 WebSocket 路由。频道名不在channels列表里就会抛ChannelsException，想动态开频道就设arbitrary_channels_allowed=True。

消息发出去之后，到底经历了什么？

把LISTEN/NOTIFY想象成数据库的公告栏：谁LISTEN了某个频道，数据库就把NOTIFY的载荷原样贴给谁。同一台数据库上的所有进程、所有连接都共享这块公告栏——这就是"跨进程广播"的全部魔法来源。

一条消息的完整旅程用箭头写出来就是：

你的代码 publish() → 插件内部队列（同步非阻塞） → 后台 pub worker → backend.publish()：开一条新短连接，逐频道 NOTIFY → PostgreSQL 按 LISTEN 分发 → 专用监听连接上的 _listen 协程捞到通知 → 事件队列（put_nowait） → stream_events() 逐个 yield → 插件 sub worker 分发给各 Subscriber → 你的 WebSocket 客户端

有两个地方值得留意：

发布用短连接：每次publish都新开一条连接、发完即关，不和接收通知的长连接抢资源。核心 SQL 长这样：

async with await AsyncConnection.connect(self._pg_dsn, autocommit=True) as conn: for channel in channels: await conn.execute( SQL("NOTIFY {channel}, {data}").format(channel=Identifier(channel), data=dec_data) )

频道名走Identifier转义，数据先decode("utf-8")变回文本（NOTIFY载荷本质是字符串）；接收端再encode("utf-8")，维持了 Channels 接口"事件一律是 bytes"的约定。

插件层的publish()是同步的：它只是把消息扔进内部队列就返回，不保证已经落到数据库。需要确认送达时用await channels.wait_published(data, channels)，它会绕过队列直接调backend.publish。

一次订阅，是怎么变成 LISTEN 语句的？

subscribe的实现（litestar/channels/backends/psycopg.py 里的subscribe/unsubscribe）有三个设计点：

差集计算：只对"请求的频道 - 已订阅的频道"执行LISTEN，重复订阅天然幂等。unsubscribe用交集，逻辑对称；
全程持锁：所有订阅变更都在self._listener_lock里串行执行，_subscribed_channels集合和监听连接不会被并发修改。单测test_subscription_mutations_are_serialized用假连接统计并发操作数，断言它永远不超过 1，就是为了钉死这个约束；
变更前先停监听：LISTEN/UNLISTEN和notifies()迭代共用同一条监听连接，所以改订阅前要先_stop_listener()，改完在finally里（除非正在关闭）重新_start_listener()。这保证了"订阅集合"和"连接上实际监听的状态"永远一致。

从启动到关闭，都发生了什么

启动（on_startup）做四件事：

新建AsyncExitStack和事件队列（重置状态，允许复用）；
连出专用监听连接_listener_conn——长期驻留，只负责收通知，登记进 exit stack，异常路径下也能被统一关闭；
置位标志、启动后台监听协程_listen()；
插件侧随后会拿channels列表调一次subscribe，把初始订阅挂上。

关闭（on_shutdown）反着来，且在锁内执行：

置_shutting_down = True（后续订阅变更的finally分支看到它就不会再重启监听）；
_stop_listener()停掉监听协程；
清空订阅集合；
exit_stack.aclose()关闭监听连接。

_stop_listener()本身是"先礼后兵"：先设_stop_listening = True让循环自己退出，最多等 5 秒（_STOP_LISTENER_TIMEOUT）；超时才cancel()强制取消并等它收尾。这个 5 秒窗口给了监听循环体面退出的机会，避免每次订阅变更都靠取消任务硬切。

监听器坏了会怎样？🧯

监听协程_listen()的主循环长这样：

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(timeout=0.1)每 0.1 秒（_LISTEN_POLL_INTERVAL）一轮，既能及时捞到通知，也给循环留下检查停止标志的机会。异常处理上分两类：

CancelledError原样重抛——被强制取消就取消，不多事；
其他任何异常（连接断了、数据库重启……）被装进事件队列，当作一条"事件"发给消费者。

消费者stream_events()那边接住：

event = await self._event_queue.get() if isinstance(event, Exception): raise event if event[0] in self._subscribed_channels: yield event

第一段就是故障传播：监听端死了，异常顺着队列一路抛到插件的订阅 worker，上层立刻感知到"这条链路断了"，而不是静默丢消息。

第二段处理另一个隐蔽竞态：UNLISTEN发出去的瞬间，队列里可能还躺着旧频道的通知。所以每条事件取出后都要再校验一次"该频道现在还在订阅集合里吗"，不在就丢弃。单测test_stream_events_filters_queued_events_after_unsubscribe和test_stream_events_propagates_listener_failures分别钉死了这两个行为——整个测试文件用假连接对象伪造notifies()，不依赖真实数据库，是理解这套内部状态的最佳阅读入口（tests/unit/test_channels/test_psycopg_backend.py）。

它做不了的一件事：历史回放

get_history()直接抛NotImplementedError。LISTEN/NOTIFY是纯广播语义，数据库不会替你存"过去贴过什么"。

受影响的入口有三个，都会走到get_history：

channels.subscribe(..., history=N)；
channels.put_subscriber_history(...)；
生成 WebSocket 路由时设的ws_handler_send_history。

如果你的需求是"客户端连上来先补发最近 N 条"，换后端就行：RedisChannelsStreamBackend（基于 Redis Streams）或MemoryChannelsBackend(history=20)都支持。同为 PostgreSQL 实现的AsyncPgChannelsBackend（asyncpg 驱动）也一样不支持历史，所以这不是 psycopg 驱动特有的缺陷，而是LISTEN/NOTIFY机制的天花板。

五个后端放一起看

后端	消息介质	历史回放	一句话选型建议
`MemoryChannelsBackend`	进程内存	支持（`history`参数）	单进程、测试、本地开发；速度最快
`RedisChannelsStreamBackend`	Redis Streams	支持	要补发历史，Redis 阵营首选
`RedisChannelsPubSubBackend`	Redis Pub/Sub	不支持	低延迟纯扇出，对延迟最敏感
`PsycoPgChannelsBackend`	PostgreSQL（psycopg3）	不支持	已深度使用 psycopg3，不想加中间件
`AsyncPgChannelsBackend`	PostgreSQL（asyncpg）	不支持	已深度使用 asyncpg；功能与上一行对齐，参数名不同