docs: review 未修项记录为已知语义(README/CLAUDE.md)

用户拍板:#5 批间不保序、#7 停机强刷无超时、#4 start 无回滚、#10 三处
快照样板重复,记录为已知语义/结构注意,不按缺陷修。

- README:批量坑段补两条业务方可感知的已知语义(批间不保序;停机会排队
  等业务批量方法、挂死会卡住优雅停机)。
- CLAUDE.md:dispatcher 段新增"已知语义/结构注意"条目(四条全记,含
  改快照样板须三处同步的提醒)。

全景手册(gitignored)已同步,不进本 commit。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

CLAUDE.md

@@ -46,7 +46,7 @@ mq-spring-boot-starter      聚合 jar，业务方只引这一个（恒含 idemp
 各模块 `META-INF/spring.factories` 注册 AutoConfiguration。装配开关是**唯一的** `mq.type`：
 
 1. **driver 二选一**：`RedisStreamMqDriverAutoConfiguration` / `RocketMqDriverAutoConfiguration` 都用 `@ConditionalOnProperty(name="mq.type", havingValue=...)` 按 `mq.type` 条件装配，各自产出一个 `MqDriver` bean。没配 `mq.type` 则两个都不装，`MqTemplate`/`@MqListener` 全部不生效。
-2. **去重 store 不按 mq.type 门控**：`MqIdempotentRedisAutoConfiguration`（idempotent-redis 模块）只按 Redis 在场门控（`@ConditionalOnClass/OnBean(StringRedisTemplate)` + `@AutoConfigureAfter(RedisAutoConfiguration)`），这是 rocketmq 部署也能用上去重的关键。注意 classpath 有 lettuce 时 Spring Boot 会用 localhost 默认值自动造 `StringRedisTemplate`，所以该条件几乎恒真——真正的启动安全网是**懒连接 + core 仅在 `idempotent=true` 时才调用 store**（该装配与 store 构造都刻意不 PING Redis，与下述 fail-fast 探测相反）。逃生开关 `mq.idempotent.enabled=false`。
+2. **去重 store 不按 mq.type 门控**：`MqIdempotentRedisAutoConfiguration`（idempotent-redis 模块）只按 Redis 在场门控（`@ConditionalOnClass/OnBean(StringRedisTemplate)` + `@AutoConfigureAfter(RedisAutoConfiguration)`），这是 rocketmq 部署也能用上去重的关键。注意 classpath 有 lettuce 时 Spring Boot 会用 localhost 默认值自动造 `StringRedisTemplate`，所以该条件几乎恒真——真正的启动安全网只剩**懒连接**（该装配与 store 构造都刻意不 PING Redis，与下述 fail-fast 探测相反）——`@MqListener` 的 `idempotent` 自 2026-07-14 起**默认 `true`**，消费链默认就会调用 store，Redis 不可达在消费期（而非启动期）以连接异常暴露。逃生开关 `mq.idempotent.enabled=false`（关掉后去重不生效、启动仅打 WARN）。
 3. **core 依赖 driver**：`MqCoreAutoConfiguration` 里 `MqTemplate` 和 `MqListenerRegistry` 都是 `@ConditionalOnBean(MqDriver.class)` —— 有 driver 才有门面。driver 与 idempotent-redis 的 AutoConfiguration 都用 `@AutoConfigureBefore(MqCoreAutoConfiguration.class)` 保证自己先装。
 4. **注解扫描**：`MqListenerAnnotationBeanPostProcessor`（一个 `BeanPostProcessor`）扫描所有 bean 的方法，把 `@MqListener` 方法收集成 `MqListenerEndpoint` 列表。
 5. **启动订阅**：`MqListenerRegistry` 实现 `SmartLifecycle`（`phase = Integer.MAX_VALUE - 100`，最后启动），在 `start()` 里为每个 endpoint 组装 `MqConsumeDispatcher`，包成 `MqSubscription` 交给 `driver.subscribe()`。
@@ -63,7 +63,9 @@ 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`）三路径刷出（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）。
 
@@ -82,6 +84,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 校验（近期新增，改动时留意）
 

README.md

@@ -67,6 +67,10 @@ rocketmq:
   name-server: localhost:9876
   producer:
     group: my-producer-group
+spring:
+  redis:             # 去重默认开启、状态存 Redis，所以 rocketmq 部署也要配 Redis；
+    host: localhost  # 确实不用去重时可去掉，并配 mq.idempotent.enabled=false 全局关闭
+    port: 6379
 ```
 
 `mq.type` 是装配开关。注意 rocketmq driver 带 `matchIfMissing=true`：**没配 `mq.type` 会默认装配 rocketmq**（聚合坐标下 classpath 恒有 rocketmq），此时若 `rocketmq.*` 没配好通常直接启动报错；想用 redis-stream **必须显式**写 `mq.type: redis-stream`；配了两端都不认的错误值（拼错）**默认直接 fail-fast**——启动抛 `IllegalStateException` 并附中文排查提示，不是静默不装。仅当显式 `mq.startup-check.enabled=false` 关掉启动校验时，才退化为两个 driver 都不装、`MqTemplate` 和 `@MqListener` 都不生效。
