refactor!: EventQueue 收敛为单向业务事件通道,出站缓冲归 ws 适配器

协议分流后 reply_queue 只剩 WebsocketConnecter 一个使用者(emitter 不再
回流应答、webhook 完全不用),它已不是"连接层与分发层之间的通道",而是
ws 私有的出站缓冲,却仍挂在共享 EventQueue 上——queue.py 的模块注释
"连接层与分发层之间唯一的跨任务通道"对 reply_queue 不再成立。

删除 EventQueue.reply_queue 与 put_reply/get_reply,EventQueue 只承载
业务事件(op=0,单向)。ws 的出站帧改由 WebsocketConnecter 私有
_outbound 队列承载,heartbeat_helper 与协议应答生产者投递、reply_helper
单任务发送——维持 aiohttp ws 单写者不变量,生产者也不会被发送阻塞。

破坏性变更:EventQueue.reply_queue / put_reply / get_reply 不再存在
(这两个方法本就只由 SDK 内部使用,非用户面)。

文档同步 queue.py 注释、ARCHITECTURE(图与心跳/应答路径)、API 的
EventQueue 说明、CHANGELOG;测试 80 全绿,ruff/pyright 干净。
This commit is contained in:
snowy-yang committed 2026-09-20 22:02:07 +08:00
1 parent 7accbd1ead
commit 4d19db24e1
6 files changed
+24 -23

No files matched your search

