From b0e0422c10f4974af35da14f0af81a2c8c22938d Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Tue, 14 Jul 2026 15:07:21 +0100 Subject: [PATCH] =?UTF-8?q?docs:=20=E6=89=B9=E9=87=8F=20idle=20flush=20?= =?UTF-8?q?=E8=90=BD=E5=9C=B0=E5=90=8E=E5=8F=A3=E5=BE=84=E5=90=8C=E6=AD=A5?= =?UTF-8?q?(README/CLAUDE.md)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 批量语义从"无 idle flush、低频停滞、重启丢批"改为"超时(缺省 60s)/停机 强制刷出,仅 kill -9 丢未满批";注解属性表补 batchTimeoutMs 行,TODO 勾选。 全景手册(docs/,gitignored)已同步修改,不在本 commit。 Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 3 ++- README.md | 13 +++++++------ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index b9fd5eb..ae70949 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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` 会让标记卡在 processing，reclaim 重投时被去重挡掉并 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 每次 reclaim（xClaim）使 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`，且**必须声明具体元素类型**（`List`，不能裸 `List`）—— driver 仍按 `T` 逐条反序列化，dispatcher 负责攒批。 - `batchSize` 与首参是否 `List` 的**错配只打 WARN、不阻断启动**（2026-07-14 决策，刻意软校验）：走不走批量始终只由首参决定——`batchSize>1` 配单 payload 按单条跑，`List` 首参配 `batchSize<=1` 按 1 条一批跑。 +- `batchTimeoutMs` 仅批量监听器生效；`batchSize<=1` 且显式配了非缺省值时同样只打 WARN 软校验、不阻断（恰好显式写 60000 检测不出，已知可接受）。 ## 启动期 fail-fast 校验（近期新增，改动时留意） diff --git a/README.md b/README.md index 00b8008..fe7e6ed 100644 --- a/README.md +++ b/README.md @@ -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 msgs) { } @@ -164,7 +164,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler { > ⚠️ **不配 `errorHandler` = 失败静默丢弃**。未指定 `errorHandler` 时，单条 / 批量 / 重试耗尽的任何消费失败最终都会被 ACK、不落库、不留痕迹（ACK 终止是内建行为，与是否落库无关）。关键消费务必配上 `errorHandler` 落库，否则失败即丢、事后无从排查。 > -> ⚠️ **批量有几个坑**：参数必须写带具体元素类型的 `List`（裸 `List` 启动扫描直接抛 `IllegalStateException`）；没有 idle flush，低流量攒不满会一直等，进程重启会丢内存里没攒满的那批；整批业务失败只回调 SINGLE 落库、不重投（所以批量务必配 `errorHandler`）。只适合高流量、能接受这些取舍的场景。批量方法同样可以加第二参 `MqContext`（拿到的是该批最后一条的元数据）。 +> ⚠️ **批量有几个坑**：参数必须写带具体元素类型的 `List`（裸 `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 撞车：发送方没填或写死了 key（redis-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 flush（2026-07-14 落地：batchTimeoutMs 超时强刷 + 停机强刷；kill -9 仍丢内存未满批） - [ ] 顺序消息 - [ ] 事务消息