@@ -111,16 +115,32 @@ public class FrontDataConsumer {
 
 方法签名只认这两种：`(payload)` 或 `(payload, MqContext)`。
 
+### tag：同一个 topic 下再分小类
+
+topic 是大频道，tag 是频道内给消息贴的标签——多个下游订同一个 topic、各自只认自己的标签，就不用为每个小类单独开 topic。发送方打标签、消费方声明要哪类，就这两步：
+
+```java
+mq.send("File_Topic", "node-01", fileData);   // 三参 send 的第二参就是 tag
+
+@MqListener(topic = "File_Topic", group = "file_consumer", tag = "node-01")
+public void onFile(FileData msg) { }          // 只收 tag=node-01 的消息，其余不会进这个方法
+```
+
+- 消费端不写 `tag` 默认 `*`，整个 topic 全收；发送端用两参 `send` 则消息不带 tag——**对端若按 tag 过滤，不带标签的消息它收不到**，跨系统对接时这是个经典坑。
+- rocketmq 模式：过滤发生在 **broker 服务端**，不匹配的消息根本不会投到消费者；tag 表达式直接透传 RocketMQ 原生语法，`"A || B"` 多选一也支持。
+- redis-stream 模式：**当前不过滤、全收**（见能力表），tag 只是随消息带到消费端，可从 `MqContext.getTag()` 读到，需要分流的话业务自己判断——这也是切换 `mq.type` 时的行为翻转点之一。
+
 ## 进阶用法
 
 去重、批量、失败落库都是 `@MqListener` 上的属性，业务方法本身不用改：
 
 ```java
-// 去重：重复消息不会再进业务（去重状态存 Redis，两个 driver 都支持，见下方说明）
-@MqListener(topic = "LN_Topic", group = "g", idempotent = true)
+// 去重默认开启：重复消息不会再进业务（去重状态存 Redis，两个 driver 都支持，见下方说明）
+// 确实不需要去重的监听器显式关掉：
+@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) { }
 
@@ -144,7 +164,9 @@ 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 秒，从本批第一条算起）会强制刷出，正常停机也会把没攒满的批刷出去（若停机瞬间恰有一批正在业务方法里跑，停机会等它跑完，上限 30 秒），但 kill -9 / 断电仍会丢内存里那批（最多一个超时窗口内的量）；整批业务失败只回调 SINGLE 落库、不重投（所以批量务必配 `errorHandler`）。批量方法同样可以加第二参 `MqContext`（拿到的是该批最后一条的元数据）。
+>
+> 两条**已知语义**（设计如此，不按缺陷处理）：**批与批之间不保证顺序**——超时刷出的批可能被恰好凑满的新批插队，业务先收到新批再收到旧批（MQ 重投本身也不保序，消费逻辑别依赖批间先后）；**停机会排队等正在跑的业务批量方法**——等定时刷批那步有 30 秒上限，但停机强刷没有，业务方法若挂死（如数据库连接僵住），优雅停机会被一直卡住。
 >
 > 另外 `batchSize` 与首参是否 `List` **错配时不报错**：启动时打一条 WARN 日志、仍按首参决定走不走批量（`batchSize=50` 配单个 payload 退化为逐条；`List` 首参配 `batchSize` 缺省按 1 条一批）。见启动日志 `[mq] @MqListener batchSize=...`。
 
@@ -156,10 +178,10 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
 |---|:---:|:---:|
 | 收发 | ✅ | ✅ |
 | 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 |
-| 去重 `idempotent` | ✅（需 Redis） | ✅（需 Redis） |
+| 去重 `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 |
@@ -171,7 +193,37 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
 > 2. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。
 > 3. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。
 
-去重的状态存在 Redis（`processing / success / fail` 三态 + TTL），由独立模块 `mq-starter-idempotent-redis` 提供，**不看 `mq.type`**——所以 RocketMQ 部署只要能连上 Redis（标准 `spring.redis.*` 配置），`idempotent = true` 一样生效。引聚合坐标的话这个模块已经打在里面；单独只引 `mq-starter-rocketmq` 的话需要自己再加一个 `mq-starter-idempotent-redis` 依赖。没开去重时它完全不碰 Redis（懒连接），不会因为 Redis 没配好而影响启动。
+去重按**消息 key** 判重：经 `MqTemplate` 发送时自动生成雪花 ID、天然唯一，业务无感；**不经 `MqTemplate` 的异构发送方必须自己保证 key 逐条唯一**（redis-stream = 信封 `key` field，rocketmq = 报文头 KEYS 属性；为空或固定值会把不同消息误判成重复而静默丢弃，详见「Redis Stream 消息结构」一节及其末尾的 RocketMQ 补注）。去重的状态存在 Redis（`processing / success / fail` 三态 + TTL），由独立模块 `mq-starter-idempotent-redis` 提供，**不看 `mq.type`**——所以 RocketMQ 部署只要能连上 Redis（标准 `spring.redis.*` 配置），去重一样生效。引聚合坐标的话这个模块已经打在里面；单独只引 `mq-starter-rocketmq` 的话需要自己再加一个 `mq-starter-idempotent-redis` 依赖。**去重默认开启**，消费链路会读写 Redis，务必保证 Redis 可达——连不上时消费抛连接异常走重投（日志可见、不是静默丢，但业务会一直消费不进去）。确实不需要去重的监听器显式写 `idempotent = false`；或配 `mq.idempotent.enabled=false` 全局关掉 store（关掉后去重不生效、启动打 WARN 提示，完全不碰 Redis、不影响启动）。
+
+## Redis Stream 消息结构（异构发送方对接）
+
+redis-stream 模式下，消息在 Redis 里**不是裸 JSON**，而是固定五字段的信封（XADD 的 field），契约与内部 `redis-stream-springboot-starter` 一致（该格式当初就是和 C++ 端对死的）。不经 mq-starter 的发送方（C++ / 脚本 / 其他语言）直接 XADD 进对应 stream，`@MqListener` 一样能收，只要按这个结构写：
+
+| field | 内容 |
+|---|---|
+| `key` | 消息业务 key，**每条必须唯一**——消费端去重按它判重（`idempotent` 默认开启），key 重复或恒定会把不同消息误判成重复而丢弃。经 `MqTemplate` 发送时自动生成雪花 ID，无需关心 |
+| `tag` | 路由 / 过滤标记，可为空串（注意 redis-stream 消费端当前不按 tag 过滤，全收） |
+| `enc` | **body 的编码方式，压缩标识就放这里**：`plain`（明文 JSON）或 `gzip`（压缩，见下） |
+| `body` | 消息体：`enc=plain` 时为 JSON 明文；`enc=gzip` 时为 `base64(gzip(json))`——先标准 gzip（RFC1952，带 gzip 头），再 base64（基础编码、不带换行） |
+| `ts` | 发送时间戳（毫秒） |
+
+发送方需要压缩时（大报文省 Redis 内存）：`enc` 填 `gzip`，`body` 填「JSON → gzip → base64」的结果；不压缩就 `enc` 填 `plain`、`body` 放 JSON 明文。**只有 `body` 参与压缩，其余四个字段恒明文**。消费端按 `enc` 自动解码，两种都能收。`enc` 别自造取值：未知值会被兜底当明文处理，body 若实际是压缩数据就会解析失败、按毒消息 ACK 丢弃。
+
+```bash
+# 明文
+XADD LN_Topic * key 1815728470893350912 tag "" enc plain body '{"id":1,"name":"..."}' ts 1752480000000
+# 压缩：body 为 base64(gzip(json))
+XADD LN_Topic * key 1815728470893350913 tag "" enc gzip body 'H4sIAAAAAAAA...' ts 1752480000000
+```
+
+两个注意点：
+
+- **stream 名**：默认就是 topic；若开了中间件的环境隔离（`redis-stream.env-isolation`），实际 stream 名是 `topic_环境后缀`（如 `LN_Topic_test`），异构发送方要写到带后缀的那个。
+- **方向差**：mq-starter 自己**发送恒为 plain**（`enc` 照写、恒填 `plain`），压缩解码能力只在消费侧。需要 mq-starter 发送侧也压缩目前做不到，要改 driver。
+
+RocketMQ 模式没有这层信封也没有 `enc`——消息体就是裸 JSON，大报文由 RocketMQ 客户端自动透明压缩（发送端超约 4KB 自动压、消费端自动解），应用层不感知。
+
+但信封里 `key` 的角色在 RocketMQ 模式下由**报文头 KEYS 属性**承担（消费端去重取的是报文头 KEYS，**不读消息体里的任何字段**），异构发送方要遵守同一条契约：**KEYS 逐条唯一、同一条消息重投时保持不变**（发送时 `setKeys(唯一id)`，复用消息体里已有的唯一业务 id 即可）。注意 KEYS 在原生 RocketMQ 里只是可选的运维检索字段，不填也收发一切正常、多年无症状——但接入 mq-starter 消费后它就是去重钥匙：**为空或固定值时所有消息共用一把钥匙，第一条成功后去重窗口内其余全部被判重复、静默 ACK 丢弃**。发送方暂时无法保证 KEYS 唯一的，对应消费端必须显式 `@MqListener(idempotent = false)`。
 
 ## 架构
 
@@ -208,12 +260,14 @@ core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测
 
 | 属性 | 默认 | 说明 |
 |---|---|---|
-| `topic` / `group` | 必填 | 主题 / 消费组 |
+| `topic` | 必填 | 订阅的主题，和发送端 `send(topic, ...)` 传的 topic 一致才收得到 |
+| `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 过滤 |
-| `idempotent` | false | 是否去重 |
+| `tag` | `*` | 只收打了指定 tag 的消息（发送端三参 `send` 的第二参打 tag）。rocketmq 为 broker 服务端过滤、支持原生表达式（如 `A \|\| B`）；redis-stream 当前不过滤、全收。详见「消费」一节的 tag 小节 |
+| `idempotent` | true | 是否去重（默认开启，需 Redis 可达；单个监听器关闭写 `idempotent = false`，全局关闭配 `mq.idempotent.enabled=false`） |
 | `errorHandler` | `""` | 失败落库的 bean 名 |
 
 去重参数（前缀 `mq.idempotent`，driver 中立，都有默认值）：
@@ -246,14 +300,17 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`nam
 **`@MqListener` 方法不触发？**
 依次查：① 所在类是不是 Spring bean；② `topic` / `group` 和发送端是否一致；③ 方法签名是否合法（单条 `(payload)` 或 `(payload, MqContext)`，批量首参 `List<T>` 带具体元素类型）。
 
-**`idempotent = true` 在 rocketmq 下不生效？**
+**去重在 rocketmq 下不生效？**
 依次查：① Redis 是否可达（标准 `spring.redis.*` 配置）；② classpath 有没有 `mq-starter-idempotent-redis`（聚合坐标已含，单独引 rocketmq 模块要自己加）；③ 是否配了 `mq.idempotent.enabled=false`。开了去重但 Redis 连不上时，消费会抛连接异常走重投，日志可见，不是静默丢。
 
 **消息重复消费？**
-开 `idempotent = true`，两个 driver 都支持；确认去重 store 已装配（默认实现随 `mq-starter-idempotent-redis` 自动注册，不看 `mq.type`）。
+去重默认开启（`idempotent` 默认 true），正常不该重复。依次查：① 是否显式关掉了（监听器上 `idempotent = false`，或全局 `mq.idempotent.enabled=false`）；② 去重 store 是否已装配、Redis 是否可达（默认实现随 `mq-starter-idempotent-redis` 自动注册，不看 `mq.type`）。
 
-**批量消费收不到 / 停滞 / 丢消息？**
-没有 idle flush，攒不满 `batchSize` 不触发；整批失败会被吞掉不重投。低流量把 `batchSize` 调小或改单条消费。
+**对接异构发送方后，只有第一条消息能消费、后面成片丢失？**
+几乎必是去重 key 撞车：发送方没填或写死了 key（redis-stream 信封 `key` field / rocketmq 报文头 KEYS 属性），所有消息共用一把去重钥匙，第一条成功后其余全被判重复静默 ACK。让发送方保证 key 逐条唯一（见「Redis Stream 消息结构」一节），兑现前先在对应监听器上显式 `idempotent = false`。
+
+**批量消费收不到 / 延迟 / 丢消息？**
+攒不满 `batchSize` 时最多等 `batchTimeoutMs`（缺省 60 秒）才刷出，嫌慢就把它调小；整批失败会被吞掉不重投（务必配 `errorHandler`）；kill -9 / 断电会丢内存里没攒满的批，正常停机会强制刷出、不丢。
 
 **消费失败后消息不见了、事后查不到？**
 多半是没配 `errorHandler`：没配时单条 / 批量 / 重试耗尽的失败一律只 ACK 丢弃、不留痕迹。关键消费用 `@MqListener(errorHandler = "beanName")` 落库，见「进阶用法」。
@@ -272,7 +329,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 仍丢内存未满批）
 - [ ] 顺序消息
 - [ ] 事务消息
 

mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java

@@ -13,8 +13,11 @@ public @interface MqListener {
     int batchSize() default 1;
     int maxRetry() default 3;
     String tag() default "*";
-    /** 是否启用消费去重（按 key）。默认关闭。 */
-    boolean idempotent() default false;
+    /** 是否启用消费去重（按 key）。默认开启，不需要去重时显式置 false。 */
+    boolean idempotent() default true;
     /** 异常落库回调的 bean 名；空表示不落库。 */
     String errorHandler() default "";
+    /** 批量攒批的最长等待毫秒数（从本批第一条进入缓冲起算，到点未满也强制刷出，与满批先到先刷）。
+     *  仅批量监听器（batchSize>1 且首参 List）有意义；0 或负数 = 关闭超时刷出（回到攒满才刷）。缺省 60 秒。 */
+    int batchTimeoutMs() default 60_000;
 }

mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java

@@ -10,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.LongSupplier;
 
 /**
  * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。
@@ -18,16 +19,29 @@ import java.util.List;
 @Slf4j
 public class MqConsumeDispatcher {
 
-    // >1 时攒批；注意：无 idle flush，低流量下未满批会停滞、进程重启丢未入库批（与 legacy 一致，适合高流量流）。
+    // >1 时攒批；满批 / 超时(batchTimeoutMs) / 停机三路径刷出。kill -9 仍丢内存未刷批（上界 = 一个超时窗口内的量）。
     private final int batchSize;
+    /** 批量超时刷出阈值（毫秒，从本批第一条进入缓冲起算）；<=0 表示关闭超时刷出。 */
+    private final int batchTimeoutMs;
+    /** 当前批第一条进入缓冲的时刻（buffer 锁保护；缓冲清空后由下一条进入时重新记录）。 */
+    private long batchStartAtMs;
+    /** 时钟（Builder 可注入，单测用假时钟免真等待）。 */
+    private final LongSupplier clock;
     private final boolean idempotent;
     private final MqIdempotentStore store;
     private final MqConsumeErrorHandler errorHandler;
     private final MqBatchConsumer consumer;
     private final List<MqMessage> buffer = new ArrayList<>();
+    /** 串行化三条刷出路径（满批/超时/停机）的业务调用段：业务批量方法任一时刻至多一个调用在跑。 */
+    private final Object flushLock = new Object();
+    /** 停机打烊标志:forceFlush 置位后 dispatch 一律拒收抛 InProgress——driver 不 ACK、下次启动重投,
+     *  堵住"driver.stop() 异步返回后残余线程仍入缓冲"的丢消息缝隙。volatile 供 driver 线程快速路径读取。 */
+    private volatile boolean closed = false;
 
     private MqConsumeDispatcher(Builder b) {
         this.batchSize = b.batchSize;
+        this.batchTimeoutMs = b.batchTimeoutMs;
+        this.clock = b.clock;
         this.idempotent = b.idempotent;
         this.store = b.store;
         this.errorHandler = b.errorHandler;
@@ -35,6 +49,10 @@ public class MqConsumeDispatcher {
     }
 
     public void dispatch(MqMessage msg) throws Exception {
+        if (closed) {
+            // 已打烊:抛 InProgress 信号让 driver 不 ACK,消息留在 MQ 下次启动重投(不碰去重标记)。
+            throw new MqMessageInProgressException(msg.getKey());
+        }
         if (idempotent && store != null && !store.tryAcquire(msg.getKey())) {
             // 跳过分两类，决定 driver 能否 ACK：
             //  - 已 success：正常返回 → driver ACK，挡掉重复投递（正常去重）；
@@ -65,33 +83,97 @@ public class MqConsumeDispatcher {
         }
         // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。
         List<MqMessage> flush = null;
+        boolean rejected = false;
         synchronized (buffer) {
-            buffer.add(msg);
-            if (buffer.size() >= batchSize) {
-                flush = new ArrayList<>(buffer);
-                buffer.clear();
+            if (closed) {
+                rejected = true;
+            } else {
+                if (buffer.isEmpty()) {
+                    batchStartAtMs = clock.getAsLong();
+                }
+                buffer.add(msg);
+                if (buffer.size() >= batchSize) {
+                    flush = new ArrayList<>(buffer);
+                    buffer.clear();
+                }
             }
         }
+        if (rejected) {
+            // 打烊二次检查发生在 tryAcquire 之后：拒收前必须把 processing 标记复位（markFail 是
+            // Redis 往返，放 buffer 锁外做），否则重投每次都撞 stale 标记再抛 InProgress，白烧重试
+            // 计数——redis-stream 端可能业务一次没跑就被判耗尽。
+            if (idempotent && store != null) {
+                store.markFail(msg.getKey());
+            }
+            throw new MqMessageInProgressException(msg.getKey());
+        }
         if (flush != null) {
+            doFlush(flush);
+        }
+    }
+
+    /**
+     * 刷出一批：业务消费 → 成功逐条 markSuccess / 失败逐条 SINGLE 落库 + 吞。
+     * 满批 / 超时 / 停机三条路径都汇聚到这里，失败语义只写一份：
+     * 批量消息已逐条 ACK 无法精确重投整批 → 对齐 legacy 不卡队列：落库 + 吞掉，
+     * 不抛、不标 success（去重 processing 标记自然过期）。
+     */
+    private void doFlush(List<MqMessage> flush) {
+        synchronized (flushLock) {
             try {
                 consumer.consume(flush);
             } catch (Exception e) {
-                // 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列：
-                // 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。
                 log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e);
                 for (MqMessage m : flush) {
                     safeOnError(m, MqErrorIdentity.SINGLE, e);
                 }
                 return;
             }
-            if (idempotent && store != null) {
-                for (MqMessage m : flush) {
-                    store.markSuccess(m.getKey());
-                }
+        }
+        // markSuccess 逐条 store 往返，放锁外：flushLock 只串行化业务调用，
+        // 批已出缓冲、key 逐条独立，标记不必让下一批的业务刷出等待。
+        if (idempotent && store != null) {
+            for (MqMessage m : flush) {
+                store.markSuccess(m.getKey());
             }
         }
     }
 
+    /** 供定时线程调用：本批首条进入缓冲已满 batchTimeoutMs 则强制刷出未满批；未到点 / 空缓冲 / 未启用超时直接返回。 */
+    public void flushIfTimeout() {
+        if (batchTimeoutMs <= 0) return;
+        List<MqMessage> flush = null;
+        synchronized (buffer) {
+            if (!buffer.isEmpty() && clock.getAsLong() - batchStartAtMs >= batchTimeoutMs) {
+                flush = new ArrayList<>(buffer);
+                buffer.clear();
+            }
+        }
+        if (flush != null) {
+            doFlush(flush);
+        }
+    }
+
+    /** 停机强制刷出残留批并打烊(非空即刷,不看超时;打烊后 dispatch 一律拒收,见 closed 字段)。 */
+    public void forceFlush() {
+        List<MqMessage> flush = null;
+        synchronized (buffer) {
+            closed = true;
+            if (!buffer.isEmpty()) {
+                flush = new ArrayList<>(buffer);
+                buffer.clear();
+            }
+        }
+        if (flush != null) {
+            doFlush(flush);
+        }
+    }
+
+    /** registry 据此决定是否把本 dispatcher 纳入定时超时扫描。 */
+    public boolean needsTimeoutFlush() {
+        return batchSize > 1 && batchTimeoutMs > 0;
+    }
+
     /** 调用异常落库回调，并隔离其自身异常（落库是 best-effort，不得掩盖/中断主流程）。 */
     private void safeOnError(MqMessage msg, MqErrorIdentity identity, Exception cause) {
         if (errorHandler == null) return;
@@ -106,11 +188,17 @@ public class MqConsumeDispatcher {
 
     public static class Builder {
         private int batchSize = 1;
+        private int batchTimeoutMs;                              // 缺省 0 = 关闭（注解缺省 60s 由 registry 传入）
+        // 单调时钟:只做相对差值(now - batchStartAtMs),用 nanoTime 免受 NTP 回拨影响
+        // (墙钟回拨会让差值变负,批多躺回拨量,突破"延迟上界一个超时窗口"承诺)。
+        private LongSupplier clock = () -> System.nanoTime() / 1_000_000;
         private boolean idempotent = false;
         private MqIdempotentStore store;
         private MqConsumeErrorHandler errorHandler;
         private MqBatchConsumer consumer;
         public Builder batchSize(int n) { this.batchSize = n; return this; }
+        public Builder batchTimeoutMs(int ms) { this.batchTimeoutMs = ms; return this; }
+        public Builder clock(LongSupplier c) { this.clock = c; return this; }
         public Builder idempotent(boolean v) { this.idempotent = v; return this; }
         public Builder store(MqIdempotentStore s) { this.store = s; return this; }
         public Builder errorHandler(MqConsumeErrorHandler h) { this.errorHandler = h; return this; }

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java

@@ -47,6 +47,12 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor
             } else if (batch && meta.batchSize() <= 1) {
                 log.warn("[mq] @MqListener 首参为 List 但 batchSize={}，将按 1 条一批回调（如需攒批请设置 batchSize>1）: {}", meta.batchSize(), m);
             }
+            // batchTimeoutMs 仅批量攒批时有意义；0/负数 = 显式关闭超时，任何监听器上都无害不告警。
+            // 已知局限：显式写 60000（恰为缺省值）检测不出，可接受。
+            if (meta.batchSize() <= 1 && meta.batchTimeoutMs() != 60_000 && meta.batchTimeoutMs() > 0) {
+                log.warn("[mq] @MqListener batchTimeoutMs={} 仅对批量监听器（batchSize>1）生效，当前 batchSize={} 下不生效: {}",
+                        meta.batchTimeoutMs(), meta.batchSize(), m);
+            }
             m.setAccessible(true);
             endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch));
         }

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java

@@ -12,6 +12,9 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.SmartLifecycle;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class MqListenerRegistry implements SmartLifecycle {
@@ -20,6 +23,9 @@ public class MqListenerRegistry implements SmartLifecycle {
     private final MqIdempotentStore store;
     private final java.util.function.Function<String, MqConsumeErrorHandler> errorHandlerResolver;
     private volatile boolean running = false;
+    private final List<MqConsumeDispatcher> dispatchers = new ArrayList<>();
+    /** 批量超时刷出定时器：仅存在"批量且 batchTimeoutMs>0"监听器时创建（全应用一条单线程 daemon）。 */
+    private ScheduledExecutorService flushScheduler;
 
     public MqListenerRegistry(MqDriver driver, List<MqListenerEndpoint> endpoints) {
         this(driver, endpoints, null, null);
@@ -37,13 +43,14 @@ public class MqListenerRegistry implements SmartLifecycle {
         for (MqListenerEndpoint ep : endpoints) {
             MqListener meta = ep.getMeta();
             if (meta.idempotent() && store == null) {
-                log.warn("[mq] @MqListener(idempotent=true) 但容器中无 MqIdempotentStore bean，去重不生效: topic={} group={}",
+                log.warn("[mq] @MqListener 去重已启用（idempotent 默认 true）但容器中无 MqIdempotentStore bean，去重不生效: topic={} group={}",
                         meta.topic(), meta.group());
             }
             MqConsumeErrorHandler eh = resolveErrorHandler(meta);
 
             MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder()
                     .batchSize(meta.batchSize())
+                    .batchTimeoutMs(meta.batchTimeoutMs())
                     .consumer(batch -> consume(ep, batch));
             if (meta.idempotent()) {
                 db.idempotent(true).store(store);
@@ -52,6 +59,7 @@ public class MqListenerRegistry implements SmartLifecycle {
                 db.errorHandler(eh);
             }
             MqConsumeDispatcher dispatcher = db.build();
+            dispatchers.add(dispatcher);
 
             java.util.function.Consumer<MqMessage> retryExhausted = eh == null ? null
                     : msg -> eh.onError(msg, MqErrorIdentity.RETRY_EXHAUSTED, null);
@@ -65,6 +73,7 @@ public class MqListenerRegistry implements SmartLifecycle {
                     .build();
             driver.subscribe(sub);
         }
+        startFlushScheduler();
         running = true;
     }
 
@@ -108,7 +117,67 @@ public class MqListenerRegistry implements SmartLifecycle {
         }
     }
 
-    @Override public synchronized void stop() { running = false; driver.stop(); }
+    /** 有需要超时刷出的批量监听器才起定时线程；每 1s 扫一遍（触发最多晚 1s，对 60s 级超时无感）。 */
+    private void startFlushScheduler() {
+        List<MqConsumeDispatcher> timed = new ArrayList<>();
+        for (MqConsumeDispatcher d : dispatchers) {
+            if (d.needsTimeoutFlush()) timed.add(d);
+        }
+        if (timed.isEmpty()) return;
+        flushScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "mq-batch-flush");
+            t.setDaemon(true);
+            return t;
+        });
+        flushScheduler.scheduleWithFixedDelay(() -> {
+            // 任务体逐个 catch-all：异常不得逃逸，否则调度器取消后续执行，所有监听器的超时刷出静默失效。
+            for (MqConsumeDispatcher d : timed) {
+                try {
+                    d.flushIfTimeout();
+                } catch (Exception e) {
+                    log.error("[mq] 批量超时刷出异常（跳过本轮，不影响其他监听器）", e);
+                }
+            }
+        }, 1, 1, TimeUnit.SECONDS);
+    }
+
+    /** 包内可见，供单测断言定时器生命周期。 */
+    ScheduledExecutorService getFlushScheduler() { return flushScheduler; }
+
+    @Override public synchronized void stop() {
+        // 停机顺序：driver 先停（不再有新消息入缓冲）→ 定时器停 → 强刷残留批 → 置 running。
+        // driver.stop 异常必须吞掉：否则跳过后面整个强刷尾巴（缓冲里已 ACK 的批直接丢），
+        // 且 SmartLifecycle 回调没走完会卡 Spring 关闭。
+        try {
+            driver.stop();
+        } catch (Exception e) {
+            log.error("[mq] driver 停止失败（吞掉，继续停机流程强刷残留批）", e);
+        }
+        if (flushScheduler != null) {
+            // 温和关闭(不打断正在执行的最后一次刷出;后续执行取消,残留批由下方 forceFlush 接住)。
+            flushScheduler.shutdown();
+            // 必须等在途刷批跑完:超时快照已把批移出 buffer(下方 forceFlush 看不见它),
+            // 不等就返回会让 JVM 退出杀掉 daemon 线程,已逐条 ACK 的整批静默丢失。
+            try {
+                if (!flushScheduler.awaitTermination(30, TimeUnit.SECONDS)) {
+                    log.warn("[mq] 等待在途批量刷出超过 30s 仍未完成，放弃等待继续停机（该批可能丢失）");
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.warn("[mq] 等待在途批量刷出被中断，继续停机（该批可能丢失）");
+            }
+            flushScheduler = null;
+        }
+        for (MqConsumeDispatcher d : dispatchers) {
+            try {
+                d.forceFlush();
+            } catch (Exception e) {
+                log.error("[mq] 停机强制刷出失败（吞掉，不阻塞停机）", e);
+            }
+        }
+        dispatchers.clear();
+        running = false;
+    }
     @Override public boolean isRunning() { return running; }
     @Override public boolean isAutoStartup() { return true; }
     @Override public int getPhase() { return Integer.MAX_VALUE - 100; }

mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java

@@ -4,10 +4,17 @@ import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqConsumeErrorHandler;
 import com.njcn.mq.spi.MqErrorIdentity;
 import com.njcn.mq.spi.MqIdempotentStore;
+import com.njcn.mq.spi.MqMessageInProgressException;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -210,4 +217,314 @@ class MqConsumeDispatcherTest {
         assertEquals(1, failed.size(),
                 "errorHandler 抛异常时仍须 markFail，否则去重标记停在 processing 会导致 reclaim 时被去重挡掉而丢消息");
     }
+
+    @Test
+    void timeoutFlush_notDue_doesNothing() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        now[0] = 59_999;
+        d.flushIfTimeout();
+
+        assertEquals(0, consumed.size(), "未到超时不应刷出");
+    }
+
+    @Test
+    void timeoutFlush_due_flushesPartialBatchAndMarksSuccess() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        List<String> succeeded = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) { succeeded.add(key); }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b"));
+        now[0] = 60_000;
+        d.flushIfTimeout();
+
+        assertEquals(1, consumed.size(), "到点应刷出未满批");
+        assertEquals(2, consumed.get(0).size());
+        assertEquals(Arrays.asList("k1", "k2"), succeeded, "超时刷出成功后应逐条 markSuccess");
+    }
+
+    @Test
+    void timeoutFlush_onException_swallowsAndFilesSinglePerMessage() throws Exception {
+        long[] now = {0};
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        List<String> succeeded = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) { succeeded.add(key); }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeErrorHandler eh = (msg, identity, ex) -> errors.add(identity);
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .idempotent(true).store(store).errorHandler(eh)
+                .consumer(batch -> { throw new RuntimeException("db down"); })
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b"));
+        now[0] = 60_000;
+        assertDoesNotThrow(d::flushIfTimeout, "超时刷出失败应吞掉（消息已 ACK 无法重投）");
+
+        assertEquals(2, errors.size(), "失败应逐条落库 SINGLE");
+        assertEquals(MqErrorIdentity.SINGLE, errors.get(0));
+        assertEquals(0, succeeded.size(), "失败不得 markSuccess");
+    }
+
+    @Test
+    void fullBatchFlush_resetsTimer_noDoubleFlush() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(2).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b")); // 满批刷出
+        assertEquals(1, consumed.size());
+
+        now[0] = 120_000;
+        d.flushIfTimeout();
+        assertEquals(1, consumed.size(), "满批刷出后缓冲已空，超时扫描不应重复刷");
+    }
+
+    @Test
+    void emptyBuffer_timeoutAndForceFlush_doNothing() {
+        long[] now = {999_999};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.flushIfTimeout();
+        d.forceFlush();
+
+        assertEquals(0, consumed.size(), "空缓冲任何刷出都应是无动作");
+    }
+
+    @Test
+    void timeoutDisabled_neverTimeoutFlushes() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(0).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        now[0] = Long.MAX_VALUE / 2;
+        d.flushIfTimeout();
+
+        assertEquals(0, consumed.size(), "batchTimeoutMs=0 应关闭超时刷出");
+        assertFalse(d.needsTimeoutFlush(), "关闭超时的批量监听器不应纳入定时扫描");
+    }
+
+    @Test
+    void batchMarkSuccess_doesNotHoldFlushLock() throws Exception {
+        // #9:markSuccess 是逐条 store 往返,不得占着 flushLock(它只该串行化业务调用)——
+        // 慢 store 标记期间,其他线程的满批业务刷出不应被阻塞。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        AtomicBoolean first = new AtomicBoolean(true);
+        AtomicBoolean otherFlushCompletedDuringMark = new AtomicBoolean(false);
+        CountDownLatch otherFlushDone = new CountDownLatch(1);
+        MqConsumeDispatcher[] ref = new MqConsumeDispatcher[1];
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) {
+                if (first.compareAndSet(true, false)) {
+                    // 首批标记进行中:另一线程投满一批(其 doFlush 需要 flushLock)
+                    new Thread(() -> {
+                        try {
+                            ref[0].dispatch(MqMessage.of("k3", "", "c"));
+                            ref[0].dispatch(MqMessage.of("k4", "", "d"));
+                            otherFlushDone.countDown();
+                        } catch (Exception ignored) { }
+                    }).start();
+                    try {
+                        otherFlushCompletedDuringMark.set(otherFlushDone.await(800, TimeUnit.MILLISECONDS));
+                    } catch (InterruptedException ignored) { }
+                }
+            }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(2).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+        ref[0] = d;
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b")); // 满批 → doFlush → markSuccess(k1) 触发上面回调
+
+        assertTrue(otherFlushCompletedDuringMark.get(),
+                "markSuccess 期间另一线程的满批刷出不应被 flushLock 挡住(标记应在锁外)");
+        assertEquals(2, consumed.size(), "两批都应完成业务消费");
+    }
+
+    @Test
+    void defaultClock_freshBatchNotDue() throws Exception {
+        // 不注入 clock,守护缺省单调时钟的纳秒→毫秒换算:错用裸 nanoTime 会让差值以纳秒计,
+        // 刚入缓冲的批立即被判超时刷出。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.flushIfTimeout();
+
+        assertEquals(0, consumed.size(), "刚入缓冲远未到 60s,缺省时钟不应判超时");
+    }
+
+    @Test
+    void forceFlush_flushesLeftovers() throws Exception {
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.forceFlush();
+
+        assertEquals(1, consumed.size(), "停机强刷应刷出残留批");
+        assertEquals(1, consumed.get(0).size());
+    }
+
+    @Test
+    void timeoutFlush_newMessageAfterFlush_restartsTimer() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        now[0] = 60_000;
+        d.flushIfTimeout();                       // 第一批超时刷出
+        assertEquals(1, consumed.size());
+
+        now[0] = 61_000;
+        d.dispatch(MqMessage.of("k2", "", "b"));  // 新批从 61s 重新计时
+        now[0] = 120_000;
+        d.flushIfTimeout();
+        assertEquals(1, consumed.size(), "新批仅 59s 未到超时，不应刷出");
+
+        now[0] = 121_000;
+        d.flushIfTimeout();
+        assertEquals(2, consumed.size(), "新批满 60s 应刷出");
+    }
+
+    @Test
+    void timeoutThenFullBatch_noDupNoLoss() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(50).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        for (int i = 0; i < 49; i++) d.dispatch(MqMessage.of("t" + i, "", "x"));
+        now[0] = 60_000;
+        d.flushIfTimeout();                       // 超时刷出 49 条
+
+        for (int i = 0; i < 50; i++) d.dispatch(MqMessage.of("f" + i, "", "y")); // 满批刷出 50 条
+
+        assertEquals(2, consumed.size());
+        assertEquals(49, consumed.get(0).size(), "超时批应恰好 49 条");
+        assertEquals(50, consumed.get(1).size(), "满批应恰好 50 条，无重复无丢失");
+        Set<String> keys = new HashSet<>();
+        for (List<MqMessage> batch : consumed) for (MqMessage m : batch) keys.add(m.getKey());
+        assertEquals(99, keys.size(), "99 个 key 应全部恰好出现一次");
+    }
+
+    @Test
+    void forceFlush_thenDispatch_rejectsSoDriverWontAck() throws Exception {
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.forceFlush(); // 停机强刷 + 打烊
+        assertEquals(1, consumed.size(), "残留批应被停机强刷刷出");
+
+        assertThrows(MqMessageInProgressException.class,
+                () -> d.dispatch(MqMessage.of("k2", "", "b")),
+                "打烊后迟到的消息应拒收抛 InProgress,driver 不 ACK、下次启动重投");
+        d.forceFlush();
+        assertEquals(1, consumed.size(), "拒收的消息不得进入缓冲(再次强刷无动作)");
+    }
+
+    @Test
+    void closedDuringBatchDispatch_afterAcquire_resetsProcessingMark() {
+        // 竞态窗口:dispatch 已过打烊快速检查、tryAcquire 已把标记置 processing,
+        // 此刻停机 forceFlush 打烊 → 批量分支锁内二次检查拒收抛 InProgress。
+        // 拒收前必须 markFail 复位标记:否则重投每次都撞 stale processing 再抛 InProgress,
+        // 白烧重试计数(redis-stream 端 deliveryCount 超 maxRetry 且未配 errorHandler 时,
+        // 业务一次没跑就被判耗尽、静默丢弃)。
+        List<String> failed = new ArrayList<>();
+        MqConsumeDispatcher[] ref = new MqConsumeDispatcher[1];
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) {
+                ref[0].forceFlush(); // 模拟 tryAcquire 成功瞬间停机打烊
+                return true;
+            }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { failed.add(key); }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000)
+                .idempotent(true).store(store)
+                .consumer(batch -> { })
+                .build();
+        ref[0] = d;
+
+        assertThrows(MqMessageInProgressException.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "打烊后批量路径锁内二次检查应拒收");
+        assertEquals(Arrays.asList("k1"), failed,
+                "已 tryAcquire 的拒收必须 markFail 复位,否则重投一直撞 processing 标记");
+    }
+
+    @Test
+    void closed_rejectsSingleModeToo_beforeTouchingDedupStore() {
+        List<String> acquired = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { acquired.add(key); return true; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(batch -> { })
+                .build();
+
+        d.forceFlush(); // 空缓冲仅打烊
+        assertThrows(MqMessageInProgressException.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "打烊后单条路径同样拒收");
+        assertEquals(0, acquired.size(), "拒收须发生在去重之前,不得留下 processing 标记");
+    }
 }

mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java

@@ -14,6 +14,9 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import static org.junit.jupiter.api.Assertions.*;
@@ -149,4 +152,125 @@ class MqListenerRegistryTest {
         assertEquals(1, errors.size());
         assertEquals(MqErrorIdentity.RETRY_EXHAUSTED, errors.get(0), "重试耗尽应落库 RETRY_EXHAUSTED");
     }
+
+    static class NoTimeoutBatchBiz {
+        @MqListener(topic = "B0", group = "bg0", batchSize = 2, batchTimeoutMs = 0)
+        public void onBatch(List<String> msgs) { }
+    }
+    static class LeftoverBiz {
+        final List<String> events;
+        LeftoverBiz(List<String> events) { this.events = events; }
+        @MqListener(topic = "B", group = "bg", batchSize = 3)
+        public void onBatch(List<String> msgs) { events.add("flush:" + msgs.size()); }
+    }
+    /** fake driver：捕获 subscription，stop 时记录事件序。 */
+    static class OrderedDriver implements MqDriver {
+        final List<String> events;
+        MqSubscription captured;
+        OrderedDriver(List<String> events) { this.events = events; }
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() { events.add("driverStop"); }
+        @Override public void subscribe(MqSubscription s) { this.captured = s; }
+    }
+
+    @Test
+    void start_withBatchListener_createsFlushScheduler() {
+        BatchBiz biz = new BatchBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        MqListenerRegistry reg = new MqListenerRegistry(new CaptureDriver(), bpp.getEndpoints());
+        reg.start();
+        try {
+            assertNotNull(reg.getFlushScheduler(), "有批量监听器应创建超时刷出定时器");
+        } finally {
+            reg.stop();
+        }
+    }
+
+    @Test
+    void start_withoutTimedBatchListener_noScheduler() {
+        Biz single = new Biz();
+        MqListenerAnnotationBeanPostProcessor bpp1 = new MqListenerAnnotationBeanPostProcessor();
+        bpp1.postProcessAfterInitialization(single, "biz");
+        MqListenerRegistry reg1 = new MqListenerRegistry(new CaptureDriver(), bpp1.getEndpoints());
+        reg1.start();
+        assertNull(reg1.getFlushScheduler(), "无批量监听器不应创建定时器");
+
+        NoTimeoutBatchBiz off = new NoTimeoutBatchBiz();
+        MqListenerAnnotationBeanPostProcessor bpp2 = new MqListenerAnnotationBeanPostProcessor();
+        bpp2.postProcessAfterInitialization(off, "biz");
+        MqListenerRegistry reg2 = new MqListenerRegistry(new CaptureDriver(), bpp2.getEndpoints());
+        reg2.start();
+        assertNull(reg2.getFlushScheduler(), "批量但 batchTimeoutMs=0 也不应创建定时器");
+    }
+
+    @Test
+    void stop_flushesLeftoverBatchAfterDriverStopped() throws Exception {
+        List<String> events = new ArrayList<>();
+        LeftoverBiz biz = new LeftoverBiz(events);
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        OrderedDriver driver = new OrderedDriver(events);
+        MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints());
+        reg.start();
+        driver.captured.getListener().onMessage(MqMessage.of("k1", "", "a"));
+        driver.captured.getListener().onMessage(MqMessage.of("k2", "", "b")); // batchSize=3，2 条不满批
+
+        reg.stop();
+
+        assertEquals(Arrays.asList("driverStop", "flush:2"), events,
+                "停机应先停 driver 再强刷残留批（2 条未满批不丢）");
+        assertNull(reg.getFlushScheduler(), "停机后定时器应已关闭置空");
+    }
+
+    /** fake driver：stop 时抛异常。 */
+    static class FailingStopDriver implements MqDriver {
+        MqSubscription captured;
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() { throw new RuntimeException("driver stop fail"); }
+        @Override public void subscribe(MqSubscription s) { this.captured = s; }
+    }
+
+    @Test
+    void stop_whenDriverStopThrows_stillFlushesLeftoverAndCompletes() throws Exception {
+        List<String> events = new ArrayList<>();
+        LeftoverBiz biz = new LeftoverBiz(events);
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        FailingStopDriver driver = new FailingStopDriver();
+        MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints());
+        reg.start();
+        driver.captured.getListener().onMessage(MqMessage.of("k1", "", "a"));
+        driver.captured.getListener().onMessage(MqMessage.of("k2", "", "b")); // batchSize=3，2 条不满批
+
+        reg.stop(); // driver.stop 抛异常不得中断停机流程
+
+        assertEquals(Arrays.asList("flush:2"), events, "driver.stop 抛异常也必须强刷残留批（否则已 ACK 的批静默丢）");
+        assertNull(reg.getFlushScheduler(), "driver.stop 抛异常也应关闭定时器");
+        assertFalse(reg.isRunning(), "driver.stop 抛异常也应完成停机置 running=false");
+    }
+
+    @Test
+    void stop_waitsForInFlightTimedFlushBeforeReturning() throws Exception {
+        BatchBiz biz = new BatchBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        MqListenerRegistry reg = new MqListenerRegistry(new CaptureDriver(), bpp.getEndpoints());
+        reg.start();
+
+        // 模拟停机瞬间定时线程正在 doFlush 跑业务方法（超时快照已清空 buffer，批只在该线程手上）：
+        // 直接往真定时线程塞一个在途任务。stop 若不等它跑完，JVM 退出即静默丢批。
+        CountDownLatch entered = new CountDownLatch(1);
+        AtomicBoolean completed = new AtomicBoolean(false);
+        reg.getFlushScheduler().execute(() -> {
+            entered.countDown();
+            try { Thread.sleep(300); } catch (InterruptedException ignored) { }
+            completed.set(true);
+        });
+        assertTrue(entered.await(2, TimeUnit.SECONDS), "在途刷批任务应已开始执行");
+
+        reg.stop();
+
+        assertTrue(completed.get(), "stop 返回前应等待定时线程在途的刷批任务完成");
+    }
 }

mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java

@@ -101,4 +101,48 @@ class MqListenerScanTest {
         bpp.postProcessAfterInitialization(new BatchBiz(), "batchBiz");
         assertEquals(0, warnCount(), "batchSize 与首参匹配时不应有 WARN");
     }
+
+    static class TimeoutButSinglePayload {
+        @MqListener(topic = "W3", group = "wg3", batchTimeoutMs = 5000)
+        public void onOne(String msg) { }
+    }
+
+    static class BatchWithCustomTimeout {
+        @MqListener(topic = "W4", group = "wg4", batchSize = 10, batchTimeoutMs = 5000)
+        public void onBatch(List<String> msgs) { }
+    }
+
+    @Test
+    void scan_batchTimeoutOnSingleListener_warnsButNotBlocks() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new TimeoutButSinglePayload(), "biz");
+        assertEquals(1, bpp.getEndpoints().size(), "软校验不应阻断启动");
+        assertEquals(1, warnCount(), "batchSize<=1 配非缺省 batchTimeoutMs 应打一条 WARN");
+        assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchTimeoutMs"));
+    }
+
+    @Test
+    void scan_batchTimeoutOnBatchListener_noWarn() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new BatchWithCustomTimeout(), "biz");
+        assertEquals(0, warnCount(), "批量监听器配 batchTimeoutMs 属正常用法，不应 WARN");
+    }
+
+    static class TimeoutOffZeroOnSingle {
+        @MqListener(topic = "W5", group = "wg5", batchTimeoutMs = 0)
+        public void onOne(String msg) { }
+    }
+
+    static class TimeoutOffNegativeOnSingle {
+        @MqListener(topic = "W6", group = "wg6", batchTimeoutMs = -1)
+        public void onOne(String msg) { }
+    }
+
+    @Test
+    void scan_batchTimeoutOffOnSingleListener_noWarn() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new TimeoutOffZeroOnSingle(), "biz1");
+        bpp.postProcessAfterInitialization(new TimeoutOffNegativeOnSingle(), "biz2");
+        assertEquals(0, warnCount(), "0/负数 = 关闭超时，单条监听器显式关闭语义无害，不应误报 WARN");
+    }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -79,7 +79,7 @@ public class RedisStreamMqDriver implements MqDriver {
                 }
             } catch (Exception e) {
                 if (!running) break;
-                log.error("[mq-redis] loop error stream={} group={}", stream, group, e);
+                log.error("[mq-redis] 消费轮询异常 stream={} group={}", stream, group, e);
                 sleep(props.getBlockMs());
             }
         }
