docs(CLAUDE.md): 2026-07-14 复核对齐——batchSize 软校验决策 + 三处口径补正
- 方法签名约束:记录 batchSize 与首参 List 错配只打 WARN 不阻断(刻意软校验,走不走批量只由首参决定) - 失败落库口径:单条/批量落库均仅在配了 errorHandler 时发生(safeOnError 对 null handler 静默跳过);MqConsumeErrorHandler 标注可选、未配则任何失败仅 ACK 丢弃 - InProgress 白跑烧重试计数改记为两端共性(rocketmq reconsumeTimes / redis-stream PEL delivery),缓解同构:processing-ttl < maxRetry 轮重投累计时长 - fail-fast 校验补充:仅显式配 mq.type 时介入,缺省落 rocketmq 时三层校验不触发(已知遗留) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
11
CLAUDE.md
11
CLAUDE.md
@@ -62,17 +62,17 @@ mq-spring-boot-starter 聚合 jar,业务方只引这一个(恒含 idemp
|
||||
`mq-starter-core/.../container/MqConsumeDispatcher.java` 是消费链的心脏,纯逻辑无外部依赖。有几处**不能想当然改动**的正确性约束(都写在注释里):
|
||||
|
||||
- **去重的两类跳过**决定 driver 能否 ACK:已 `success` → 正常返回让 driver ACK 挡掉重复;仍 `processing`(并发中或消费者崩溃残留标记)→ 抛 `MqMessageInProgressException` 让 driver **不要 ACK**、保留消息待 `processing` 标记过期后重投。搞错就是**静默丢消息**。
|
||||
- **单条消费失败**:先落库 + `store.markFail()`(把去重标记从 processing 复位),再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processing,reclaim 重投时被去重挡掉并 ACK → 静默丢消息。
|
||||
- **批量消费**:无 idle flush,攒够 `batchSize` 才 flush;低流量下未满批会停滞,进程重启丢内存里没入库的那批(与 legacy 行为一致,适合高流量流)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库 + 吞掉不抛。
|
||||
- **单条消费失败**:先落库(**仅当配了 `errorHandler`**;`safeOnError` 对 null handler 静默跳过、不落库)+ `store.markFail()`(把去重标记从 processing 复位),再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processing,reclaim 重投时被去重挡掉并 ACK → 静默丢消息。
|
||||
- **批量消费**:无 idle flush,攒够 `batchSize` 才 flush;低流量下未满批会停滞,进程重启丢内存里没入库的那批(与 legacy 行为一致,适合高流量流)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库(同样只在配了 `errorHandler` 时)+ 吞掉不抛。
|
||||
|
||||
driver 侧对 `MqMessageInProgressException` 的对应处理:`RedisStreamMqDriver.java` 捕获后不 `acknowledge()`(留 PEL 待 reclaim);`RocketMqDriver.java` 在通用 `catch(Exception)` **之前**单独捕获并返回 `RECONSUME_LATER`(绝不进耗尽落库分支——catch 顺序不能动,它是 `Exception` 子类)。边缘:rocketmq 下 InProgress 反复重投超过 `maxReconsumeTimes` 会进原生 `%DLQ%`(进死信、非丢消息);缓解靠 `mq.idempotent.processing-ttl` < 业务重试节奏累计时长。
|
||||
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)。
|
||||
|
||||
## SPI 契约(`com.njcn.mq.spi`)
|
||||
|
||||
driver 只需实现 `MqDriver`(`send` / `subscribe` / `stop`)。其余是可选扩展点,由业务方或 driver 提供 bean:
|
||||
|
||||
- **`MqIdempotentStore`** —— 去重状态机 `processing / success / fail`。注意 `isCompleted(key)` 的默认实现返回 `true`(向后兼容旧行为),**有真状态机的实现必须覆写**,否则 processing 中的消息会被误判为已完成而被 ACK 丢掉。默认实现 `RedisMqIdempotentStore` 在 `mq-starter-idempotent-redis` 模块(包 `com.njcn.mq.idempotent.redis`),配置前缀 `mq.idempotent.*`(TTL/前缀),旧键 `mq.redis-stream.idempotent.*` 经 Binder 逐键回退向后兼容(用到会打一条聚合弃用 WARN)。用户自带 store bean 会让自动装配让位(`@ConditionalOnMissingBean`)。
|
||||
- **`MqConsumeErrorHandler`** —— 失败落库回调,通过 `@MqListener(errorHandler="beanName")` 按 bean 名解析。`MqErrorIdentity` 区分 `SINGLE`(单次失败)和 `RETRY_EXHAUSTED`(重试耗尽,由 driver 在超过 `maxRetry` 时回调 `MqSubscription.retryExhaustedHandler`)。
|
||||
- **`MqConsumeErrorHandler`** —— 失败落库回调(**可选**,无默认实现),通过 `@MqListener(errorHandler="beanName")` 按 bean 名解析——**未配 `errorHandler` 则任何失败(SINGLE / 批量 / RETRY_EXHAUSTED)都不落库、仅 ACK 丢弃**;ACK 终止本身内建、无条件,与是否落库无关。`MqErrorIdentity` 区分 `SINGLE`(单次失败)和 `RETRY_EXHAUSTED`(重试耗尽,由 driver 在超过 `maxRetry` 时回调 `MqSubscription.retryExhaustedHandler`)。
|
||||
- **`MqSubscription`** —— core 传给 driver 的订阅描述(topic/group/tag/payloadType/concurrency/batchSize/maxRetry/listener/retryExhaustedHandler)。
|
||||
|
||||
## `@MqListener` 方法签名约束
|
||||
@@ -81,12 +81,13 @@ driver 只需实现 `MqDriver`(`send` / `subscribe` / `stop`)。其余是可
|
||||
|
||||
- 参数只能是 `(payload)` 或 `(payload, MqContext)`,第二参必须是 `MqContext`。
|
||||
- `batchSize > 1` 时首参要用 `List<T>`,且**必须声明具体元素类型**(`List<FrontData>`,不能裸 `List`)—— driver 仍按 `T` 逐条反序列化,dispatcher 负责攒批。
|
||||
- `batchSize` 与首参是否 `List` 的**错配只打 WARN、不阻断启动**(2026-07-14 决策,刻意软校验):走不走批量始终只由首参决定——`batchSize>1` 配单 payload 按单条跑,`List` 首参配 `batchSize<=1` 按 1 条一批跑。
|
||||
|
||||
## 启动期 fail-fast 校验(近期新增,改动时留意)
|
||||
|
||||
三处独立校验,都受统一逃生开关 `mq.startup-check.enabled`(默认 `true`,仅显式 `false` 关闭;见 `MqStartupChecks`)控制:
|
||||
|
||||
1. `MqDriverPresenceAutoConfiguration`(core):配了 `mq.type` 却没装出任何 `MqDriver`(忘引 starter 或 `mq.type` 拼错)→ 启动失败并给中文提示。放在 core 是因为只有 core 有"到底有没有 driver"的全局视角。
|
||||
1. `MqDriverPresenceAutoConfiguration`(core):配了 `mq.type` 却没装出任何 `MqDriver`(忘引 starter 或 `mq.type` 拼错)→ 启动失败并给中文提示。放在 core 是因为只有 core 有"到底有没有 driver"的全局视角。**注意本校验只在显式配了 `mq.type` 时介入(`@ConditionalOnProperty(name="mq.type")`);缺省不配 `mq.type` 时靠 rocketmq 的 `matchIfMissing=true` 默认落 rocketmq,这三层校验均不触发(手册 Roadmap 13 列为已知遗留、行为不直观)。**
|
||||
2. `RedisStreamStartupProbeAutoConfiguration`:`mq.type=redis-stream` 时主动 `PING` 一次 Redis(lettuce 默认懒连接,不主动探测会拖到运行时才暴露连不上)。
|
||||
3. `RocketMqStartupProbeAutoConfiguration`:`mq.type=rocketmq` 时对 NameServer 做 TCP 短超时探测(任一地址可达即通过)。
|
||||
|
||||
|
||||
Reference in New Issue
Block a user