From 4d19db24e1a138397e6e21094c184bce18c9fbe8 Mon Sep 17 00:00:00 2001 From: snowy-yang <102860110+snowy-yang@users.noreply.github.com> Date: Sun, 20 Sep 2026 22:02:07 +0800 Subject: [PATCH] =?UTF-8?q?refactor!:=20EventQueue=20=E6=94=B6=E6=95=9B?= =?UTF-8?q?=E4=B8=BA=E5=8D=95=E5=90=91=E4=B8=9A=E5=8A=A1=E4=BA=8B=E4=BB=B6?= =?UTF-8?q?=E9=80=9A=E9=81=93=EF=BC=8C=E5=87=BA=E7=AB=99=E7=BC=93=E5=86=B2?= =?UTF-8?q?=E5=BD=92=20ws=20=E9=80=82=E9=85=8D=E5=99=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 协议分流后 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 干净。 --- CHANGELOG.md | 1 + docs/API.md | 2 +- docs/ARCHITECTURE.md | 4 ++-- src/qqbotsdk/connecter.py | 11 +++++++---- src/qqbotsdk/queue.py | 15 ++++++--------- tests/test_emitter.py | 14 +++++++------- 6 files changed, 24 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 67eee3c..5d2f372 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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` 不再存在 ### 移除 diff --git a/docs/API.md b/docs/API.md index ed6f13d..32532d6 100644 --- a/docs/API.md +++ b/docs/API.md @@ -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)` | diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 7376a5c..0d8c191 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -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、后登记组件"的装配顺序可用。 diff --git a/src/qqbotsdk/connecter.py b/src/qqbotsdk/connecter.py index 9c2ec30..f238575 100644 --- a/src/qqbotsdk/connecter.py +++ b/src/qqbotsdk/connecter.py @@ -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) diff --git a/src/qqbotsdk/queue.py b/src/qqbotsdk/queue.py index f47dc01..83794d6 100644 --- a/src/qqbotsdk/queue.py +++ b/src/qqbotsdk/queue.py @@ -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() diff --git a/tests/test_emitter.py b/tests/test_emitter.py index 102edc0..ca879c6 100644 --- a/tests/test_emitter.py +++ b/tests/test_emitter.py @@ -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