@@ -92,7 +92,7 @@ public class RedisStreamMqDriver implements MqDriver {
             String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
             payload = JSON.parseObject(json, sub.getPayloadType());
         } catch (Exception poison) {
-            log.error("[mq-redis] poison message acked, key={}", fields.get(StreamMessageConstant.F_KEY), poison);
+            log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={}", fields.get(StreamMessageConstant.F_KEY), poison);
             return true; // 毒消息直接 ACK，不卡队列
         }
         MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
@@ -102,10 +102,10 @@ public class RedisStreamMqDriver implements MqDriver {
         } catch (MqMessageInProgressException inProgress) {
             // 去重标记仍 processing（并发处理中/上次崩溃残留）：不 ACK，留 PEL，
             // 待 processing 标记过期后由 reclaimPending 重新 acquire 并真正重投。
-            log.debug("[mq-redis] in-progress, leave PEL key={}", msg.getKey());
+            log.debug("[mq-redis] 消息仍在处理中，暂不 ACK、留 PEL 待重投 key={}", msg.getKey());
             return false;
         } catch (Exception e) {
-            log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e);
+            log.warn("[mq-redis] 消费失败，留 PEL 待重投 key={} maxRetry={}", msg.getKey(), maxRetry, e);
             return false; // 留 PEL，由 reclaimPending 重投或落库
         }
     }
