docs: 批量 idle flush 落地后口径同步(README/CLAUDE.md)

批量语义从"无 idle flush、低频停滞、重启丢批"改为"超时(缺省 60s)/停机
强制刷出,仅 kill -9 丢未满批";注解属性表补 batchTimeoutMs 行,TODO 勾选。
全景手册(docs/,gitignored)已同步修改,不在本 commit。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-14 15:07:21 +01:00
parent 244d4d3854
commit b0e0422c10
2 changed files with 9 additions and 7 deletions

View File

@@ -63,7 +63,7 @@ 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` 会让标记卡在 processingreclaim 重投时被去重挡掉并 ACK → 静默丢消息。
- **批量消费**无 idle flush攒够 `batchSize` 才 flush低流量下未满批会停滞进程重启丢内存里没入库的那批与 legacy 行为一致,适合高流量流)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库(同样只在配了 `errorHandler` 时)+ 吞掉不抛。
- **批量消费**满批 / 超时(`batchTimeoutMs` 缺省 60s从本批首条算起registry 全局单线程 `mq-batch-flush` 每 1s 扫)/ 停机stop 时 `forceFlush`)三路径刷出,统一走 `doFlush``flushLock` 串行化业务批量方法任一时刻至多一个调用kill -9 仍丢内存未刷批(上界 = 一个超时窗口)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库(同样只在配了 `errorHandler` 时)+ 吞掉不抛。
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 每次 reclaimxClaim使 PEL delivery +1`deliveryCount > maxRetry` 时**不再 dispatch** 直接判耗尽(配了 `errorHandler` 落库,没配则静默丢弃——业务可能一次都没重跑)。缓解同构:`mq.idempotent.processing-ttl` 小于 maxRetry 轮重投累计时长rocketmq 默认 ~100s / redis-stream 为 `maxRetry × reclaim-min-idle` 默认 90s
@@ -82,6 +82,7 @@ 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 条一批跑。
- `batchTimeoutMs` 仅批量监听器生效;`batchSize<=1` 且显式配了非缺省值时同样只打 WARN 软校验、不阻断(恰好显式写 60000 检测不出,已知可接受)。
## 启动期 fail-fast 校验(近期新增,改动时留意)

View File

@@ -140,7 +140,7 @@ public void onFile(FileData msg) { } // 只收 tag=node-01 的消息,
@MqListener(topic = "LN_Topic", group = "g", idempotent = false)
public void onData(FrontData msg) { }
// 批量:攒够 batchSize 条一次给你,参数要用 List
// 批量:攒够 batchSize 条或超时 batchTimeoutMs缺省 60s一次给你,参数要用 List
@MqListener(topic = "LN_Topic", group = "g", batchSize = 50)
public void onBatch(List<FrontData> msgs) { }
@@ -164,7 +164,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
> ⚠️ **不配 `errorHandler` = 失败静默丢弃**。未指定 `errorHandler` 时,单条 / 批量 / 重试耗尽的任何消费失败最终都会被 ACK、不落库、不留痕迹ACK 终止是内建行为,与是否落库无关)。关键消费务必配上 `errorHandler` 落库,否则失败即丢、事后无从排查。
>
> ⚠️ **批量有几个坑**:参数必须写带具体元素类型的 `List<FrontData>`(裸 `List` 启动扫描直接抛 `IllegalStateException`没有 idle flush低流量攒不满会一直等进程重启会丢内存里没攒满的那批;整批业务失败只回调 SINGLE 落库、不重投(所以批量务必配 `errorHandler`)。只适合高流量、能接受这些取舍的场景。批量方法同样可以加第二参 `MqContext`(拿到的是该批最后一条的元数据)。
> ⚠️ **批量有几个坑**:参数必须写带具体元素类型的 `List<FrontData>`(裸 `List` 启动扫描直接抛 `IllegalStateException`攒不满时最多等 `batchTimeoutMs`(缺省 60 秒,从本批第一条算起)会强制刷出,正常停机也会把没攒满的批刷出去,但 kill -9 / 断电仍会丢内存里那批(最多一个超时窗口内的量);整批业务失败只回调 SINGLE 落库、不重投(所以批量务必配 `errorHandler`)。批量方法同样可以加第二参 `MqContext`(拿到的是该批最后一条的元数据)。
>
> 另外 `batchSize` 与首参是否 `List` **错配时不报错**:启动时打一条 WARN 日志、仍按首参决定走不走批量(`batchSize=50` 配单个 payload 退化为逐条;`List` 首参配 `batchSize` 缺省按 1 条一批)。见启动日志 `[mq] @MqListener batchSize=...`。
@@ -179,7 +179,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
| 去重 `idempotent`(默认开启) | ✅(需 Redis | ✅(需 Redis |
| 失败落库 `RETRY_EXHAUSTED`(需配 `errorHandler`,否则仅 ACK 丢弃) | ✅ | ✅ 落库+ACK |
| 失败落库 `SINGLE` | ⚠️ 单条:落库+重投;批量:失败被吞(两端一致) | ⚠️ 单条:落库+重投;批量:失败被吞(两端一致) |
| 批量 `batchSize` | ⚠️ core 内存攒批driver 逐条喂);无 idle flush低频停滞/重启丢未满批(两端一致) | ⚠️ core 内存攒批driver 逐条喂);无 idle flush低频停滞/重启丢未满批(两端一致) |
| 批量 `batchSize` | ⚠️ core 内存攒批driver 逐条喂);超时/停机强制刷出,仅 kill -9 丢未满批(两端一致) | ⚠️ core 内存攒批driver 逐条喂);超时/停机强制刷出,仅 kill -9 丢未满批(两端一致) |
| 重试 | ✅ PEL reclaim | ✅ 原生重投 |
| 死信队列 | ❌ 只落库 | ⚠️ 原生 `%DLQ%` 通常被耗尽 ACK 旁路InProgress 白跑烧光计数时仍会进) |
| 顺序消息 | ❌ | ❌ 用的 Concurrently |
@@ -262,6 +262,7 @@ core 是中立的,编排逻辑不依赖 Spring 和 Redis可以直接单测
| `group` | 必填 | 消费组名。同一组的多个实例**分摊**消费(一条消息只被组内一个实例消费),不同组**各收一份**全量。消费进度跟着组走:换个组名就是新组,两端都**从最新消息开始收、不回放历史**rocketmq 新组默认最新位点redis-stream 建组用 `$` |
| `concurrency` | 1 | 并发消费线程数,**当前两 driver 均未按此字段兑现**redis 单线程 poll、rocketmq 用原生线程池默认值) |
| `batchSize` | 1 | 攒批大小,大于 1 要用 `List` 参数;与首参是否 `List` 错配时启动打 WARN、仍按首参跑 |
| `batchTimeoutMs` | 60000 | 批量攒批最长等待毫秒数从本批第一条算起到点未满也强制刷出0/负数关闭;仅批量监听器生效 |
| `maxRetry` | 3 | 最大重试次数redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker `maxReconsumeTimes`;被去重标记挡回的 InProgress 白跑**两端都计入** |
| `tag` | `*` | 只收打了指定 tag 的消息(发送端三参 `send` 的第二参打 tag。rocketmq 为 broker 服务端过滤、支持原生表达式(如 `A \|\| B`redis-stream 当前不过滤、全收。详见「消费」一节的 tag 小节 |
| `idempotent` | true | 是否去重(默认开启,需 Redis 可达;单个监听器关闭写 `idempotent = false`,全局关闭配 `mq.idempotent.enabled=false` |
@@ -306,8 +307,8 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置(`nam
**对接异构发送方后,只有第一条消息能消费、后面成片丢失?**
几乎必是去重 key 撞车发送方没填或写死了 keyredis-stream 信封 `key` field / rocketmq 报文头 KEYS 属性所有消息共用一把去重钥匙第一条成功后其余全被判重复静默 ACK让发送方保证 key 逐条唯一Redis Stream 消息结构一节兑现前先在对应监听器上显式 `idempotent = false`
**批量消费收不到 / 停滞 / 丢消息?**
没有 idle flush攒不满 `batchSize` 不触发整批失败会被吞掉不重投低流量把 `batchSize` 调小或改单条消费
**批量消费收不到 / 延迟 / 丢消息?**
攒不满 `batchSize` 时最多等 `batchTimeoutMs`缺省 60 才刷出嫌慢就把它调小整批失败会被吞掉不重投务必配 `errorHandler`kill -9 / 断电会丢内存里没攒满的批正常停机会强制刷出不丢
**消费失败后消息不见了、事后查不到?**
多半是没配 `errorHandler`没配时单条 / 批量 / 重试耗尽的失败一律只 ACK 丢弃不留痕迹关键消费用 `@MqListener(errorHandler = "beanName")` 落库进阶用法」。
@@ -326,7 +327,7 @@ P0放量到生产前想补的
- [x] 失败落库去重批量消费core 已做
- [x] rocketmq driver 补上去重和 `RETRY_EXHAUSTED` 落库 redis-stream 对齐去重 store 已抽成 driver 中立的 `mq-starter-idempotent-redis`
- [ ] 批量消费加 idle flush现在低流量下攒不满会卡着进程重启还会丢内存里没攒满的那批
- [x] 批量消费加 idle flush2026-07-14 落地batchTimeoutMs 超时强刷 + 停机强刷kill -9 仍丢内存未满批
- [ ] 顺序消息
- [ ] 事务消息