docs: review 修复后停机口径同步(README/CLAUDE.md)
- CLAUDE.md:dispatcher 段补打烊拒收 markFail 复位一笔;新增"停机顺序" 约束条目(driver 停(异常吞掉)→ 定时器 shutdown+awaitTermination(30s) 等在途刷批 → 逐监听器 forceFlush → running=false)。 - README:批量坑位说明补"停机瞬间恰有一批在业务方法里跑时,停机等它 跑完,上限 30 秒"。 全景手册(gitignored)已同步更新,不进本 commit。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -63,7 +63,8 @@ mq-spring-boot-starter 聚合 jar,业务方只引这一个(恒含 idemp
|
||||
|
||||
- **去重的两类跳过**决定 driver 能否 ACK:已 `success` → 正常返回让 driver ACK 挡掉重复;仍 `processing`(并发中或消费者崩溃残留标记)→ 抛 `MqMessageInProgressException` 让 driver **不要 ACK**、保留消息待 `processing` 标记过期后重投。搞错就是**静默丢消息**。
|
||||
- **单条消费失败**:先落库(**仅当配了 `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 异步返回的缝隙),统一走 `doFlush`(`flushLock` 串行化,业务批量方法任一时刻至多一个调用);kill -9 仍丢内存未刷批(上界 = 一个超时窗口)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库(同样只在配了 `errorHandler` 时)+ 吞掉不抛。
|
||||
- **批量消费**:满批 / 超时(`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`。改停机代码时别破坏这条链的任何一环。
|
||||
|
||||
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)。
|
||||
|
||||
|
||||
@@ -164,7 +164,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
|
||||
|
||||
> ⚠️ **不配 `errorHandler` = 失败静默丢弃**。未指定 `errorHandler` 时,单条 / 批量 / 重试耗尽的任何消费失败最终都会被 ACK、不落库、不留痕迹(ACK 终止是内建行为,与是否落库无关)。关键消费务必配上 `errorHandler` 落库,否则失败即丢、事后无从排查。
|
||||
>
|
||||
> ⚠️ **批量有几个坑**:参数必须写带具体元素类型的 `List<FrontData>`(裸 `List` 启动扫描直接抛 `IllegalStateException`);攒不满时最多等 `batchTimeoutMs`(缺省 60 秒,从本批第一条算起)会强制刷出,正常停机也会把没攒满的批刷出去,但 kill -9 / 断电仍会丢内存里那批(最多一个超时窗口内的量);整批业务失败只回调 SINGLE 落库、不重投(所以批量务必配 `errorHandler`)。批量方法同样可以加第二参 `MqContext`(拿到的是该批最后一条的元数据)。
|
||||
> ⚠️ **批量有几个坑**:参数必须写带具体元素类型的 `List<FrontData>`(裸 `List` 启动扫描直接抛 `IllegalStateException`);攒不满时最多等 `batchTimeoutMs`(缺省 60 秒,从本批第一条算起)会强制刷出,正常停机也会把没攒满的批刷出去(若停机瞬间恰有一批正在业务方法里跑,停机会等它跑完,上限 30 秒),但 kill -9 / 断电仍会丢内存里那批(最多一个超时窗口内的量);整批业务失败只回调 SINGLE 落库、不重投(所以批量务必配 `errorHandler`)。批量方法同样可以加第二参 `MqContext`(拿到的是该批最后一条的元数据)。
|
||||
>
|
||||
> 另外 `batchSize` 与首参是否 `List` **错配时不报错**:启动时打一条 WARN 日志、仍按首参决定走不走批量(`batchSize=50` 配单个 payload 退化为逐条;`List` 首参配 `batchSize` 缺省按 1 条一批)。见启动日志 `[mq] @MqListener batchSize=...`。
|
||||
|
||||
|
||||
Reference in New Issue
Block a user