@@ -142,11 +142,11 @@ public class RedisStreamMqDriver implements MqDriver {
                     try {
                         sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields));
                     } catch (Exception ex) {
-                        log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex);
+                        log.error("[mq-redis] 重试耗尽落库回调执行异常 key={}（不影响主流程）", fields.get(StreamMessageConstant.F_KEY), ex);
                     }
                 }
                 redis.opsForStream().acknowledge(stream, group, rec.getId());
-                log.warn("[mq-redis] retry exhausted, acked key={} delivery={}", fields.get(StreamMessageConstant.F_KEY), deliveryCount);
+                log.warn("[mq-redis] 重试耗尽，已 ACK 终止 key={} delivery={}", fields.get(StreamMessageConstant.F_KEY), deliveryCount);
             } else {
                 boolean ack = dispatch(sub, fields, maxRetry);
                 if (ack) redis.opsForStream().acknowledge(stream, group, rec.getId());

mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java

@@ -54,7 +54,7 @@ public class RocketMqDriver implements MqDriver {
             consumer.start();
             consumers.add(consumer);
         } catch (Exception e) {
-            throw new IllegalStateException("rocket subscribe failed: " + sub.getTopic(), e);
+            throw new IllegalStateException("RocketMQ 订阅失败 topic=" + sub.getTopic(), e);
         }
     }
 
