From ffa30a7631f559762e45314a2f653d72ce6d01c9 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Wed, 15 Jul 2026 03:40:07 +0100 Subject: [PATCH] =?UTF-8?q?docs:=20review=20=E6=9C=AA=E4=BF=AE=E9=A1=B9?= =?UTF-8?q?=E8=AE=B0=E5=BD=95=E4=B8=BA=E5=B7=B2=E7=9F=A5=E8=AF=AD=E4=B9=89?= =?UTF-8?q?(README/CLAUDE.md)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 用户拍板:#5 批间不保序、#7 停机强刷无超时、#4 start 无回滚、#10 三处 快照样板重复,记录为已知语义/结构注意,不按缺陷修。 - README:批量坑段补两条业务方可感知的已知语义(批间不保序;停机会排队 等业务批量方法、挂死会卡住优雅停机)。 - CLAUDE.md:dispatcher 段新增"已知语义/结构注意"条目(四条全记,含 改快照样板须三处同步的提醒)。 全景手册(gitignored)已同步,不进本 commit。 Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 1 + README.md | 2 ++ 2 files changed, 3 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index 463deb8..84f9136 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -65,6 +65,7 @@ mq-spring-boot-starter 聚合 jar，业务方只引这一个（恒含 idemp - **单条消费失败**：先落库（**仅当配了 `errorHandler`**；`safeOnError` 对 null handler 静默跳过、不落库）+ `store.markFail()`（把去重标记从 processing 复位），再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processing，reclaim 重投时被去重挡掉并 ACK → 静默丢消息。 - **批量消费**：满批 / 超时（`batchTimeoutMs` 缺省 60s，从本批首条算起；registry 全局单线程 `mq-batch-flush` 每 1s 扫）/ 停机（stop 时 `forceFlush`）三路径刷出（forceFlush 同时打烊：停机后迟到的消息拒收抛 InProgress、不 ACK、下次启动重投，堵住 driver.stop 异步返回的缝隙；已过 `tryAcquire` 才撞上打烊的拒收会先 `markFail` 复位标记再抛，避免重投一直撞 stale processing 白烧重试计数），统一走 `doFlush`（`flushLock` 串行化，业务批量方法任一时刻至多一个调用）；kill -9 仍丢内存未刷批（上界 = 一个超时窗口）。批量失败因消息已逐条 ACK 无法精确重投整批，故落库（同样只在配了 `errorHandler` 时）+ 吞掉不抛。 - **停机顺序**（`MqListenerRegistry.stop()`，2026-07-15 review 修复后）：driver 停（**异常吞掉打 error，不得中断后续强刷**）→ 定时器 `shutdown()` + `awaitTermination(30s)` **等在途的超时刷批跑完**（超时快照已把批移出 buffer，forceFlush 看不见它，不等就是停机丢批）→ 逐监听器 `forceFlush` → `running=false`。改停机代码时别破坏这条链的任何一环。 +- **已知语义 / 结构注意**（2026-07-15 review 逐条拍板"记录不修"）：① **批间不保序**——超时批快照后、抢到 `flushLock` 前可能被新满批插队（MQ 重投本身不保序，框架不承诺批间有序，README 已写给业务方）；② **停机强刷无超时**——`forceFlush → doFlush` 排队 `flushLock` 无上界，挂死的业务方法会卡住 stop（与满批路径同风险，已接受，README 已写）；③ **start 中途失败无回滚**——先 `dispatchers.add` 后 `driver.subscribe`，中途抛异常再重试 start 会重复订阅先成功的 endpoint（既有行为，Spring 实际几乎不重试 start）；④ **"锁内快照→清空→锁外 doFlush"样板在三处重复**——dispatch 满批分支 / `flushIfTimeout` / `forceFlush`，仅 drain 条件不同，**改其中一处务必对照另外两处**，防静默分叉。 driver 侧对 `MqMessageInProgressException` 的对应处理：`RedisStreamMqDriver.java` 捕获后不 `acknowledge()`（留 PEL 待 reclaim）；`RocketMqDriver.java` 在通用 `catch(Exception)` **之前**单独捕获并返回 `RECONSUME_LATER`（绝不进耗尽落库分支——catch 顺序不能动，它是 `Exception` 子类）。边缘（**两端共性**，2026-07-14 复核）：InProgress 白跑会烧重试计数——rocketmq 每次 `RECONSUME_LATER` 使 `reconsumeTimes` +1，烧光进原生 `%DLQ%`（进死信、非丢消息）；redis-stream 每次 reclaim（xClaim）使 PEL delivery +1，`deliveryCount > maxRetry` 时**不再 dispatch** 直接判耗尽（配了 `errorHandler` 落库，没配则静默丢弃——业务可能一次都没重跑）。缓解同构：`mq.idempotent.processing-ttl` 小于 maxRetry 轮重投累计时长（rocketmq 默认 ~100s / redis-stream 为 `maxRetry × reclaim-min-idle` 默认 90s）。 diff --git a/README.md b/README.md index e608362..e4faac2 100644 --- a/README.md +++ b/README.md @@ -166,6 +166,8 @@ public class MyErrorHandler implements MqConsumeErrorHandler { > > ⚠️ **批量有几个坑**：参数必须写带具体元素类型的 `List`（裸 `List` 启动扫描直接抛 `IllegalStateException`）；攒不满时最多等 `batchTimeoutMs`（缺省 60 秒，从本批第一条算起）会强制刷出，正常停机也会把没攒满的批刷出去（若停机瞬间恰有一批正在业务方法里跑，停机会等它跑完，上限 30 秒），但 kill -9 / 断电仍会丢内存里那批（最多一个超时窗口内的量）；整批业务失败只回调 SINGLE 落库、不重投（所以批量务必配 `errorHandler`）。批量方法同样可以加第二参 `MqContext`（拿到的是该批最后一条的元数据）。 > +> 两条**已知语义**（设计如此，不按缺陷处理）：**批与批之间不保证顺序**——超时刷出的批可能被恰好凑满的新批插队，业务先收到新批再收到旧批（MQ 重投本身也不保序，消费逻辑别依赖批间先后）；**停机会排队等正在跑的业务批量方法**——等定时刷批那步有 30 秒上限，但停机强刷没有，业务方法若挂死（如数据库连接僵住），优雅停机会被一直卡住。 +> > 另外 `batchSize` 与首参是否 `List` **错配时不报错**：启动时打一条 WARN 日志、仍按首参决定走不走批量（`batchSize=50` 配单个 payload 退化为逐条；`List` 首参配 `batchSize` 缺省按 1 条一批）。见启动日志 `[mq] @MqListener batchSize=...`。 ## 两个 driver 不太一样