+1
View File
@@ -30,6 +30,7 @@
- 事件分发热路径减负:`emit()` 先取事件名、无 listener 时不再构造 `Event`(每条未订阅事件/协议帧省一次对象与 dict 拷贝);`Event.data` 改为惰性拷贝(只用 `.typed`/`.raw` 的 handler 不再触发 `dict(d)`)
- ws 重连收尾补 `await`:取消 `reply`/`heartbeat` 任务后 `gather(..., return_exceptions=True)` 取回结果,避免 "Task exception was never retrieved" 并确保它们在下一轮重连前真正停下
- `EventQueue.put_event`/`put_reply` 去掉多余的 `return await`
- `EventQueue` 收敛为**单向业务事件通道**:删除 `reply_queue` 与 `put_reply`/`get_reply`(协议分流后引用它的只剩 `WebsocketConnecter`)。ws 的出站帧(心跳 + 协议应答)改由 `WebsocketConnecter` 私有出站缓冲承载、`reply_helper` 单任务发送(维持 aiohttp ws 单写者不变量);webhook 的应答直接写 HTTP 响应体。破坏性变更:`EventQueue.reply_queue` 与 `put_reply`/`get_reply` 不再存在
### 移除
+1 -1
View File
@@ -19,7 +19,7 @@ from qqbotsdk import (
| `run_loop(emitter=None)` | 异步入口:`asyncio.run(run_loop(ee))` 的内部实现,可自行 await |
| `EventEmitter(queue=None)` | 分发主体,`@ee.on(..., background=True)` 注册 handler(可选后台并发);`queue` 不传自动创建 |
| `ee.services` | `dict[type, Any]`,按类型登记可注入组件;`run_loop` 装配 SDK 内置组件,也可登记自定义类型 |
| `EventQueue()` | 事件/应答队列;`EventEmitter` 默认自建,需多处共享时手动传入(`ee.queue` 取实例) |
| `EventQueue()` | 业务事件通道(连接层 → 分发层,单向);`EventEmitter` 默认自建,需多处共享时手动传入(`ee.queue` 取实例) |
| `Config` | frozen dataclass,环境变量一次性读齐(见下文) |
| `Connecter` | 连接适配器 Protocol(接口),按 `CONNECTER` 自动选择实现 |
| `BaseProtocol` | 协议处理器抽象基类(`protocol.py`),自定义接入方式时继承并实现 `on_frame(payload)` |
+2 -2
View File
@@ -17,7 +17,7 @@ flowchart TB
end
subgraph 分发层["分发层 emitter.py"]
Q[("EventQueue\nmsg_queue / reply_queue")]
Q[("EventQueue\n业务事件通道")]
EE["EventEmitter\nop=0/t 路由 + 参数注入\nhandler 异常隔离\nbackground 后台并发"]
end
@@ -107,7 +107,7 @@ sequenceDiagram
## 关键设计决策
1. **业务事件走管线,协议帧走专有处理器**:只有 op=0 的业务事件经 `EventQueue` 异步流转给 `EventEmitter` 分发;HELLO/READY/INVALID_SESSION/op=13 等协议帧由 connecter 逐帧调用其专有协议处理器(`BaseProtocol.on_frame`)就地处理。这样用户 handler 不会看到协议帧,且"断线前重置会话"这类时序不再取决于分发泵的进度——`INVALID_SESSION` 的会话清空在 connecter `break` 重连前已同步完成,不会误 RESUME 一个已死会话。代价是多一跳队列延迟(可忽略)。
2. **协议层只管会话语义,心跳归传输层**:`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号(说什么),`WebsocketConnecter` 负责连接与心跳(怎么保活)——心跳任务随连接生灭(HELLO 给出间隔后启动,断线即取消),断线期间不在 reply 队列堆积过期心跳,也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值由 connecter 写回(ws 走 reply 队列,webhook 走 HTTP 响应体)。
2. **协议层只管会话语义,心跳归传输层**:`WebsocketProtocol` 负责 IDENTIFY/RESUME/序列号(说什么),`WebsocketConnecter` 负责连接与心跳(怎么保活)——心跳任务随连接生灭(HELLO 给出间隔后启动,断线即取消),断线期间不在出站缓冲堆积过期心跳,也免去了调度器的生命周期管理。协议处理器的应答经 `on_frame` 返回值交给 connecter:ws 与心跳一起进 connecter 私有的出站缓冲、由单任务发送(维持 ws 单写者不变量),webhook 则直接写进 HTTP 响应体。
3. **看门狗两层互补**(见 `connecter.py` 模块注释):静默超时抓连接彻底失联,ACK 期限检查抓"有事件流但心跳已死"。阈值 `_SILENCE_FACTOR=1.1`、`_ACK_FACTOR=1.2` 是按 ACK 自然节奏校准的,**不能压到 1× 整**,否则正常抖动会周期性误杀(误杀由 RESUME 兜底,不丢消息)。重连退避只在连接稳定存活(`_STABLE_SECONDS`)后才归一,抖动的服务端不会被 1s 间隔反复探测。
4. **handler 异常就地隔离**:`_invoke` 捕获记日志,`dispatch` 主循环另有兜底;单条事件的单个 handler 失败不影响进程。`background=True` 的 handler 进后台任务(emit 不等它),任务由 emitter 持强引用、`close()` 统一取消。
5. **注入约定**:handler 参数按形参标注解析——`payloads.py` 里的 dataclass → 解析后的 payload 对象;`Event` → 事件封装;登记进 `emitter.services` 的组件类型(如 `BotApi`)→ 实例;未标注或未登记 → 回退传**原始 d**。解析是纯同步查表(`get_type_hints` + 签名缓存),无框架、无运行时魔法。签名与类型标注每个 handler 只解析一次并缓存,但 `services` 命中在每条事件时实时判定,保证"先注册 handler、后登记组件"的装配顺序可用。
+7 -4
View File
@@ -63,6 +63,9 @@ class WebsocketConnecter:
self._session = session
self._queue = queue
self._protocol = protocol
# 出站帧缓冲(心跳 + 协议应答):由 reply_helper 单任务发送,
# 维持 aiohttp ws 单写者不变量,生产者(心跳)也不会被发送阻塞
self._outbound: asyncio.Queue[Payload] = asyncio.Queue()
async def run(self) -> None:
backoff: float = 1
@@ -145,7 +148,7 @@ class WebsocketConnecter:
# 会话重置在下面 break 之前已同步完成,重连不会误 RESUME 死会话。
response = await self._protocol.on_frame(data)
if response is not None:
await self._queue.put_reply(cast(Payload, response))
await self._outbound.put(cast(Payload, response))
if op == Opcode.DISPATCH:
await self._queue.put_event(data)
@@ -159,12 +162,12 @@ class WebsocketConnecter:
interval = await interval_fut
while True:
await asyncio.sleep(interval)
# 服务端不会推送 op=1,心跳只由本任务经 reply 队列发出
await self._queue.put_reply(payload_of(Opcode.HEARTBEAT, self._session.seq))
# 服务端不会推送 op=1,心跳只由本任务投进出站缓冲
await self._outbound.put(payload_of(Opcode.HEARTBEAT, self._session.seq))
async def reply_helper(self, ws: aiohttp.ClientWebSocketResponse) -> None:
while not ws.closed:
payload: Payload = await self._queue.get_reply()
payload = await self._outbound.get()
body = {k: v for k, v in payload.items() if v != NOT_SET}
try:
await ws.send_json(body)
+6 -9
View File
@@ -1,4 +1,8 @@
"""事件/应答队列:连接层与分发层之间唯一的跨任务通道。"""
"""事件队列:连接层 → 分发层的单向跨任务通道。
只承载业务事件(op=0)。协议帧由连接适配器就地处理,出站帧(ws 心跳与
协议应答)由适配器自己缓冲,都不经过这里。
"""
from asyncio import Queue
@@ -6,20 +10,13 @@ from .model import Payload
class EventQueue:
"""连接层与分发层之间的事件/应答通道。"""
"""连接层投递、分发层消费的业务事件通道。"""
def __init__(self) -> None:
self.msg_queue: Queue[Payload] = Queue()
self.reply_queue: Queue[Payload] = Queue()
async def put_event(self, item: Payload) -> None:
await self.msg_queue.put(item)
async def get_event(self) -> Payload:
return await self.msg_queue.get()
async def put_reply(self, item: Payload) -> None:
await self.reply_queue.put(item)
async def get_reply(self) -> Payload:
return await self.reply_queue.get()
+7 -7
View File
@@ -4,7 +4,7 @@ import pytest
from qqbotsdk.emitter import EventEmitter
from qqbotsdk.events import Event
from qqbotsdk.model import Opcode, payload_of
from qqbotsdk.model import Opcode
from qqbotsdk.payloads import GroupAtMessage
from qqbotsdk.queue import EventQueue
from qqbotsdk.session import Session
@@ -40,16 +40,16 @@ async def test_handler_exception_is_isolated():
@pytest.mark.asyncio
async def test_protocol_frame_never_reaches_business_handler():
"""协议帧改由适配器的协议处理器就近处理,不再经 emitter 回流应答。"""
queue = EventQueue()
emitter = EventEmitter(queue)
"""协议帧不进分发路径:即使注册了同名 handler 也不该被调用。"""
emitter = EventEmitter()
called: list[object] = []
@emitter.on(Opcode.HEARTBEAT)
async def beat(_):
return payload_of(Opcode.HEARTBEAT, 42)
async def beat(d):
called.append(d)
await emitter.emit({"op": Opcode.HEARTBEAT, "d": 0})
assert queue.reply_queue.empty()
assert called == []
@pytest.mark.asyncio