@@ -67,7 +67,7 @@ public class RocketMqDriver implements MqDriver {
             } catch (MqMessageInProgressException inProgress) {
                 // 去重"处理中"信号:不 ACK,交 broker 重投;绝不进耗尽落库分支(否则并发下静默丢消息)。
                 // 语义对齐 RedisStreamMqDriver#dispatch 对该异常的 return false(留 PEL)。
-                log.debug("[mq-rocket] in-progress, reconsume later key={}", ext.getKeys());
+                log.debug("[mq-rocket] 消息仍在处理中，返回 RECONSUME_LATER 待重投 key={}", ext.getKeys());
                 return ConsumeConcurrentlyStatus.RECONSUME_LATER;
             } catch (Exception e) {
                 if (ext.getReconsumeTimes() >= maxRetry) {
@@ -75,7 +75,7 @@ public class RocketMqDriver implements MqDriver {
                     //（对齐 redis-stream reclaim 耗尽语义,兑现"换 MQ 业务不动"契约)。
                     onRetryExhausted(sub, ext, e);
                 } else {
-                    log.warn("[mq-rocket] handle failed msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e);
+                    log.warn("[mq-rocket] 消费失败，交 broker 重投 msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e);
                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                 }
             }
@@ -85,7 +85,7 @@ public class RocketMqDriver implements MqDriver {
 
     /** 重试耗尽落库(best-effort:毒消息时 payload 为 null，至少保留 key/tag),并隔离 handler 自身异常。 */
     private void onRetryExhausted(MqSubscription sub, MessageExt ext, Exception cause) {
-        log.warn("[mq-rocket] retry exhausted, acked msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), cause);
+        log.warn("[mq-rocket] 重试耗尽，已 ACK 终止 msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), cause);
         Consumer<MqMessage> handler = sub.getRetryExhaustedHandler();
         if (handler == null) return;
         Object payload = null;
@@ -97,7 +97,7 @@ public class RocketMqDriver implements MqDriver {
         try {
             handler.accept(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
         } catch (Exception ex) {
-            log.error("[mq-rocket] retryExhausted handler error msgId={}（不影响主流程)", ext.getMsgId(), ex);
+            log.error("[mq-rocket] 重试耗尽落库回调执行异常 msgId={}（不影响主流程）", ext.getMsgId(), ex);
         }
     }