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>
feat(mq-starter-core): @MqListener batchSize 与首参 List 错配时打 WARN 软校验
2026-07-14 04:19:36 +01:00 · 2026-07-14 04:19:17 +01:00 · 2026-07-14 11:14:30 +08:00 · 2026-07-10 15:33:53 +08:00 · 2026-07-10 15:12:02 +08:00 · 2026-07-10 13:18:09 +08:00
43 changed files with 1752 additions and 82 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -30,3 +30,6 @@ hs_err_pid*
 replay_pid*

 .DS_Store
+
+# 本地文档（AI 生成的参考/教学资料，不纳入版本库）
+docs/
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -0,0 +1,100 @@
+# CLAUDE.md
+
+This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
+
+## 这是什么
+
+统一 MQ 抽象层：把 RocketMQ 和 Redis Stream 封装成一套接口。业务侧只用 `MqTemplate.send()` 发、`@MqListener` 收，底层走哪个 MQ 由 `mq.type` 一行配置决定，切换时业务代码不动。类比 JDBC —— 面向 `Connection`/`Statement` 写代码，换库只改驱动和连接串。
+
+面向使用者的完整说明在 `README.md`（快速开始、配置项、能力对照表、TODO），这里只补代码结构和踩坑点。
+
+## 构建 / 测试
+
+JDK 1.8、Maven 3.6+。`mvn` 命令在仓库根目录（多模块聚合）执行。
+
+```bash
+mvn clean install              # 全量构建 + 装本地仓库（业务方引 starter 前必须先 install）
+mvn test                       # 单元测试，不依赖任何外部中间件
+mvn verify                     # 含集成测试
+mvn -pl mq-starter-core test   # 只测某个模块（-am 连带编译依赖模块）
+mvn -pl mq-starter-core test -Dtest=MqConsumeDispatcherTest            # 单个测试类
+mvn -pl mq-starter-core test -Dtest=MqConsumeDispatcherTest#someMethod # 单个方法
+```
+
+**测试命名约定**：`*Test` 是纯单元测试；`*IT` 是集成测试，需要本地 Redis（`localhost:6379`）。IT 测试内部用 JUnit `assumeTrue` 自检连接，**连不上就自动 skip 而非失败**，所以即便 `mq-starter-redis-stream` 的 surefire 把 `*IT.java` 也纳入 `mvn test`，无 Redis 环境跑 `mvn test` 依然全绿。
+
+**依赖内部私服**：`redis-stream-springboot-starter`（`com.njcn.middle.stream.*`，Redis Stream 底层封装）和 `spring-boot-dependencies` BOM 等来自 Nexus（`192.168.1.22:8001`，见根 `pom.xml` 的 `distributionManagement`）。离线或私服不可达时构建会失败。
+
+**docs/ 整个目录被 gitignore**（含 `.md` 和 `.html`）。要看/验证 docs 里的改动，用 Read 或 ripgrep 直接读文件，**别用 `git diff`/`git status`**（它们看不到 docs 的任何变化）。
+
+## 模块结构
+
+```
+mq-starter-core             门面 + 注解扫描 + 消费链 + SPI 契约（中立，不依赖 Spring/Redis 的核心编排逻辑，可纯单测）
+mq-starter-idempotent-redis 消费去重的 Redis 存储（driver 中立，不按 mq.type 门控；任何 driver 有 Redis 就能用）
+mq-starter-redis-stream     Redis Stream driver（去重/批量/落库/reclaim 重试全接入；去重 store 依赖 idempotent-redis 模块）
+mq-starter-rocketmq         RocketMQ driver（收发 + 原生重试/死信 + RETRY_EXHAUSTED 落库 + 去重信号处理；去重 store 经聚合坐标间接获得）
+mq-spring-boot-starter      聚合 jar，业务方只引这一个（恒含 idempotent-redis）
+```
+
+> ⚠️ **两个 driver 仍有差异**（README 有对照表）：去重、失败落库两端已对齐（2026-07 起 rocketmq `idempotent=true` 生效，需 Redis 可达）；剩余差距是死信模型（redis 只落库 / rocketmq 原生 `%DLQ%`）、批量攒批语义、`concurrency` 两端都未兑现。改跨 driver 行为前先对照 README 能力表。
+>
+> **单独只引 `mq-starter-rocketmq`（不走聚合坐标）时去重需自行加 `mq-starter-idempotent-redis` 依赖**——rocketmq 模块本身不依赖它（刻意零耦合）。
+
+## 装配链路（关键：怎么被 Spring 拉起来的）
+
+各模块 `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`。
+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()`。
+
+## 核心执行流
+
+**发送**：`MqTemplate.send()` → `MqMessage`（中立信封，每条自动生成雪花 ID 当 `key`，去重要用）→ `driver.send()` 序列化投递。
+
+**消费**：`driver` 拉到消息 → 反序列化成 `MqSubscription.payloadType` → 回调 `MqSubscription.listener`（即 `MqConsumeDispatcher::dispatch`）→ 编排 **去重 →（攒批）→ 业务方法 → 成功标记 / 异常落库**。`MqListenerRegistry.consume()` 用反射调业务方法。
+
+### `MqConsumeDispatcher` —— 改这里务必读懂 ACK 语义
+
+`mq-starter-core/.../container/MqConsumeDispatcher.java` 是消费链的心脏，纯逻辑无外部依赖。有几处**不能想当然改动**的正确性约束（都写在注释里）：
+
+- **去重的两类跳过**决定 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` 时）+ 吞掉不抛。
+
+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 名解析——**未配 `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` 方法签名约束
+
+`MqListenerAnnotationBeanPostProcessor` 在启动扫描时强校验，违反直接抛 `IllegalStateException`：
+
+- 参数只能是 `(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"的全局视角。**注意本校验只在显式配了 `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 短超时探测（任一地址可达即通过）。
+
+**relaxed binding 坑**：读用户配置（如 `rocketmq.name-server`）时用 `Binder.get(env).bind(...)` 而非 `env.getProperty(精确键)`，否则 `camelCase` 等变体会漏读，把"其实配好了"误判成"没配"而错误 fail-fast。
+
+## 版本 / 依赖坑
+
+- 根 `pom.xml` 统一管理版本（Spring Boot 2.3.12、fastjson 1.2.83、hutool、lombok）。
+- `mq-starter-rocketmq` 额外 `import` 了 `spring-boot-dependencies` BOM —— 用于收敛 `rocketmq-spring-boot-starter` 传递进来的 Spring 版本（曾出现 5.2.15 ↔ 5.3 裂开导致测试跑不起来）。动 rocketmq 模块依赖时注意别破坏这个收敛。
+- **序列化两端都是 fastjson**（`RocketMqDriver` 与 `RedisStreamMqDriver` 均直接调 `JSON.toJSONString/parseObject`）。历史文档曾说"redis-stream 用 Jackson"——那是错的，Jackson 只出现在无关的上游 middleware，不在 driver 路径。但序列化目前**硬编码不可插拔**，替换需改两个 driver。
--- a/README.md
+++ b/README.md
@@ -1,46 +1,297 @@
-# mq-starter — 统一 MQ 抽象层（"MQ 界的 JDBC"）
+# mq-starter

-一个独立、全平台通用的 starter。业务配一行 `mq.type` 选底层（`rocketmq` | `redis-stream`），
-只面向 starter 接口编程，不关心用哪个 MQ、不管理生产者消费者；换 MQ = 改配置，业务代码一行不动。
+把消息队列封装成一套统一接口的 Spring Boot Starter。业务代码用 `MqTemplate` 发、用 `@MqListener` 收，底层到底走 RocketMQ 还是 Redis Stream，交给一行配置决定，切换的时候业务代码一行都不用动。

- 坐标 groupId：`com.njcn`，版本 `1.0.0-SNAPSHOT`（Phase 2 阶段本地 `mvn install`）
- JDK 1.8 / Spring Boot 2.3.12.RELEASE / `META-INF/spring.factories` 自动装配
- **零业务依赖**：不依赖 `message-api` 等任何业务模块
+思路和 JDBC 差不多：你面向 `Connection` / `Statement` 写代码，换 MySQL 还是 Oracle 只是改驱动和连接串；这里换 MQ 只是改 `mq.type`。

-## 业务视角（对外契约）
+[![License](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](./LICENSE)
+[![JDK](https://img.shields.io/badge/JDK-1.8-orange.svg)](https://www.oracle.com/java/)
+[![Spring Boot](https://img.shields.io/badge/Spring%20Boot-2.3.12-brightgreen.svg)](https://spring.io/projects/spring-boot)

 ```yaml
 mq:
-  type: redis-stream      # 或 rocketmq
+  type: redis-stream   # 换成 rocketmq 也行，业务代码不动
 ```

 ```java
-// 发
@Resource MqTemplate mq;
 mq.send("LN_Topic", payload);
-mq.send("File_Topic", nodeId, payload);   // 带路由 key

-// 收（业务方法打注解，零 MQ 概念）
@MqListener(topic = "LN_Topic", group = "ln_consumer")
-public void onFrontData(FrontData msg) { /* 纯业务 */ }
+public void onData(FrontData msg) {
+    // msg 已经反序列化好了，这里只写业务
+}
 ```

-## 模块结构
+## 为什么写它

-| 模块 | 职责 |
+之前项目里跟 MQ 打交道，producer / consumer 的注册、ack、序列化、失败重试、去重这些样板代码散落得到处都是，真要把 RocketMQ 换成别的，消费层基本得重写一遍。
+
+mq-starter 想做的就是把这些脏活收进 starter：业务侧只保留「发什么」和「收什么」，剩下的连接管理、线程模型、确认机制都不用管。顺手把消费侧几个常见需求（去重、批量、失败落库）也做成了注解开关。
+
+## 快速开始
+
+### 引入
+
+业务方只需要引一个聚合坐标，两套 driver 都打在里面，运行时按 `mq.type` 二选一装配：
+
+```xml
+<dependency>
+    <groupId>com.njcn</groupId>
+    <artifactId>mq-spring-boot-starter</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+</dependency>
+```
+
+目前还没发到中央仓库，先在项目根目录 `mvn install` 装到本地仓库用。
+
+### 配置
+
+用 Redis Stream：
+
+```yaml
+mq:
+  type: redis-stream
+spring:
+  redis:
+    host: localhost
+    port: 6379
+```
+
+用 RocketMQ：
+
+```yaml
+mq:
+  type: rocketmq
+rocketmq:
+  name-server: localhost:9876
+  producer:
+    group: my-producer-group
+```
+
+`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` 都不生效。
+
+### 发送
+
+```java
+@Resource
+private MqTemplate mq;
+
+mq.send("LN_Topic", frontData);            // topic + 消息体
+mq.send("File_Topic", nodeId, fileData);   // 带路由 key，映射成 tag
+```
+
+消息体传任意对象就行，driver 负责序列化（两端都是 fastjson，POJO 记得留公开无参构造和 getter/setter，收发两端用同一个类型）。每条消息会自动生成一个雪花 ID 当 key，去重时会用到。
+
+发送语义有两点要知道：
+
+- `send` 是**同步**调用、返回 `void`（不回传消息 ID）：方法正常返回就代表已入队（rocketmq 是等 broker 确认的 `syncSend`），失败直接抛异常，业务自己 try/catch 或交给上层事务回滚。
+- 消费端**反序列化失败**两端处理不一样：redis-stream 当"毒消息"直接 ACK 丢弃，rocketmq 会反复重试直到耗尽。改消息类字段时注意兼容，建议先自测能 round-trip。
+
+### 消费
+
+方法上打个注解就行，方法参数写你期望的类型，starter 按类型反序列化：
+
+```java
+@Component
+public class FrontDataConsumer {
+
+    @MqListener(topic = "LN_Topic", group = "ln_consumer")
+    public void onFrontData(FrontData msg) {
+        // ...
+    }
+
+    // 想拿 key / tag / timestamp / headers 这些元数据，就再加一个 MqContext 参数
+    @MqListener(topic = "File_Topic", group = "file_consumer")
+    public void onFile(FileData msg, MqContext ctx) {
+        String key = ctx.getKey();
+    }
+}
+```
+
+方法签名只认这两种：`(payload)` 或 `(payload, MqContext)`。
+
+## 进阶用法
+
+去重、批量、失败落库都是 `@MqListener` 上的属性，业务方法本身不用改：
+
+```java
+// 去重：重复消息不会再进业务（去重状态存 Redis，两个 driver 都支持，见下方说明）
+@MqListener(topic = "LN_Topic", group = "g", idempotent = true)
+public void onData(FrontData msg) { }
+
+// 批量：攒够 batchSize 条一次给你，参数要用 List
+@MqListener(topic = "LN_Topic", group = "g", batchSize = 50)
+public void onBatch(List<FrontData> msgs) { }
+
+// 失败落库：消费失败的消息交给你自己的 handler 落库
+@MqListener(topic = "LN_Topic", group = "g", errorHandler = "myErrorHandler")
+public void onData2(FrontData msg) { }
+```
+
+落库 handler 自己实现一个 bean：
+
+```java
+@Component("myErrorHandler")
+public class MyErrorHandler implements MqConsumeErrorHandler {
+    @Override
+    public void onError(MqMessage message, MqErrorIdentity identity, Exception ex) {
+        // identity 区分是单次失败(SINGLE)还是重试耗尽(RETRY_EXHAUSTED)
+        // 把 message 存到异常表，后面人工或定时补偿
+    }
+}
+```
+
+> ⚠️ **不配 `errorHandler` = 失败静默丢弃**。未指定 `errorHandler` 时，单条 / 批量 / 重试耗尽的任何消费失败最终都会被 ACK、不落库、不留痕迹（ACK 终止是内建行为，与是否落库无关）。关键消费务必配上 `errorHandler` 落库，否则失败即丢、事后无从排查。
+>
+> ⚠️ **批量有几个坑**：参数必须写带具体元素类型的 `List<FrontData>`（裸 `List` 启动扫描直接抛 `IllegalStateException`）；没有 idle flush，低流量攒不满会一直等，进程重启会丢内存里没攒满的那批；整批业务失败只回调 SINGLE 落库、不重投（所以批量务必配 `errorHandler`）。只适合高流量、能接受这些取舍的场景。批量方法同样可以加第二参 `MqContext`（拿到的是该批最后一条的元数据）。
+>
+> 另外 `batchSize` 与首参是否 `List` **错配时不报错**：启动时打一条 WARN 日志、仍按首参决定走不走批量（`batchSize=50` 配单个 payload 退化为逐条；`List` 首参配 `batchSize` 缺省按 1 条一批）。见启动日志 `[mq] @MqListener batchSize=...`。
+
+## 两个 driver 不太一样
+
+两套实现的能力大头已经对齐：去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义：
+
+| 能力 | Redis Stream | RocketMQ |
+|---|:---:|:---:|
+| 收发 | ✅ | ✅ |
+| 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 |
+| 去重 `idempotent` | ✅（需 Redis） | ✅（需 Redis） |
+| 失败落库 `RETRY_EXHAUSTED`（需配 `errorHandler`，否则仅 ACK 丢弃） | ✅ | ✅ 落库+ACK |
+| 失败落库 `SINGLE` | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） |
+| 批量 `batchSize` | ⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致） | ⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致） |
+| 重试 | ✅ PEL reclaim | ✅ 原生重投 |
+| 死信队列 | ❌ 只落库 | ⚠️ 原生 `%DLQ%` 通常被耗尽 ACK 旁路（InProgress 白跑烧光计数时仍会进） |
+| 顺序消息 | ❌ | ❌ 用的 Concurrently |
+| 事务 / 延迟消息 | ❌ | ❌ 没暴露 |
+| 序列化 | fastjson | fastjson |
+
+> ⚠️ **切换 `mq.type` 业务代码零改动，但三处运行时行为会跟着翻转**，切换前按这三条过一遍消费方假设：
+> 1. **tag 过滤**：redis-stream 消费端不过滤（全收）、rocketmq 走 broker 服务端按 tag 过滤——依赖 tag 分流的业务切 rocketmq 会"只收匹配的"、切回 redis 会"收到全部"。
+> 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 没配好而影响启动。
+
+## 架构
+
+```
+             业务代码（只见 MqTemplate / @MqListener）
+                            │
+     ┌──────────────────────┴──────────────────────┐
+     │                mq-starter-core               │
+     │  MqTemplate · @MqListener 扫描 · MqMessage    │
+     │  消费链：去重 → 攒批 → 消费 → 落库/重试        │
+     │  SPI：MqDriver / MqSubscription / ...         │
+     └──────────────┬──────────────────┬────────────┘
+        @ConditionalOnProperty(mq.type)  二选一
+     ┌──────────────┴─────┐   ┌─────────┴───────────┐
+     │ redis-stream driver │   │  rocketmq driver    │
+     └─────────────────────┘   └─────────────────────┘
+```
+
+五个模块：
+
+| 模块 | 干什么 |
 |---|---|
-| `mq-starter-core` | 门面 `MqTemplate` + `@MqListener` 注解扫描 + 中立信封 `MqMessage` + driver SPI + 生命周期 |
-| `mq-starter-rocketmq` | rocket driver（基于官方 rocketmq-spring 编程式注册） |
-| `mq-starter-redis-stream` | redis driver（复用 `redis-stream-springboot-starter` 低层件，零改动 Phase 1） |
-| `mq-spring-boot-starter` | 聚合 jar，业务方只引这一个坐标 |
+| `mq-starter-core` | 门面 `MqTemplate`、`@MqListener` 扫描注册、中立信封 `MqMessage`、消费处理链、几个 SPI，还有基于 `SmartLifecycle` 的启停 |
+| `mq-starter-idempotent-redis` | 消费去重的 Redis 存储（driver 中立，不看 `mq.type`，有 Redis 就装配；配置前缀 `mq.idempotent`） |
+| `mq-starter-redis-stream` | Redis Stream 实现（XADD / 消费组 XREADGROUP / XACK / XPENDING+XCLAIM reclaim），去重经依赖 idempotent-redis 获得 |
+| `mq-starter-rocketmq` | RocketMQ 实现，基于 rocketmq-spring 编程式注册 |
+| `mq-spring-boot-starter` | 聚合 jar，业务方引这一个就够（恒含 idempotent-redis） |

-## 装配机制
+core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测。想看更细的架构图文（消费链、SPI 契约、driver 能力矩阵、改进 Roadmap），打开 [`docs/mq-starter-全景手册.html`](./docs/mq-starter-全景手册.html)；上手和排错看本 README 就够了。

-`@ConditionalOnProperty("mq.type")` 静态二选一：两套 driver 各只在对应值下装配，
-同一上下文只存在一个 `MqDriver` bean。
+## 配置项

-## 关联设计
+`@MqListener` 的属性：

- 设计稿：`cn-data-platform/docs/superpowers/specs/2026-06-25-mq-starter-design.md`
- 能力清单与轻量版范围：`cn-data-platform/docs/2026-06-25-mq-starter能力清单与轻量版范围.html`
- 实现计划（轻量版 Phase 2）：`cn-data-platform/docs/superpowers/plans/2026-06-25-mq-starter-轻量版-phase2.md`
+| 属性 | 默认 | 说明 |
+|---|---|---|
+| `topic` / `group` | 必填 | 主题 / 消费组 |
+| `concurrency` | 1 | 并发消费线程数，**当前两 driver 均未按此字段兑现**（redis 单线程 poll、rocketmq 用原生线程池默认值） |
+| `batchSize` | 1 | 攒批大小，大于 1 要用 `List` 参数；与首参是否 `List` 错配时启动打 WARN、仍按首参跑 |
+| `maxRetry` | 3 | 最大重试次数（redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker `maxReconsumeTimes`；被去重标记挡回的 InProgress 白跑**两端都计入**） |
+| `tag` | `*` | tag 过滤 |
+| `idempotent` | false | 是否去重 |
+| `errorHandler` | `""` | 失败落库的 bean 名 |
+
+去重参数（前缀 `mq.idempotent`，driver 中立，都有默认值）：
+
+| 配置 | 默认 | 说明 |
+|---|---|---|
+| `enabled` | true | 逃生开关，显式 false 关闭去重 store 装配 |
+| `processing-ttl` | 60s | 「处理中」标记存多久，必须大于最长单条处理耗时，且**两端都有上界**（被标记挡回的 InProgress 白跑会烧重试计数）：redis-stream 下需 **≥ `reclaim-min-idle`** 且 **< `maxRetry` × `reclaim-min-idle`**（下界防慢消费双消费；标记活过全部 reclaim 轮次则 delivery 烧光、业务没重跑就判耗尽——未配 `errorHandler` 即静默丢弃），默认 30 ≤ 60 < 90 自洽；rocketmq 下需小于前 `maxRetry` 轮 broker 重投累计时长（默认 10s+30s+1m≈100s），否则并发中 / 崩溃残留的消息把 `reconsumeTimes` 烧光、进原生 `%DLQ%` |
+| `success-ttl` | 300s | 「已完成」标记存多久（去重窗口） |
+| `fail-ttl` | 300s | 失败标记保留多久（fail 态不挡重投，重投后允许重新消费） |
+| `key-prefix` | `mq:idem:` | 去重键前缀 |
+
+> 旧键 `mq.redis-stream.idempotent.*` 仍向后兼容（会打一条弃用告警），中立键和旧键同时配置时以 `mq.idempotent.*` 为准。建议尽快迁移。
+
+Redis Stream 这边的参数（前缀 `mq.redis-stream`）：
+
+| 配置 | 默认 | 说明 |
+|---|---|---|
+| `reclaim-min-idle` | 30s | 消息投出去多久没 ack 才 reclaim 重投，要大于单条处理耗时；**不要调到比 `processing-ttl` 还大**（需保持 `processing-ttl ≥ reclaim-min-idle`，否则存活的慢消费者会被 reclaim 重投副本并行双消费）；也别调得太小——每次 reclaim（xClaim）都使 PEL delivery +1，被 processing 挡回的白跑同样计入，需保持 `maxRetry × reclaim-min-idle > processing-ttl`，否则崩溃残留会在业务没重跑的情况下烧光重试直接判耗尽 |
+
+Redis 连接走标准 `spring.redis.*`；更底层的 `redis-stream.*`（readCount / blockMs / consumerName 等）来自内部 stream 中间件（`redis-stream-springboot-starter`），一般不用动。
+
+RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`name-server`、`producer.group` 等）。
+
+## 排错 FAQ
+
+**启动找不到 `MqTemplate` bean / 一启动就抛 MQ 相关异常？**
+先看 `mq.type`：拼错了默认直接 fail-fast（`IllegalStateException` + 中文排查提示，检查依赖和拼写）；完全没配则默认装 rocketmq，`rocketmq.*` 没配好通常直接启动报错。想用 redis-stream 必须显式写 `mq.type: redis-stream`。详见上文「配置」一节。
+
+**`@MqListener` 方法不触发？**
+依次查：① 所在类是不是 Spring bean；② `topic` / `group` 和发送端是否一致；③ 方法签名是否合法（单条 `(payload)` 或 `(payload, MqContext)`，批量首参 `List<T>` 带具体元素类型）。
+
+**`idempotent = true` 在 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`）。
+
+**批量消费收不到 / 停滞 / 丢消息？**
+没有 idle flush，攒不满 `batchSize` 不触发；整批失败会被吞掉不重投。低流量把 `batchSize` 调小或改单条消费。
+
+**消费失败后消息不见了、事后查不到？**
+多半是没配 `errorHandler`：没配时单条 / 批量 / 重试耗尽的失败一律只 ACK 丢弃、不留痕迹。关键消费用 `@MqListener(errorHandler = "beanName")` 落库，见「进阶用法」。
+
+**开了去重后某条消息迟迟不消费 / 反复重投？**
+多半是「处理中」标记没过期挡住了重投：消费者崩溃残留标记，或单条处理耗时超过 `mq.idempotent.processing-ttl`。等 TTL 过期会自动恢复；调 `processing-ttl` 时注意配置表里的上下界约束。rocketmq 下持续重投烧光 `maxReconsumeTimes` 会进原生 `%DLQ%`（进死信，不是丢消息）。
+
+**从 redis-stream 切到 rocketmq 后行为变了（多收 / 漏收、并发不一样）？**
+正常，见「两个 driver 不太一样」里的三条运行时翻转：tag 过滤、并发模型、顺序保证。
+
+## 已知限制 / TODO
+
+按优先级列，详细的在 [`docs/mq-starter-改进优化清单.md`](./docs/mq-starter-改进优化清单.md)。
+
+P0，放量到生产前想补的：
+
+- [x] 失败落库、去重、批量消费（core 已做）
+- [x] 给 rocketmq driver 补上去重和 `RETRY_EXHAUSTED` 落库，跟 redis-stream 对齐（去重 store 已抽成 driver 中立的 `mq-starter-idempotent-redis`）
+- [ ] 批量消费加 idle flush，现在低流量下攒不满会卡着，进程重启还会丢内存里没攒满的那批
+- [ ] 顺序消息
+- [ ] 事务消息
+
+P1：延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔（两端都是 fastjson，但硬编码换不掉）、广播消费。
+
+P2：并发度语义统一、headers 完整透传、监控告警。（启动期 fail-fast 自检 —— `mq.type` 错配 / 中间件连不上时启动报错 —— 已于 2026-07-09 实现，逃生开关 `mq.startup-check.enabled=false`；详见全景手册 Roadmap 13。）
+
+## 构建
+
+```bash
+git clone https://github.com/EdmundNightmare/mq-starter.git
+cd mq-starter
+mvn clean install
+```
+
+需要 JDK 1.8+ 和 Maven 3.6+。单元测试 `mvn test` 不依赖外部中间件；带真实 Redis 的集成测试（`*IT`，连 `localhost:6379`）在 `mvn verify` 阶段跑，测试内部会自检连接、连不上 Redis 自动跳过而非失败。
+
+有想法欢迎提 issue 或 PR，优先认领上面 P0 的几项，提交前跑一下 `mvn verify`，改动尽量带上测试。
+
+## License
+
+[Apache License 2.0](./LICENSE)，Copyright 2026 EdmundNightmare。
--- a/mq-spring-boot-starter/pom.xml
+++ b/mq-spring-boot-starter/pom.xml
@@ -5,6 +5,7 @@
    <artifactId>mq-spring-boot-starter</artifactId>
    <dependencies>
        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-idempotent-redis</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-rocketmq</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-redis-stream</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
    </dependencies>
--- a/mq-starter-core/pom.xml
+++ b/mq-starter-core/pom.xml
@@ -8,6 +8,7 @@
        <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
        <dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>${hutool.version}</version></dependency>
        <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
+        <dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.30</version><scope>provided</scope></dependency>
        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><version>${springboot.version}</version><optional>true</optional></dependency>
        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
    </dependencies>
--- a/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqDriverPresenceAutoConfiguration.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqDriverPresenceAutoConfiguration.java
@@ -0,0 +1,51 @@
+package com.njcn.mq.autoconfig;
+
+import com.njcn.mq.spi.MqDriver;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.beans.factory.SmartInitializingSingleton;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+
+/**
+ * ③ 通用校验：显式配了 {@code mq.type}，却没有装配任何 {@link MqDriver}，
+ * 说明忘引对应 starter 或 mq.type 拼错 —— 启动失败并给中文提示。
+ *
+ * <p>为何“有没有 MqDriver”就够：两个 driver 的 AutoConfiguration 都用
+ * {@code @ConditionalOnProperty(name="mq.type", havingValue=...)} 按 mq.type 条件装配，
+ * 因此“存在 MqDriver bean”已等价于“配的 mq.type 有对应 driver”。
+ *
+ * <p>仅当显式配了 {@code mq.type} 时装配；没配则完全不介入（存量零影响）。
+ * 放在 core 是因为：redis/rocketmq 包没引入时其模块内的类自己都不加载，无法自证缺失；
+ * 只有 core 有“到底有没有 MqDriver”的全局视角。
+ */
+@Configuration
+@ConditionalOnProperty(name = "mq.type")
+public class MqDriverPresenceAutoConfiguration implements SmartInitializingSingleton {
+
+    private final ObjectProvider<MqDriver> drivers;
+    private final Environment env;
+
+    public MqDriverPresenceAutoConfiguration(ObjectProvider<MqDriver> drivers, Environment env) {
+        this.drivers = drivers;
+        this.env = env;
+    }
+
+    @Override
+    public void afterSingletonsInstantiated() {
+        if (!MqStartupChecks.enabled(env)) {
+            return;
+        }
+        String configured = env.getProperty("mq.type");
+        // 用 stream() 而非 getIfAvailable()：同时引多个 driver 包时不会抛 NoUniqueBeanDefinitionException
+        if (drivers.stream().findAny().isPresent()) {
+            return; // 有 driver 装配，放行
+        }
+        throw new IllegalStateException(
+                "MQ 启动失败：已配置 mq.type=" + configured + "，但没有装配任何 MqDriver。\n"
+                + "请检查：\n"
+                + "  1) 是否引入对应 starter 依赖（redis-stream 需 mq-starter-redis-stream + spring-boot-starter-data-redis；\n"
+                + "     rocketmq 需 mq-starter-rocketmq + rocketmq starter）；\n"
+                + "  2) mq.type 是否拼写正确（支持：redis-stream / rocketmq）。");
+    }
+}
--- a/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqStartupChecks.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqStartupChecks.java
@@ -0,0 +1,22 @@
+package com.njcn.mq.autoconfig;
+
+import org.springframework.core.env.Environment;
+
+/**
+ * MQ 启动校验的公共工具：统一读取逃生开关 {@code mq.startup-check.enabled}（默认 true）。
+ *
+ * <p>只接受严格的 {@code true}/{@code false}（忽略大小写、去首尾空白）。仅当显式配为
+ * {@code false} 时才关闭校验；缺省或任何非法值（如 {@code disabled}）都按默认 {@code true}
+ * 继续校验——避免逃生开关自身因非法布尔值在启动期抛类型转换异常，反而把启动炸掉。
+ */
+public final class MqStartupChecks {
+
+    private MqStartupChecks() {
+    }
+
+    /** @return true 表示应执行启动校验；false 表示使用者主动关闭。 */
+    public static boolean enabled(Environment env) {
+        String value = env.getProperty("mq.startup-check.enabled");
+        return value == null || !"false".equalsIgnoreCase(value.trim());
+    }
+}
--- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java
@@ -4,6 +4,8 @@ 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 lombok.extern.slf4j.Slf4j;

 import java.util.ArrayList;
 import java.util.Collections;
@@ -13,8 +15,10 @@ import java.util.List;
 * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。
 * 纯逻辑、不依赖 spring / redis，便于单测。重试本身交给 driver 的 MQ 原生重投。
 */
+@Slf4j
 public class MqConsumeDispatcher {

+    // >1 时攒批；注意：无 idle flush，低流量下未满批会停滞、进程重启丢未入库批（与 legacy 一致，适合高流量流）。
    private final int batchSize;
    private final boolean idempotent;
    private final MqIdempotentStore store;
@@ -32,19 +36,27 @@ public class MqConsumeDispatcher {

    public void dispatch(MqMessage msg) throws Exception {
        if (idempotent && store != null && !store.tryAcquire(msg.getKey())) {
-            return; // processing 中或已 success，跳过
+            // 跳过分两类，决定 driver 能否 ACK：
+            //  - 已 success：正常返回 → driver ACK，挡掉重复投递（正常去重）；
+            //  - 仍 processing（并发处理中，或消费者在 mark 前崩溃残留的标记）：抛信号异常 →
+            //    driver 不 ACK、保留消息，待 processing 标记过期后真正重投。否则该消息可能从未被
+            //    成功消费却被去重挡回并 ACK → 静默丢消息。
+            if (store.isCompleted(msg.getKey())) {
+                return;
+            }
+            throw new MqMessageInProgressException(msg.getKey());
        }
        if (batchSize <= 1) {
            try {
                consumer.consume(Collections.singletonList(msg));
            } catch (Exception e) {
-                if (errorHandler != null) {
-                    errorHandler.onError(msg, MqErrorIdentity.SINGLE, e);
-                }
+                // errorHandler 自身失败不得跳过 markFail：否则去重标记停在 processing，
+                // reclaim 重投时会被去重挡掉并 ACK → 静默丢消息。
+                safeOnError(msg, MqErrorIdentity.SINGLE, e);
                if (idempotent && store != null) {
                    store.markFail(msg.getKey());
                }
-                throw e; // 交 driver 重投
+                throw e; // 抛原始异常，交 driver 重投
            }
            if (idempotent && store != null) {
                store.markSuccess(msg.getKey());
@@ -66,10 +78,9 @@ public class MqConsumeDispatcher {
            } catch (Exception e) {
                // 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列：
                // 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。
-                if (errorHandler != null) {
-                    for (MqMessage m : flush) {
-                        errorHandler.onError(m, MqErrorIdentity.SINGLE, e);
-                    }
+                log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e);
+                for (MqMessage m : flush) {
+                    safeOnError(m, MqErrorIdentity.SINGLE, e);
                }
                return;
            }
@@ -81,6 +92,16 @@ public class MqConsumeDispatcher {
        }
    }

+    /** 调用异常落库回调，并隔离其自身异常（落库是 best-effort，不得掩盖/中断主流程）。 */
+    private void safeOnError(MqMessage msg, MqErrorIdentity identity, Exception cause) {
+        if (errorHandler == null) return;
+        try {
+            errorHandler.onError(msg, identity, cause);
+        } catch (Exception ehEx) {
+            log.error("[mq] errorHandler 落库失败 key={}（不影响主流程）", msg.getKey(), ehEx);
+        }
+    }
+
    public static Builder builder() { return new Builder(); }

    public static class Builder {
--- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java
@@ -1,6 +1,7 @@
 package com.njcn.mq.container;
 import com.njcn.mq.annotation.MqListener;
 import com.njcn.mq.core.MqContext;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.core.annotation.AnnotatedElementUtils;
@@ -8,6 +9,7 @@ import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.List;

+@Slf4j
 public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor {

    private final List<MqListenerEndpoint> endpoints = new ArrayList<>();
@@ -39,6 +41,12 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor
            } else {
                payloadType = first;
            }
+            // 走不走批量只由首参是否 List 决定，batchSize 与之错配时静默按首参跑，这里 WARN 提示而不阻断启动。
+            if (!batch && meta.batchSize() > 1) {
+                log.warn("[mq] @MqListener batchSize={} 但首参非 List，batchSize 不生效、将按单条逐一回调（如需批量攒批请把首参改为 List<T>）: {}", meta.batchSize(), m);
+            } else if (batch && meta.batchSize() <= 1) {
+                log.warn("[mq] @MqListener 首参为 List 但 batchSize={}，将按 1 条一批回调（如需攒批请设置 batchSize>1）: {}", meta.batchSize(), m);
+            }
            m.setAccessible(true);
            endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch));
        }
--- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java
@@ -8,10 +8,12 @@ import com.njcn.mq.spi.MqDriver;
 import com.njcn.mq.spi.MqIdempotentStore;
 import com.njcn.mq.spi.MqMessageListener;
 import com.njcn.mq.spi.MqSubscription;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.SmartLifecycle;
 import java.util.ArrayList;
 import java.util.List;

+@Slf4j
 public class MqListenerRegistry implements SmartLifecycle {
    private final MqDriver driver;
    private final List<MqListenerEndpoint> endpoints;
@@ -34,6 +36,10 @@ public class MqListenerRegistry implements SmartLifecycle {
        if (running) return;
        for (MqListenerEndpoint ep : endpoints) {
            MqListener meta = ep.getMeta();
+            if (meta.idempotent() && store == null) {
+                log.warn("[mq] @MqListener(idempotent=true) 但容器中无 MqIdempotentStore bean，去重不生效: topic={} group={}",
+                        meta.topic(), meta.group());
+            }
            MqConsumeErrorHandler eh = resolveErrorHandler(meta);

            MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder()
@@ -64,7 +70,12 @@ public class MqListenerRegistry implements SmartLifecycle {

    private MqConsumeErrorHandler resolveErrorHandler(MqListener meta) {
        if (errorHandlerResolver == null || meta.errorHandler().isEmpty()) return null;
-        return errorHandlerResolver.apply(meta.errorHandler());
+        try {
+            return errorHandlerResolver.apply(meta.errorHandler());
+        } catch (Exception e) {
+            throw new IllegalStateException("@MqListener(errorHandler=\"" + meta.errorHandler()
+                    + "\") 指定的 bean 未找到: topic=" + meta.topic() + " group=" + meta.group(), e);
+        }
    }

    /** 业务调用适配：批量 endpoint 调 (List&lt;payload&gt;[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */
--- a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java
@@ -21,4 +21,22 @@ public interface MqIdempotentStore {

    /** 消费失败,标记 fail(允许后续重新消费)。 */
    void markFail(String key);
+
+    /**
+     * {@link #tryAcquire(String)} 返回 false(跳过)后,判断该 key 是否已<b>消费成功</b>。
+     *
+     * <p>用于让 driver 区分两类"跳过",这决定了 driver 能否 ACK 该消息:
+     * <ul>
+     *   <li>已 success:可安全 ACK —— 正常去重,挡掉重复投递;</li>
+     *   <li>仍 processing(并发处理中,或消费者在 mark 前崩溃残留的标记):<b>不可 ACK</b> ——
+     *       该消息可能从未被成功消费,必须保留在 MQ,待 processing 标记过期后真正重投;
+     *       否则会被去重挡回并 ACK,造成<b>静默丢消息</b>。</li>
+     * </ul>
+     *
+     * @return true=已成功(driver 可 ACK);false=processing 中(driver 应保留待重投)。
+     *         默认返回 true 以保持旧行为(向后兼容);具备状态机的实现应覆写。
+     */
+    default boolean isCompleted(String key) {
+        return true;
+    }
 }
--- a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageInProgressException.java
+++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageInProgressException.java
@@ -0,0 +1,21 @@
+package com.njcn.mq.spi;
+
+/**
+ * 控制流信号:消费去重标记仍处于 {@code processing}(并发处理中,或消费者在 mark 前崩溃残留的标记)。
+ *
+ * <p>本次投递既不能视为成功(否则 driver 会 ACK,导致该消息若从未被成功消费即<b>静默丢失</b>),
+ * 也不属于业务处理失败。driver 捕获后应<b>保留消息(不 ACK)</b>,待 processing 标记过期后由 MQ 真正重投。
+ *
+ * <p>不携带堆栈(纯控制流,规避噪音与开销)。
+ */
+public class MqMessageInProgressException extends RuntimeException {
+
+    public MqMessageInProgressException(String key) {
+        super("message in-progress (dedup processing), leave for replay: key=" + key);
+    }
+
+    @Override
+    public synchronized Throwable fillInStackTrace() {
+        return this;
+    }
+}
--- a/mq-starter-core/src/main/resources/META-INF/spring.factories
+++ b/mq-starter-core/src/main/resources/META-INF/spring.factories
@@ -1,2 +1,3 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-com.njcn.mq.autoconfig.MqCoreAutoConfiguration
+com.njcn.mq.autoconfig.MqCoreAutoConfiguration,\
+com.njcn.mq.autoconfig.MqDriverPresenceAutoConfiguration
--- a/mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqDriverPresenceAutoConfigurationTest.java
+++ b/mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqDriverPresenceAutoConfigurationTest.java
@@ -0,0 +1,53 @@
+package com.njcn.mq.autoconfig;
+
+import com.njcn.mq.spi.MqDriver;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+class MqDriverPresenceAutoConfigurationTest {
+
+    private final ApplicationContextRunner runner = new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(MqDriverPresenceAutoConfiguration.class));
+
+    @Test
+    void noMqType_doesNotFail() {
+        runner.run(ctx -> assertThat(ctx).hasNotFailed());
+    }
+
+    @Test
+    void mqTypeButNoDriver_fails() {
+        runner.withPropertyValues("mq.type=redis-stream")
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("redis-stream");
+                });
+    }
+
+    @Test
+    void mqTypeWithDriver_ok() {
+        runner.withPropertyValues("mq.type=redis-stream")
+                .withBean("someDriver", MqDriver.class, () -> mock(MqDriver.class))
+                .run(ctx -> assertThat(ctx).hasNotFailed());
+    }
+
+    @Test
+    void illegalEnabledValue_stillChecks_insteadOfCrashing() {
+        // 逃生开关填非法布尔值不应炸在类型转换上，而应按默认继续校验：
+        // 此处无 driver，应走到中文③提示，而不是 Spring 的「Invalid boolean」类型转换异常。
+        runner.withPropertyValues("mq.type=redis-stream", "mq.startup-check.enabled=disabled")
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("没有装配任何 MqDriver");
+                });
+    }
+
+    @Test
+    void disabled_skipsCheck() {
+        runner.withPropertyValues("mq.type=redis-stream", "mq.startup-check.enabled=false")
+                .run(ctx -> assertThat(ctx).hasNotFailed());
+    }
+}
--- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java
+++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java
@@ -142,4 +142,72 @@ class MqConsumeDispatcherTest {

        assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success");
    }
+
+    @Test
+    void idempotent_skipInProgress_throwsSoDriverWontAckAndReplays() {
+        // 崩溃残留场景:去重标记停在 processing(tryAcquire=false 且 isCompleted=false)。
+        // 消费者在 tryAcquire 之后、markSuccess/markFail 之前崩溃,消息从未被成功消费。
+        // 若此时静默返回,driver 会把这条消息 ACK → 静默丢消息(Critical)。
+        // 正确行为:抛异常,使 driver 不 ACK、保留消息,待 processing 标记过期后真正重投。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return false; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+            @Override public boolean isCompleted(String key) { return false; }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        assertThrows(Exception.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "processing 残留(非 success)跳过时应抛异常,使 driver 不 ACK、留待过期后重投,避免静默丢消息");
+        assertEquals(0, consumed.size(), "不应消费");
+    }
+
+    @Test
+    void idempotent_skipCompleted_returnsNormallySoDriverAcks() {
+        // 已成功的重复投递:应被静默去重(driver 正常 ACK),不抛异常、不重复消费。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return false; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+            @Override public boolean isCompleted(String key) { return true; }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        assertDoesNotThrow(() -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "已成功的重复投递应被静默去重(driver ACK),不抛异常");
+        assertEquals(0, consumed.size(), "不应重复消费");
+    }
+
+    @Test
+    void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() {
+        List<String> failed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { failed.add(key); }
+        };
+        // errorHandler 自身抛异常（落库 sink 挂掉，常与错误风暴同时发生）
+        MqConsumeErrorHandler eh = (m, id, ex) -> { throw new RuntimeException("sink down"); };
+        RuntimeException boom = new RuntimeException("biz boom");
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store).errorHandler(eh)
+                .consumer(batch -> { throw boom; })
+                .build();
+
+        Exception thrown = assertThrows(Exception.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")));
+
+        assertSame(boom, thrown, "errorHandler 失败不应掩盖原始消费异常");
+        assertEquals(1, failed.size(),
+                "errorHandler 抛异常时仍须 markFail，否则去重标记停在 processing 会导致 reclaim 时被去重挡掉而丢消息");
+    }
 }
--- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java
+++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java
@@ -1,11 +1,35 @@
 package com.njcn.mq.container;
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import com.njcn.mq.annotation.MqListener;
 import com.njcn.mq.core.MqContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
 import java.util.List;
 import static org.junit.jupiter.api.Assertions.*;

 class MqListenerScanTest {
+
+    private final ListAppender<ILoggingEvent> logAppender = new ListAppender<>();
+
+    @BeforeEach
+    void attachAppender() {
+        logAppender.start();
+        ((Logger) LoggerFactory.getLogger(MqListenerAnnotationBeanPostProcessor.class)).addAppender(logAppender);
+    }
+
+    @AfterEach
+    void detachAppender() {
+        ((Logger) LoggerFactory.getLogger(MqListenerAnnotationBeanPostProcessor.class)).detachAppender(logAppender);
+    }
+
+    private long warnCount() {
+        return logAppender.list.stream().filter(e -> e.getLevel() == Level.WARN).count();
+    }
    static class Biz {
        @MqListener(topic = "LN_Topic", group = "ln_consumer")
        public void onData(String msg) { }
@@ -39,4 +63,42 @@ class MqListenerScanTest {
        assertTrue(ep.isBatch(), "List 参数应识别为批量");
        assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T");
    }
+
+    static class BatchSizeButSinglePayload {
+        @MqListener(topic = "W1", group = "wg1", batchSize = 50)
+        public void onOne(String msg) { }
+    }
+
+    static class ListButBatchSizeOne {
+        @MqListener(topic = "W2", group = "wg2")
+        public void onList(List<String> msgs) { }
+    }
+
+    @Test
+    void scan_batchSizeGtOneButSinglePayload_warnsAndRunsAsSingle() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new BatchSizeButSinglePayload(), "biz");
+        MqListenerEndpoint ep = bpp.getEndpoints().get(0);
+        assertFalse(ep.isBatch(), "首参非 List 应按单条跑");
+        assertEquals(1, warnCount(), "batchSize>1 但首参非 List 应打一条 WARN");
+        assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchSize"));
+    }
+
+    @Test
+    void scan_listParamButBatchSizeOne_warnsAndRunsAsBatch() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new ListButBatchSizeOne(), "biz");
+        MqListenerEndpoint ep = bpp.getEndpoints().get(0);
+        assertTrue(ep.isBatch(), "首参 List 仍按批量跑");
+        assertEquals(1, warnCount(), "首参 List 但 batchSize<=1 应打一条 WARN");
+        assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchSize"));
+    }
+
+    @Test
+    void scan_matchedSignatures_noWarn() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new Biz(), "biz");
+        bpp.postProcessAfterInitialization(new BatchBiz(), "batchBiz");
+        assertEquals(0, warnCount(), "batchSize 与首参匹配时不应有 WARN");
+    }
 }
--- a/mq-starter-idempotent-redis/pom.xml
+++ b/mq-starter-idempotent-redis/pom.xml
@@ -0,0 +1,30 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
+    <artifactId>mq-starter-idempotent-redis</artifactId>
+    <name>mq-starter-idempotent-redis 消费去重 Redis 存储(driver 中立)</name>
+    <dependencies>
+        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-autoconfigure</artifactId><version>${autoconfigure.version}</version></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>${springboot.version}</version></dependency>
+        <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
+    </dependencies>
+    <build>
+        <plugins>
+            <plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId></plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <includes>
+                        <include>**/*Test.java</include>
+                        <include>**/*Tests.java</include>
+                        <include>**/*IT.java</include>
+                    </includes>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+</project>
--- a/mq-starter-idempotent-redis/src/main/java/com/njcn/mq/idempotent/redis/MqIdempotentProperties.java
+++ b/mq-starter-idempotent-redis/src/main/java/com/njcn/mq/idempotent/redis/MqIdempotentProperties.java
@@ -0,0 +1,42 @@
+package com.njcn.mq.idempotent.redis;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+/**
+ * 消费去重配置(driver 中立,前缀 {@code mq.idempotent})。
+ *
+ * <p>自 mq.redis-stream.idempotent.* 中立化迁移而来,<b>默认值与迁移前完全一致</b>,不改变现有行为。
+ * 旧键仍向后兼容(见 MqIdempotentRedisAutoConfiguration 的 Binder 回退),但会打弃用告警。
+ *
+ * <pre>
+ * mq:
+ *   idempotent:
+ *     enabled: true                # 逃生开关:显式 false 关闭去重 store 装配
+ *     key-prefix: "mq:idem:"
+ *     processing-ttl: 60s          # 应大于单条业务处理时延
+ *     success-ttl: 300s            # 防重复消费窗口
+ *     fail-ttl: 300s               # 允许重新消费前的保留窗口
+ * </pre>
+ */
+@Data
+@ConfigurationProperties(prefix = "mq.idempotent")
+public class MqIdempotentProperties {
+
+    /** 逃生开关:显式 false 关闭去重 store 装配。默认开。 */
+    private boolean enabled = true;
+
+    /** 去重键前缀。 */
+    private String keyPrefix = "mq:idem:";
+
+    /** processing 标记 TTL:应大于单条业务处理时延。 */
+    private Duration processingTtl = Duration.ofSeconds(60);
+
+    /** success 标记 TTL:防重复消费窗口。 */
+    private Duration successTtl = Duration.ofSeconds(300);
+
+    /** fail 标记 TTL:允许重新消费前的保留窗口。 */
+    private Duration failTtl = Duration.ofSeconds(300);
+}
--- a/mq-starter-idempotent-redis/src/main/java/com/njcn/mq/idempotent/redis/MqIdempotentRedisAutoConfiguration.java
+++ b/mq-starter-idempotent-redis/src/main/java/com/njcn/mq/idempotent/redis/MqIdempotentRedisAutoConfiguration.java
@@ -0,0 +1,86 @@
+package com.njcn.mq.idempotent.redis;
+
+import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
+import com.njcn.mq.spi.MqIdempotentStore;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 消费去重 store 自动装配(driver 中立)。
+ *
+ * <p><b>刻意不按 mq.type 门控</b>,只按「Redis 在场」门控 —— rocketmq / redis-stream / 未来任何
+ * driver 部署下,只要 classpath 有 spring-data-redis 且上下文有 StringRedisTemplate 就装配。
+ *
+ * <p><b>启动安全约束</b>:本装配与 store 构造均不得 eager 连接 / PING Redis(保持 lettuce 懒连接)。
+ * classpath 有 lettuce 时 Spring Boot 会用 localhost:6379 默认值自动造出 StringRedisTemplate,
+ * 因此 @ConditionalOnBean 几乎恒真,它不代表 Redis 已正确配置;真正的安全网是懒连接 +
+ * core 仅在 idempotent=true 时才调用 store(未开去重时 bean 造了也无人调用,零副作用)。
+ *
+ * <p><b>旧键兼容</b>:mq.redis-stream.idempotent.* 已中立化为 mq.idempotent.*。用 Binder 逐键回退
+ * (而非 env.getProperty 精确键,后者会漏 relaxed binding 变体):中立键显式设置→中立;否则旧键
+ * 显式设置→旧键值+弃用告警;否则默认。
+ */
+@Slf4j
+@Configuration
+@AutoConfigureAfter(RedisAutoConfiguration.class)
+@AutoConfigureBefore(MqCoreAutoConfiguration.class)
+@ConditionalOnClass(StringRedisTemplate.class)
+@ConditionalOnProperty(name = "mq.idempotent.enabled", havingValue = "true", matchIfMissing = true)
+@EnableConfigurationProperties(MqIdempotentProperties.class)
+public class MqIdempotentRedisAutoConfiguration {
+
+    private static final String NEUTRAL_PREFIX = "mq.idempotent.";
+    private static final String LEGACY_PREFIX = "mq.redis-stream.idempotent.";
+
+    @Bean
+    @ConditionalOnBean(StringRedisTemplate.class)
+    @ConditionalOnMissingBean(MqIdempotentStore.class)
+    public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis,
+                                               MqIdempotentProperties props,
+                                               Environment environment) {
+        Binder binder = Binder.get(environment);
+        List<String> usedLegacyKeys = new ArrayList<>();
+        String keyPrefix = resolve(binder, "key-prefix", String.class, props.getKeyPrefix(), usedLegacyKeys);
+        Duration processingTtl = resolve(binder, "processing-ttl", Duration.class, props.getProcessingTtl(), usedLegacyKeys);
+        Duration successTtl = resolve(binder, "success-ttl", Duration.class, props.getSuccessTtl(), usedLegacyKeys);
+        Duration failTtl = resolve(binder, "fail-ttl", Duration.class, props.getFailTtl(), usedLegacyKeys);
+        if (!usedLegacyKeys.isEmpty()) {
+            log.warn("[mq] 检测到已弃用配置 mq.redis-stream.idempotent.*，请迁移到 mq.idempotent.*；本次已按旧键取值：{}", usedLegacyKeys);
+        }
+        return new RedisMqIdempotentStore(redis, keyPrefix,
+                processingTtl.getSeconds(), successTtl.getSeconds(), failTtl.getSeconds());
+    }
+
+    /**
+     * 单字段回退:中立键显式设置则以中立为准(props 已按中立键绑定,字段值即中立值);
+     * 否则旧键显式设置则用旧键值并记入弃用清单;否则用默认(同样取 props 字段值)。
+     */
+    private static <T> T resolve(Binder binder, String field, Class<T> type,
+                                 T neutralOrDefault, List<String> usedLegacyKeys) {
+        if (binder.bind(NEUTRAL_PREFIX + field, type).isBound()) {
+            return neutralOrDefault;
+        }
+        T legacy = binder.bind(LEGACY_PREFIX + field, type).orElse(null);
+        if (legacy != null) {
+            usedLegacyKeys.add(LEGACY_PREFIX + field);
+            return legacy;
+        }
+        return neutralOrDefault;
+    }
+}
--- a/mq-starter-idempotent-redis/src/main/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStore.java
+++ b/mq-starter-idempotent-redis/src/main/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStore.java
@@ -1,4 +1,4 @@
-package com.njcn.mq.driver.redis;
+package com.njcn.mq.idempotent.redis;

 import com.njcn.mq.spi.MqIdempotentStore;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -41,6 +41,8 @@ public class RedisMqIdempotentStore implements MqIdempotentStore {
            return true;
        }
        // key 已存在：fail（上次失败）允许重新消费；processing / success 一律挡回。
+        // 注意：fail→reopen 的 get+set 非原子，跨实例对「共享同一 key 的不同消息」可能都放行；
+        // 同一条消息由 stream PEL/claim 序列化，故仅影响不同消息共享 key 的边缘场景。
        if (FAIL.equals(redis.opsForValue().get(k))) {
            redis.opsForValue().set(k, PROCESSING, Duration.ofSeconds(processingTtl));
            return true;
@@ -57,4 +59,11 @@ public class RedisMqIdempotentStore implements MqIdempotentStore {
    public void markFail(String key) {
        redis.opsForValue().set(prefix + key, FAIL, Duration.ofSeconds(failTtl));
    }
+
+    @Override
+    public boolean isCompleted(String key) {
+        // 仅 success 视为"已完成"(driver 可 ACK)。processing(并发处理中/崩溃残留)与
+        // 无记录(标记已过期)一律 false，使 driver 保留消息、待重投，避免静默丢消息。
+        return SUCCESS.equals(redis.opsForValue().get(prefix + key));
+    }
 }
--- a/mq-starter-idempotent-redis/src/main/resources/META-INF/spring.factories
+++ b/mq-starter-idempotent-redis/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.njcn.mq.idempotent.redis.MqIdempotentRedisAutoConfiguration
--- a/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/MqIdempotentLegacyKeyCompatTest.java
+++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/MqIdempotentLegacyKeyCompatTest.java
@@ -0,0 +1,81 @@
+package com.njcn.mq.idempotent.redis;
+
+import com.njcn.mq.spi.MqIdempotentStore;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.*;
+
+/**
+ * 旧键 mq.redis-stream.idempotent.* 向后兼容(Binder 逐键回退)。
+ *
+ * <p>规则:中立键 mq.idempotent.* 显式设置则以中立为准;否则旧键显式设置则用旧键值并打一条
+ * 聚合弃用 WARN;否则用默认。store 字段私有(类内容不改),故通过 mock Redis 的行为断言验证生效值。
+ */
+@ExtendWith(OutputCaptureExtension.class)
+class MqIdempotentLegacyKeyCompatTest {
+
+    @SuppressWarnings("unchecked")
+    private final ValueOperations<String, String> ops = mock(ValueOperations.class);
+    private final StringRedisTemplate redis = mock(StringRedisTemplate.class);
+
+    private ApplicationContextRunner runner() {
+        when(redis.opsForValue()).thenReturn(ops);
+        return new ApplicationContextRunner()
+                .withConfiguration(AutoConfigurations.of(MqIdempotentRedisAutoConfiguration.class))
+                .withBean(StringRedisTemplate.class, () -> redis);
+    }
+
+    @Test
+    void legacyKeysOnly_takeEffect_andLogDeprecationWarn(CapturedOutput output) {
+        runner().withPropertyValues(
+                        "mq.redis-stream.idempotent.key-prefix=legacy:idem:",
+                        "mq.redis-stream.idempotent.processing-ttl=15s")
+                .run(ctx -> {
+                    MqIdempotentStore store = ctx.getBean(MqIdempotentStore.class);
+                    store.tryAcquire("k");
+                    // 旧键前缀 + 旧键 processing TTL 均生效
+                    verify(ops).setIfAbsent("legacy:idem:k", "processing", Duration.ofSeconds(15));
+                    // 弃用告警必须点名旧前缀,给使用者迁移线索
+                    assertThat(output).contains("mq.redis-stream.idempotent");
+                });
+    }
+
+    @Test
+    void neutralKeyWins_whenBothSet_noWarnForOverriddenLegacy(CapturedOutput output) {
+        runner().withPropertyValues(
+                        "mq.idempotent.processing-ttl=20s",
+                        "mq.redis-stream.idempotent.processing-ttl=15s")
+                .run(ctx -> {
+                    MqIdempotentStore store = ctx.getBean(MqIdempotentStore.class);
+                    store.tryAcquire("k");
+                    // 中立键优先;key-prefix 未设,用默认 mq:idem:
+                    verify(ops).setIfAbsent("mq:idem:k", "processing", Duration.ofSeconds(20));
+                    // 没有任何字段实际落到旧键 → 不打弃用告警
+                    assertThat(output).doesNotContain("mq.redis-stream.idempotent");
+                });
+    }
+
+    @Test
+    void successAndFailTtl_alsoFallBackToLegacy() {
+        runner().withPropertyValues(
+                        "mq.redis-stream.idempotent.success-ttl=111s",
+                        "mq.redis-stream.idempotent.fail-ttl=222s")
+                .run(ctx -> {
+                    MqIdempotentStore store = ctx.getBean(MqIdempotentStore.class);
+                    store.markSuccess("k");
+                    verify(ops).set("mq:idem:k", "success", Duration.ofSeconds(111));
+                    store.markFail("k");
+                    verify(ops).set("mq:idem:k", "fail", Duration.ofSeconds(222));
+                });
+    }
+}
--- a/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/MqIdempotentPropertiesTest.java
+++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/MqIdempotentPropertiesTest.java
@@ -0,0 +1,24 @@
+package com.njcn.mq.idempotent.redis;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * {@link MqIdempotentProperties} 默认值守护:必须与迁移前 RedisStreamMqProperties.Idempotent
+ * 完全一致(processing 60s / success 300s / fail 300s / 前缀 mq:idem:),保证迁移不改变现有行为。
+ */
+class MqIdempotentPropertiesTest {
+
+    @Test
+    void defaults_matchLegacyRedisStreamIdempotentDefaults() {
+        MqIdempotentProperties p = new MqIdempotentProperties();
+        assertTrue(p.isEnabled(), "去重装配默认开启,仅显式 false 关闭");
+        assertEquals("mq:idem:", p.getKeyPrefix());
+        assertEquals(Duration.ofSeconds(60), p.getProcessingTtl());
+        assertEquals(Duration.ofSeconds(300), p.getSuccessTtl());
+        assertEquals(Duration.ofSeconds(300), p.getFailTtl());
+    }
+}
--- a/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/MqIdempotentRedisAutoConfigurationTest.java
+++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/MqIdempotentRedisAutoConfigurationTest.java
@@ -0,0 +1,85 @@
+package com.njcn.mq.idempotent.redis;
+
+import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
+import com.njcn.mq.spi.MqIdempotentStore;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.core.io.support.SpringFactoriesLoader;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+/**
+ * 装配条件矩阵(ApplicationContextRunner,不连真实 Redis):
+ * 关键契约是「不按 mq.type 门控、只按 Redis 在场门控」—— rocketmq 部署也能装出 store。
+ */
+class MqIdempotentRedisAutoConfigurationTest {
+
+    private final ApplicationContextRunner runner = new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(MqIdempotentRedisAutoConfiguration.class));
+
+    @Test
+    void withStringRedisTemplate_storeAutoConfigured_regardlessOfMqType() {
+        runner.withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class))
+                // 故意设 mq.type=rocketmq:装配必须与 mq.type 无关(这正是 rocketmq 用上去重的关键)
+                .withPropertyValues("mq.type=rocketmq")
+                .run(ctx -> {
+                    assertThat(ctx).hasSingleBean(MqIdempotentStore.class);
+                    assertThat(ctx.getBean(MqIdempotentStore.class)).isInstanceOf(RedisMqIdempotentStore.class);
+                });
+    }
+
+    @Test
+    void withoutRedisBean_noStoreAndNoStartupFailure() {
+        runner.run(ctx -> {
+            assertThat(ctx).hasNotFailed();
+            assertThat(ctx).doesNotHaveBean(MqIdempotentStore.class);
+        });
+    }
+
+    @Test
+    void enabledFalse_disablesStore() {
+        runner.withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class))
+                .withPropertyValues("mq.idempotent.enabled=false")
+                .run(ctx -> assertThat(ctx).doesNotHaveBean(MqIdempotentStore.class));
+    }
+
+    @Test
+    void userProvidedStore_backsOff() {
+        MqIdempotentStore custom = mock(MqIdempotentStore.class);
+        runner.withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class))
+                .withBean("customStore", MqIdempotentStore.class, () -> custom)
+                .run(ctx -> {
+                    assertThat(ctx).hasSingleBean(MqIdempotentStore.class);
+                    assertThat(ctx.getBean(MqIdempotentStore.class)).isSameAs(custom);
+                });
+    }
+
+    @Test
+    void declaresAutoConfigureOrderContract() {
+        // 真实 Spring Boot 按类名字母序处理 autoconfig:com.njcn.mq.autoconfig(core) 先于
+        // com.njcn.mq.idempotent(本模块)。显式声明顺序契约,守护 spec §4.4 的装配语义。
+        AutoConfigureBefore before = MqIdempotentRedisAutoConfiguration.class.getAnnotation(AutoConfigureBefore.class);
+        assertThat(before).as("必须声明 @AutoConfigureBefore(MqCoreAutoConfiguration)").isNotNull();
+        assertThat(before.value()).contains(MqCoreAutoConfiguration.class);
+
+        AutoConfigureAfter after = MqIdempotentRedisAutoConfiguration.class.getAnnotation(AutoConfigureAfter.class);
+        assertThat(after).as("必须声明 @AutoConfigureAfter(RedisAutoConfiguration),否则 @ConditionalOnBean(StringRedisTemplate) 求值时模板尚未注册").isNotNull();
+        assertThat(after.value()).contains(RedisAutoConfiguration.class);
+    }
+
+    @Test
+    void registeredInSpringFactories() {
+        List<String> names = SpringFactoriesLoader.loadFactoryNames(EnableAutoConfiguration.class,
+                getClass().getClassLoader());
+        assertThat(names).contains(MqIdempotentRedisAutoConfiguration.class.getName());
+    }
+}
--- a/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreIT.java
+++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreIT.java
@@ -1,4 +1,4 @@
-package com.njcn.mq.driver.redis;
+package com.njcn.mq.idempotent.redis;

 import org.junit.jupiter.api.Test;
 import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
--- a/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreTest.java
+++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreTest.java
@@ -0,0 +1,34 @@
+package com.njcn.mq.idempotent.redis;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * {@link RedisMqIdempotentStore#isCompleted(String)} 纯单测(mock StringRedisTemplate，不依赖真实 Redis）。
+ *
+ * <p>isCompleted 必须仅在 key 值为 {@code success} 时返回 true：processing / 无记录 一律 false，
+ * 否则消费者崩溃后残留的 processing 标记会被 driver 当成"已完成"而 ACK，造成静默丢消息。
+ */
+class RedisMqIdempotentStoreTest {
+
+    @Test
+    void isCompleted_trueOnlyForSuccessValue() {
+        StringRedisTemplate redis = mock(StringRedisTemplate.class);
+        @SuppressWarnings("unchecked")
+        ValueOperations<String, String> ops = mock(ValueOperations.class);
+        when(redis.opsForValue()).thenReturn(ops);
+        RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300);
+
+        when(ops.get("mq:idem:kSucc")).thenReturn("success");
+        when(ops.get("mq:idem:kProc")).thenReturn("processing");
+        when(ops.get("mq:idem:kNone")).thenReturn(null);
+
+        assertTrue(store.isCompleted("kSucc"), "success 应视为已完成，driver 可 ACK");
+        assertFalse(store.isCompleted("kProc"), "processing 不可视为完成，否则崩溃残留会被 ACK 丢消息");
+        assertFalse(store.isCompleted("kNone"), "无记录(标记已过期)不可视为完成");
+    }
+}
--- a/mq-starter-redis-stream/pom.xml
+++ b/mq-starter-redis-stream/pom.xml
@@ -5,6 +5,7 @@
    <artifactId>mq-starter-redis-stream</artifactId>
    <dependencies>
        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-idempotent-redis</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
        <dependency><groupId>com.njcn</groupId><artifactId>redis-stream-springboot-starter</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-autoconfigure</artifactId><version>${autoconfigure.version}</version></dependency>
        <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
--- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java
+++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java
@@ -6,6 +6,7 @@ import com.njcn.middle.stream.constant.StreamMessageConstant;
 import com.njcn.middle.stream.support.StreamKeyBuilder;
 import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqMessageInProgressException;
 import com.njcn.mq.spi.MqSubscription;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.stream.*;
@@ -28,7 +29,9 @@ public class RedisStreamMqDriver implements MqDriver {
    private volatile boolean running = true;

    public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
-        this(redis, keyBuilder, props, 5000L);
+        // reclaim 默认 idle 30s：须大于正常处理耗时，避免多实例误 reclaim 仍在处理(in-flight)的消息造成重复消费；
+        // 非幂等消费在多实例下为 at-least-once，可通过 4 参构造按业务处理时延调整。
+        this(redis, keyBuilder, props, 30_000L);
    }

    public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) {
@@ -96,6 +99,11 @@ public class RedisStreamMqDriver implements MqDriver {
        try {
            sub.getListener().onMessage(msg);
            return true;
+        } catch (MqMessageInProgressException inProgress) {
+            // 去重标记仍 processing（并发处理中/上次崩溃残留）：不 ACK，留 PEL，
+            // 待 processing 标记过期后由 reclaimPending 重新 acquire 并真正重投。
+            log.debug("[mq-redis] in-progress, leave 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);
            return false; // 留 PEL，由 reclaimPending 重投或落库
@@ -132,8 +140,7 @@ public class RedisStreamMqDriver implements MqDriver {
            if (deliveryCount > maxRetry) {
                if (sub.getRetryExhaustedHandler() != null) {
                    try {
-                        sub.getRetryExhaustedHandler().accept(MqMessage.of(
-                                fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), null));
+                        sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields));
                    } catch (Exception ex) {
                        log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex);
                    }
@@ -147,6 +154,18 @@ public class RedisStreamMqDriver implements MqDriver {
        }
    }

+    /** 解码消息体为业务 payload（best-effort：毒消息时 payload 为 null），用于重试耗尽落库回调。 */
+    private MqMessage buildMessage(MqSubscription sub, Map<String, String> fields) {
+        Object payload = null;
+        try {
+            String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
+            payload = JSON.parseObject(json, sub.getPayloadType());
+        } catch (Exception ignore) {
+            // 毒消息：payload 留 null，至少保留 key/tag 供落库
+        }
+        return MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
+    }
+
    private void ensureGroup(String stream, String group) {
        try {
            redis.execute((RedisCallback<Object>) c -> c.execute("XGROUP",
--- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java
+++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java
@@ -3,11 +3,11 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
 import com.njcn.middle.stream.support.StreamKeyBuilder;
 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
 import com.njcn.mq.spi.MqDriver;
-import com.njcn.mq.spi.MqIdempotentStore;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.core.StringRedisTemplate;
@@ -16,17 +16,14 @@ import org.springframework.data.redis.core.StringRedisTemplate;
@AutoConfigureBefore(MqCoreAutoConfiguration.class)
@ConditionalOnClass(StringRedisTemplate.class)
@ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream")
+@EnableConfigurationProperties(RedisStreamMqProperties.class)
 public class RedisStreamMqDriverAutoConfiguration {
    @Bean
    @ConditionalOnMissingBean
-    public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) {
-        return new RedisStreamMqDriver(redis, streamKeyBuilder, props);
-    }
-
-    @Bean
-    @ConditionalOnMissingBean
-    public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis) {
-        // 默认 TTL：processing 60s / success 300s / fail 300s，对齐 legacy 量级；统一前缀（修正 key 一致性）。
-        return new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300);
+    public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder,
+                             RedisStreamProperties props, RedisStreamMqProperties mqProps) {
+        // reclaim idle 从 mq.redis-stream.reclaim-min-idle 读取（默认 30s）。
+        // 去重 store 不在这里装配:已中立化到 mq-starter-idempotent-redis(不按 mq.type 门控)。
+        return new RedisStreamMqDriver(redis, streamKeyBuilder, props, mqProps.getReclaimMinIdle().toMillis());
    }
 }
--- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java
+++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java
@@ -0,0 +1,29 @@
+package com.njcn.mq.driver.redis;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.time.Duration;
+
+/**
+ * Redis Stream 驱动运行参数(前缀 {@code mq.redis-stream})。
+ *
+ * <p>去重 TTL 配置已中立化到 {@code mq.idempotent.*}(mq-starter-idempotent-redis 模块),
+ * 旧键 {@code mq.redis-stream.idempotent.*} 仍向后兼容但会打弃用告警。
+ *
+ * <pre>
+ * mq:
+ *   redis-stream:
+ *     reclaim-min-idle: 30s        # 未 ACK 超过此时长才 reclaim 重投(须 &gt; 单条处理时延)
+ * </pre>
+ */
+@Data
+@ConfigurationProperties(prefix = "mq.redis-stream")
+public class RedisStreamMqProperties {
+
+    /**
+     * PEL reclaim 触发的最小 idle:消息投递后未 ACK 超过此时长才会被 reclaim 重投。
+     * 须大于单条业务的正常处理时延,避免仍在处理的消息被误 reclaim 造成重复消费。
+     */
+    private Duration reclaimMinIdle = Duration.ofSeconds(30);
+}
--- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamStartupProbeAutoConfiguration.java
+++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamStartupProbeAutoConfiguration.java
@@ -0,0 +1,69 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.mq.autoconfig.MqStartupChecks;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+
+/**
+ * ①② redis 探测：mq.type=redis-stream 时，在 bean 初始化阶段主动 PING 一次 Redis。
+ *
+ * <p>lettuce 默认懒连接（启动不真正连、运行时才连）；此处主动 PING，连不上就 fail-fast。
+ */
+@Configuration
+@ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream")
+public class RedisStreamStartupProbeAutoConfiguration implements InitializingBean {
+
+    private final RedisConnectionFactory connectionFactory;
+    private final Environment env;
+
+    public RedisStreamStartupProbeAutoConfiguration(RedisConnectionFactory connectionFactory, Environment env) {
+        this.connectionFactory = connectionFactory;
+        this.env = env;
+    }
+
+    @Override
+    public void afterPropertiesSet() {
+        if (!MqStartupChecks.enabled(env)) {
+            return;
+        }
+        try {
+            RedisConnection conn = connectionFactory.getConnection();
+            try {
+                conn.ping();
+            } finally {
+                conn.close(); // 探测用连接必须释放，避免泄漏
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(
+                    "MQ 启动失败：mq.type=redis-stream，无法连接 Redis（" + target() + "）。原因："
+                    + e.getMessage()
+                    + "。请检查 spring.data.redis.host/port/password，或确认 Redis 服务已启动。", e);
+        }
+    }
+
+    /** 尽力取 host:port 用于提示；取不到（非 Lettuce / cluster / sentinel）则回退到有意义的文案，不再抛新异常。 */
+    private String target() {
+        try {
+            if (connectionFactory instanceof LettuceConnectionFactory) {
+                LettuceConnectionFactory lf = (LettuceConnectionFactory) connectionFactory;
+                // cluster/sentinel 模式下 getHostName()/getPort() 返回的是 standalone 默认值 localhost:6379
+                //（使用者从未配置过的假地址），会把排障引偏，故回退到指向真实配置项的文案。
+                if (lf.isClusterAware()) {
+                    return "集群模式，见 spring.data.redis.cluster.nodes";
+                }
+                if (lf.isRedisSentinelAware()) {
+                    return "哨兵模式，见 spring.data.redis.sentinel";
+                }
+                return lf.getHostName() + ":" + lf.getPort();
+            }
+        } catch (Exception ignore) {
+            // 取地址失败不能影响主流程
+        }
+        return "unknown";
+    }
+}
--- a/mq-starter-redis-stream/src/main/resources/META-INF/spring.factories
+++ b/mq-starter-redis-stream/src/main/resources/META-INF/spring.factories
@@ -1,2 +1,3 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration
+com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration,\
+com.njcn.mq.driver.redis.RedisStreamStartupProbeAutoConfiguration
--- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java
+++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java
@@ -32,18 +32,20 @@ class RedisStreamMqDriverRetryIT {

        String topic = "MQ_RETRY_IT_" + System.nanoTime();
        AtomicInteger attempts = new AtomicInteger();
-        BlockingQueue<String> exhausted = new LinkedBlockingQueue<>();
+        BlockingQueue<MqMessage> exhausted = new LinkedBlockingQueue<>();
        driver.subscribe(MqSubscription.builder()
                .topic(topic).group("g1").tag("*").payloadType(String.class)
                .concurrency(1).batchSize(1).maxRetry(2)
                .listener(m -> { attempts.incrementAndGet(); throw new RuntimeException("always fail"); })
-                .retryExhaustedHandler(m -> exhausted.add(m.getKey()))
+                .retryExhaustedHandler(exhausted::add)
                .build());

        driver.send(topic, MqMessage.of("rk1", "", "data"));

-        String key = exhausted.poll(12, TimeUnit.SECONDS);
-        assertEquals("rk1", key, "重试耗尽应回调 retryExhaustedHandler");
+        MqMessage exhaustedMsg = exhausted.poll(12, TimeUnit.SECONDS);
+        assertNotNull(exhaustedMsg, "重试耗尽应回调 retryExhaustedHandler");
+        assertEquals("rk1", exhaustedMsg.getKey());
+        assertEquals("data", exhaustedMsg.getPayload(), "耗尽回调应携带完整 payload，而非 null");
        assertTrue(attempts.get() >= 2, "应实际重试若干次后才耗尽, attempts=" + attempts.get());

        Thread.sleep(500);
--- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java
+++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java
@@ -1,15 +1,16 @@
 package com.njcn.mq.driver.redis;

 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
+import com.njcn.mq.idempotent.redis.MqIdempotentRedisAutoConfiguration;
 import com.njcn.mq.spi.MqIdempotentStore;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
-import org.springframework.context.annotation.Bean;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.data.redis.core.StringRedisTemplate;

-import java.lang.reflect.Method;
-
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;

 /**
 * 装配顺序回归护栏（mq-starter 自身）。
@@ -37,14 +38,12 @@ class RedisStreamMqDriverWiringTest {
    }

    @Test
-    void declaresIdempotentStoreBean() throws Exception {
-        Method m = RedisStreamMqDriverAutoConfiguration.class
-                .getDeclaredMethod("mqIdempotentStore", StringRedisTemplate.class);
-        assertThat(m.getAnnotation(Bean.class))
-                .as("redis-stream 模式应提供默认去重实现")
-                .isNotNull();
-        assertThat(MqIdempotentStore.class)
-                .as("默认去重实现须为 MqIdempotentStore 类型")
-                .isAssignableFrom(m.getReturnType());
+    void idempotentStoreStillOnClasspathViaIdempotentRedisModule() {
+        // store 已迁到 mq-starter-idempotent-redis(pom 传递依赖):只引 redis-stream 单模块仍开箱带去重。
+        // 若有人删掉该 pom 依赖,本测试编译即失败——这正是护栏目的。
+        new ApplicationContextRunner()
+                .withConfiguration(AutoConfigurations.of(MqIdempotentRedisAutoConfiguration.class))
+                .withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class))
+                .run(ctx -> assertThat(ctx).hasSingleBean(MqIdempotentStore.class));
    }
 }
--- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java
+++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java
@@ -0,0 +1,41 @@
+package com.njcn.mq.driver.redis;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.boot.context.properties.source.ConfigurationPropertySources;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.MutablePropertySources;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * {@link RedisStreamMqProperties} 默认值与绑定行为单测(纯 Binder，不 refresh context / 不连 Redis）。
+ *
+ * <p>守护两点:① 默认值与历史硬编码一致(不改变现有行为);② mq.redis-stream.* 前缀能覆盖默认。
+ * 去重 TTL 配置已中立化到 mq.idempotent.*,由 mq-starter-idempotent-redis 模块守护。
+ */
+class RedisStreamMqPropertiesTest {
+
+    @Test
+    void defaults_matchLegacyHardcodedValues() {
+        RedisStreamMqProperties p = new RedisStreamMqProperties();
+        assertEquals(Duration.ofSeconds(30), p.getReclaimMinIdle(), "reclaim idle 默认应保持 30s");
+    }
+
+    @Test
+    void binds_mqRedisStreamPrefix_overridesDefaults() {
+        Map<String, Object> src = new HashMap<>();
+        src.put("mq.redis-stream.reclaim-min-idle", "90s");
+        MutablePropertySources ps = new MutablePropertySources();
+        ps.addFirst(new MapPropertySource("test", src));
+
+        RedisStreamMqProperties bound = new Binder(ConfigurationPropertySources.from(ps))
+                .bind("mq.redis-stream", RedisStreamMqProperties.class).get();
+
+        assertEquals(Duration.ofSeconds(90), bound.getReclaimMinIdle());
+    }
+}
--- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamStartupProbeAutoConfigurationTest.java
+++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamStartupProbeAutoConfigurationTest.java
@@ -0,0 +1,100 @@
+package com.njcn.mq.driver.redis;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.data.redis.connection.RedisClusterConfiguration;
+import org.springframework.data.redis.connection.RedisConnection;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class RedisStreamStartupProbeAutoConfigurationTest {
+
+    private final ApplicationContextRunner runner = new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(RedisStreamStartupProbeAutoConfiguration.class));
+
+    @Test
+    void pingOk_startsUp_andReleasesConnection() {
+        RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
+        RedisConnection conn = mock(RedisConnection.class);
+        when(factory.getConnection()).thenReturn(conn);
+        when(conn.ping()).thenReturn("PONG");
+
+        runner.withPropertyValues("mq.type=redis-stream")
+                .withBean(RedisConnectionFactory.class, () -> factory)
+                .run(ctx -> assertThat(ctx).hasNotFailed());
+
+        verify(conn).close();
+    }
+
+    @Test
+    void pingFails_startupFails() {
+        RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
+        when(factory.getConnection()).thenThrow(new RuntimeException("Connection refused"));
+
+        runner.withPropertyValues("mq.type=redis-stream")
+                .withBean(RedisConnectionFactory.class, () -> factory)
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("Redis");
+                    assertThat(ctx).getFailure().hasMessageContaining("Connection refused");
+                });
+    }
+
+    @Test
+    void pingFails_messageContainsHostPort() {
+        // 用真实 LettuceConnectionFactory(standalone) 覆盖 target() 的 host:port 提示分支；
+        // override getConnection() 让探测立即失败，无需真实 Redis / 网络。
+        LettuceConnectionFactory factory =
+                new LettuceConnectionFactory(new RedisStandaloneConfiguration("myhost", 1234)) {
+                    @Override
+                    public RedisConnection getConnection() {
+                        throw new RuntimeException("Connection refused");
+                    }
+                };
+
+        runner.withPropertyValues("mq.type=redis-stream")
+                .withBean(RedisConnectionFactory.class, () -> factory)
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("myhost:1234");
+                });
+    }
+
+    @Test
+    void clusterMode_avoidsFakeLocalhostInMessage() {
+        // cluster 模式下 getHostName()/getPort() 会返回 standalone 默认值 localhost:6379(假地址)；
+        // 修复后应回退到指向 cluster 配置项的文案，而不是打印使用者从未配过的 localhost:6379。
+        LettuceConnectionFactory factory =
+                new LettuceConnectionFactory(
+                        new RedisClusterConfiguration(java.util.Collections.singletonList("127.0.0.1:7000"))) {
+                    @Override
+                    public RedisConnection getConnection() {
+                        throw new RuntimeException("Connection refused");
+                    }
+                };
+
+        runner.withPropertyValues("mq.type=redis-stream")
+                .withBean(RedisConnectionFactory.class, () -> factory)
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("集群模式");
+                });
+    }
+
+    @Test
+    void disabled_skipsProbe() {
+        RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
+        when(factory.getConnection()).thenThrow(new RuntimeException("should not be called"));
+
+        runner.withPropertyValues("mq.type=redis-stream", "mq.startup-check.enabled=false")
+                .withBean(RedisConnectionFactory.class, () -> factory)
+                .run(ctx -> assertThat(ctx).hasNotFailed());
+    }
+}
--- a/mq-starter-rocketmq/pom.xml
+++ b/mq-starter-rocketmq/pom.xml
@@ -3,6 +3,17 @@
    <modelVersion>4.0.0</modelVersion>
    <parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
    <artifactId>mq-starter-rocketmq</artifactId>
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>2.3.12.RELEASE</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
    <dependencies>
        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
        <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version></dependency>
--- a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java
+++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java
@@ -3,6 +3,7 @@ package com.njcn.mq.driver.rocketmq;
 import com.alibaba.fastjson.JSON;
 import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqMessageInProgressException;
 import com.njcn.mq.spi.MqSubscription;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
@@ -16,6 +17,7 @@ import org.springframework.messaging.support.MessageBuilder;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;

@Slf4j
 public class RocketMqDriver implements MqDriver {
@@ -41,23 +43,14 @@ public class RocketMqDriver implements MqDriver {
    @Override
    public void subscribe(MqSubscription sub) {
        try {
+            int maxRetry = sub.getMaxRetry() > 0 ? sub.getMaxRetry() : 3;
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(sub.getGroup());
            consumer.setNamesrvAddr(namesrvAddr);
+            consumer.setMaxReconsumeTimes(maxRetry); // 对齐 @MqListener(maxRetry),而非用 broker 默认 16 次
            String tag = sub.getTag();
            consumer.subscribe(sub.getTopic(), (tag == null || tag.isEmpty()) ? "*" : tag);
-            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
-                for (MessageExt ext : msgs) {
-                    try {
-                        String json = new String(ext.getBody(), StandardCharsets.UTF_8);
-                        Object payload = JSON.parseObject(json, sub.getPayloadType());
-                        sub.getListener().onMessage(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
-                    } catch (Exception e) {
-                        log.warn("[mq-rocket] handle failed msgId={}", ext.getMsgId(), e);
-                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
-                    }
-                }
-                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-            });
+            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) ->
+                    consumeBatch(sub, msgs, maxRetry));
            consumer.start();
            consumers.add(consumer);
        } catch (Exception e) {
@@ -65,6 +58,49 @@ public class RocketMqDriver implements MqDriver {
        }
    }

+    ConsumeConcurrentlyStatus consumeBatch(MqSubscription sub, List<MessageExt> msgs, int maxRetry) {
+        for (MessageExt ext : msgs) {
+            try {
+                String json = new String(ext.getBody(), StandardCharsets.UTF_8);
+                Object payload = JSON.parseObject(json, sub.getPayloadType());
+                sub.getListener().onMessage(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
+            } catch (MqMessageInProgressException inProgress) {
+                // 去重"处理中"信号:不 ACK,交 broker 重投;绝不进耗尽落库分支(否则并发下静默丢消息)。
+                // 语义对齐 RedisStreamMqDriver#dispatch 对该异常的 return false(留 PEL)。
+                log.debug("[mq-rocket] in-progress, reconsume later key={}", ext.getKeys());
+                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+            } catch (Exception e) {
+                if (ext.getReconsumeTimes() >= maxRetry) {
+                    // 重试耗尽:回调 retryExhaustedHandler 落库并 ACK 终止,不再无限重投
+                    //（对齐 redis-stream reclaim 耗尽语义,兑现"换 MQ 业务不动"契约)。
+                    onRetryExhausted(sub, ext, e);
+                } else {
+                    log.warn("[mq-rocket] handle failed msgId={} reconsume={}", ext.getMsgId(), ext.getReconsumeTimes(), e);
+                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+                }
+            }
+        }
+        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+    }
+
+    /** 重试耗尽落库(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);
+        Consumer<MqMessage> handler = sub.getRetryExhaustedHandler();
+        if (handler == null) return;
+        Object payload = null;
+        try {
+            payload = JSON.parseObject(new String(ext.getBody(), StandardCharsets.UTF_8), sub.getPayloadType());
+        } catch (Exception ignore) {
+            // 毒消息:payload 留 null，至少保留 key/tag 供落库(对齐 redis buildMessage)
+        }
+        try {
+            handler.accept(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
+        } catch (Exception ex) {
+            log.error("[mq-rocket] retryExhausted handler error msgId={}（不影响主流程)", ext.getMsgId(), ex);
+        }
+    }
+
    @Override
    public void stop() {
        for (DefaultMQPushConsumer c : consumers) c.shutdown();
--- a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java
+++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java
@@ -0,0 +1,85 @@
+package com.njcn.mq.driver.rocketmq;
+
+import com.njcn.mq.autoconfig.MqStartupChecks;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.util.StringUtils;
+
+import java.net.InetSocketAddress;
+import java.net.Socket;
+
+/**
+ * ①② rocketmq 探测：mq.type=rocketmq 时，对 NameServer 地址做 TCP 短超时连接探测。
+ *
+ * <p>不用 DefaultMQProducer.start()（它不主动连 NameServer、探不准且更慢）；
+ * 改用 Socket.connect + 可配超时：任一地址可达即通过，全不通则 fail-fast。
+ */
+@Configuration
+@ConditionalOnProperty(name = "mq.type", havingValue = "rocketmq")
+public class RocketMqStartupProbeAutoConfiguration implements InitializingBean {
+
+    private final Environment env;
+
+    public RocketMqStartupProbeAutoConfiguration(Environment env) {
+        this.env = env;
+    }
+
+    @Override
+    public void afterPropertiesSet() {
+        if (!MqStartupChecks.enabled(env)) {
+            return;
+        }
+        // 与 driver 同源取值：driver 走 RocketMQProperties(@ConfigurationProperties)的 getNameServer()，
+        // 享受 Spring relaxed binding；这里也用 Binder，避免 env.getProperty 精确键查找漏掉
+        // camelCase(rocketmq.nameServer)等 relaxed 变体，从而把「其实配好了」误判成「未配置」而错误 fail-fast。
+        String nameServer = Binder.get(env).bind("rocketmq.name-server", String.class).orElse(null);
+        if (!StringUtils.hasText(nameServer)) {
+            throw new IllegalStateException(
+                    "MQ 启动失败：mq.type=rocketmq，但未配置 rocketmq.name-server。"
+                    + "请配置 NameServer 地址（如 127.0.0.1:9876）。");
+        }
+
+        int timeoutMs = env.getProperty("mq.startup-check.rocketmq.timeout-ms", Integer.class, 1000);
+        Exception last = null;
+        for (String raw : nameServer.split("[;,]")) {
+            String addr = raw.trim();
+            if (addr.isEmpty()) {
+                continue;
+            }
+            int idx = addr.lastIndexOf(':');
+            if (idx < 0) {
+                last = new IllegalArgumentException("地址缺少端口: " + addr);
+                continue;
+            }
+            String host = addr.substring(0, idx);
+            int port;
+            try {
+                port = Integer.parseInt(addr.substring(idx + 1).trim());
+            } catch (NumberFormatException e) {
+                last = e;
+                continue;
+            }
+            Socket socket = new Socket();
+            try {
+                socket.connect(new InetSocketAddress(host, port), timeoutMs);
+                return; // 任一地址可达即通过
+            } catch (Exception e) {
+                last = e;
+            } finally {
+                try {
+                    socket.close();
+                } catch (Exception ignore) {
+                    // 关闭探测 socket 失败可忽略
+                }
+            }
+        }
+
+        throw new IllegalStateException(
+                "MQ 启动失败：mq.type=rocketmq，无法连接 NameServer（" + nameServer + "）。原因："
+                + (last == null ? "未知" : last.getMessage())
+                + "。请检查 rocketmq.name-server，或确认 NameServer 已启动。", last);
+    }
+}
--- a/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories
+++ b/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories
@@ -1,2 +1,3 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration
+com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration,\
+com.njcn.mq.driver.rocketmq.RocketMqStartupProbeAutoConfiguration
--- a/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java
+++ b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverConsumeTest.java
@@ -0,0 +1,123 @@
+package com.njcn.mq.driver.rocketmq;
+
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqMessageInProgressException;
+import com.njcn.mq.spi.MqSubscription;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+
+/**
+ * RocketMQ 消费处理逻辑单测(纯 POJO，不启动真实 consumer / 不连 namesrv）。
+ *
+ * <p>守护跨 driver 契约一致性:与 redis-stream 的 reclaim 耗尽语义对齐 ——
+ * 重试次数达到 maxRetry 时回调 retryExhaustedHandler 落库并 ACK 终止;未达到则 RECONSUME_LATER 交 broker 重投。
+ */
+class RocketMqDriverConsumeTest {
+
+    private RocketMqDriver newDriver() {
+        return new RocketMqDriver(mock(RocketMQTemplate.class), "localhost:9876");
+    }
+
+    private static MessageExt msg(String key, String body, int reconsumeTimes) {
+        MessageExt ext = new MessageExt();
+        ext.setKeys(key);
+        ext.setBody(body.getBytes(StandardCharsets.UTF_8));
+        ext.setReconsumeTimes(reconsumeTimes);
+        return ext;
+    }
+
+    @Test
+    void consume_retryExhausted_callsHandlerAndAcks() {
+        List<MqMessage> landed = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> { throw new RuntimeException("biz boom"); })
+                .retryExhaustedHandler(landed::add)
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 3)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status,
+                "重试耗尽应 ACK 终止,不再无限重投");
+        assertEquals(1, landed.size(), "重试耗尽应回调 retryExhaustedHandler 落库(对齐 redis)");
+        assertEquals("k1", landed.get(0).getKey(), "落库消息应保留原 key");
+    }
+
+    @Test
+    void consume_notExhausted_reconsumesLaterWithoutLanding() {
+        List<MqMessage> landed = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> { throw new RuntimeException("biz boom"); })
+                .retryExhaustedHandler(landed::add)
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 1)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status,
+                "未达 maxRetry 应交 broker 重投");
+        assertEquals(0, landed.size(), "未耗尽不应落库");
+    }
+
+    @Test
+    void consume_success_deliversPayloadAndAcks() {
+        List<Object> received = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> received.add(m.getPayload()))
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hi\"", 0)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.CONSUME_SUCCESS, status);
+        assertEquals(Collections.singletonList("hi"), received, "成功路径应把 payload 投给业务");
+    }
+
+    @Test
+    void consume_inProgress_reconsumesLaterWithoutLandingEvenIfExhausted() {
+        List<MqMessage> landed = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> { throw new MqMessageInProgressException("k1"); })
+                .retryExhaustedHandler(landed::add)
+                .build();
+
+        // reconsumeTimes=3 >= maxRetry=3:普通异常会走"耗尽落库+ACK",而 InProgress 是控制流信号,
+        // 必须仍 RECONSUME_LATER 且不落库——否则并发中的消息被当耗尽 ACK,静默丢消息。
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 3)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status,
+                "处理中信号应保留消息交 broker 重投,绝不 ACK");
+        assertEquals(0, landed.size(), "处理中信号不是重试耗尽,不得回调 retryExhaustedHandler 落库");
+    }
+
+    @Test
+    void consume_inProgress_notExhausted_alsoReconsumesLater() {
+        List<MqMessage> landed = new ArrayList<>();
+        MqSubscription sub = MqSubscription.builder()
+                .topic("t").group("g").payloadType(String.class).maxRetry(3)
+                .listener(m -> { throw new MqMessageInProgressException("k1"); })
+                .retryExhaustedHandler(landed::add)
+                .build();
+
+        ConsumeConcurrentlyStatus status =
+                newDriver().consumeBatch(sub, Collections.singletonList(msg("k1", "\"hello\"", 0)), 3);
+
+        assertEquals(ConsumeConcurrentlyStatus.RECONSUME_LATER, status);
+        assertEquals(0, landed.size());
+    }
+}
--- a/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java
+++ b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java
@@ -0,0 +1,70 @@
+package com.njcn.mq.driver.rocketmq;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+
+import java.net.ServerSocket;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RocketMqStartupProbeAutoConfigurationTest {
+
+    private final ApplicationContextRunner runner = new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(RocketMqStartupProbeAutoConfiguration.class));
+
+    @Test
+    void reachableNameServer_startsUp() throws Exception {
+        // 起一个真实监听端口模拟"可达的 NameServer"，不依赖真 rocketmq
+        try (ServerSocket server = new ServerSocket(0)) {
+            int port = server.getLocalPort();
+            runner.withPropertyValues(
+                            "mq.type=rocketmq",
+                            "rocketmq.name-server=127.0.0.1:" + port)
+                    .run(ctx -> assertThat(ctx).hasNotFailed());
+        }
+    }
+
+    @Test
+    void reachableNameServer_camelCaseKey_startsUp() throws Exception {
+        // driver 通过 RocketMQProperties 走 relaxed binding，camelCase 的 rocketmq.nameServer 也能生效；
+        // 探测取值必须与之同源，否则会把配好的 NameServer 误判为未配置而错误 fail-fast。
+        try (ServerSocket server = new ServerSocket(0)) {
+            int port = server.getLocalPort();
+            runner.withPropertyValues(
+                            "mq.type=rocketmq",
+                            "rocketmq.nameServer=127.0.0.1:" + port)
+                    .run(ctx -> assertThat(ctx).hasNotFailed());
+        }
+    }
+
+    @Test
+    void unreachableNameServer_startupFails() {
+        runner.withPropertyValues(
+                        "mq.type=rocketmq",
+                        "rocketmq.name-server=127.0.0.1:1",
+                        "mq.startup-check.rocketmq.timeout-ms=300")
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("127.0.0.1:1");
+                });
+    }
+
+    @Test
+    void missingNameServer_startupFails() {
+        runner.withPropertyValues("mq.type=rocketmq")
+                .run(ctx -> {
+                    assertThat(ctx).hasFailed();
+                    assertThat(ctx).getFailure().hasMessageContaining("name-server");
+                });
+    }
+
+    @Test
+    void disabled_skipsProbe() {
+        runner.withPropertyValues(
+                        "mq.type=rocketmq",
+                        "rocketmq.name-server=127.0.0.1:1",
+                        "mq.startup-check.enabled=false")
+                .run(ctx -> assertThat(ctx).hasNotFailed());
+    }
+}
--- a/pom.xml
+++ b/pom.xml
@@ -8,6 +8,7 @@
    <name>mq-starter 统一 MQ 抽象层</name>
    <modules>
        <module>mq-starter-core</module>
+        <module>mq-starter-idempotent-redis</module>
        <module>mq-starter-rocketmq</module>
        <module>mq-starter-redis-stream</module>
        <module>mq-spring-boot-starter</module>
Author	SHA1	Message	Date
hongawen	0f7be86bb8	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>	2026-07-14 04:19:36 +01:00
hongawen	8c39399d09	feat(mq-starter-core): @MqListener batchSize 与首参 List 错配时打 WARN 软校验走不走批量始终只由首参是否 List 决定，错配不阻断启动： - batchSize>1 但首参非 List → WARN 提示 batchSize 不生效、按单条逐一回调 - 首参 List 但 batchSize<=1 → WARN 提示将按 1 条一批回调测试：ListAppender 捕获日志，覆盖两个错配方向 + 匹配时无 WARN。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-14 04:19:17 +01:00
hongawen	14502449aa	docs(README): 承接全景手册上手教学——手册删教学 tab 后 README 成为唯一上手入口 - 发送语义:fastjson 序列化约束(无参构造+getter/setter)、send 同步/失败抛异常、反序列化失败两端行为(redis-stream ACK 丢弃/rocketmq 重试到耗尽) - 批量的坑:裸 List 启动报错、无 idle flush、整批失败被吞须配 errorHandler、可加 MqContext 第二参 - 新增「排错 FAQ」章节(8 问,自手册 FAQ 改写) - 配置表小补:fail-ttl 语义、redis-stream.* 底层配置来源 - 架构章节指向手册的描述改为架构图文/能力矩阵/Roadmap(配置速查与 FAQ 不再在手册) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-14 11:14:30 +08:00
hongawen	5191f5ace9	docs(README): batchSize 能力格对齐全景手册——两端一致(core 内存攒批/no idle flush/重启丢未满批),纠正 redis 侧 ✅ 伪差异 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-10 15:33:53 +08:00
hongawen	3c14b0cca3	docs(README): SINGLE 失败落库口径对齐全景手册——两端一致(单条落库+重投/批量失败被吞),纠正 rocketmq 记错的 RECONSUME_LATER 伪差异 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-10 15:12:02 +08:00
hongawen	96e26c9bbb	docs: README/CLAUDE.md 对齐 rocketmq 消费去重落地——能力表 ❌→✅、新模块 mq-starter-idempotent-redis、mq.idempotent.* 配置、纠正序列化过时表述 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 13:18:09 +08:00
hongawen	e2a956edb1	feat(mq-spring-boot-starter): 聚合 jar 引入 mq-starter-idempotent-redis，rocketmq 部署开箱可用消费去重 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 11:58:40 +08:00
hongawen	e6ea14046c	refactor(mq-starter-redis-stream): 去重 store 与 idempotent 配置迁至 mq-starter-idempotent-redis，模块改为依赖之 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 11:50:37 +08:00
hongawen	b449c26356	feat(mq-starter-idempotent-redis): 旧键 mq.redis-stream.idempotent.* Binder 逐键回退兼容 + 聚合弃用告警 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 11:34:21 +08:00
hongawen	f558f6303e	feat(mq-starter-idempotent-redis): mq.idempotent.* 中立配置 + 不按 mq.type 门控的去重 store 自动装配 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 03:12:06 +01:00
hongawen	c2a31c9339	feat(mq-starter-idempotent-redis): 新建 driver 中立去重模块，迁入 RedisMqIdempotentStore 及其测试 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 10:04:10 +08:00
hongawen	f4793b10c0	fix(mq-starter-rocketmq): 消费捕获 MqMessageInProgressException 返回 RECONSUME_LATER，杜绝处理中消息被耗尽落库+ACK 静默丢弃 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>	2026-07-10 02:54:09 +01:00
hongawen	ae0caafa63	docs(mq-starter): 重写 README + 新增 CLAUDE.md，gitignore docs/ - README 重写为面向使用者的完整说明（快速开始/配置项/两 driver 能力对照/架构/TODO） - 新增 CLAUDE.md：补代码结构、装配链路、消费 ACK 语义等踩坑点，供 Claude Code 参考 - .gitignore 忽略 docs/（AI 生成的本地参考/教学资料，不纳入版本库） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-09 20:41:54 +08:00
hongawen	29473e6fef	feat(mq-starter-rocketmq): 重试耗尽回调 retryExhaustedHandler 落库并 ACK，maxRetry 对齐 @MqListener 对齐 redis-stream 的 reclaim 耗尽语义，兑现"换 MQ 业务不动"契约：此前 rocketmq 只有 broker 默认 16 次重投、无 RETRY_EXHAUSTED 落库。 - setMaxReconsumeTimes 对齐 @MqListener(maxRetry)（缺省 3），不再用 broker 默认 16 - 抽出 consumeBatch：reconsumeTimes 达 maxRetry 时回调 retryExhaustedHandler 落库并 ACK 终止；未达则 RECONSUME_LATER - onRetryExhausted 落库 best-effort：毒消息 payload 反序列化失败时留 null、至少保留 key/tag，并隔离 handler 自身异常 - 补消费逻辑单测（耗尽落库 / 未耗尽重投 / 成功投递） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-09 20:41:39 +08:00
hongawen	84034fb7d0	feat(mq-starter-redis-stream): 去重 TTL / reclaim idle 外置为 mq.redis-stream.* 可配置原先 reclaim idle 与去重 TTL/前缀硬编码在 autoconfig 里，按业务处理时延调优只能改源码。 - 新增 RedisStreamMqProperties(前缀 mq.redis-stream)：reclaim-min-idle + idempotent.{key-prefix,processing/success/fail-ttl} - 默认值与历史硬编码完全一致（reclaim 30s / processing 60s / success 300s / fail 300s / mq:idem:），不改变现有行为 - autoconfig 改由 Properties 取值装配 driver 与 RedisMqIdempotentStore - 补默认值一致性 + mq.redis-stream.* 绑定覆盖单测；wiring 测试改按方法名反射以适配新签名 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-09 20:41:21 +08:00
hongawen	e660afe2a2	fix(mq-starter): 去重跳过区分 processing/success，processing 残留不 ACK 避免静默丢消息 tryAcquire 返回 false 的两类跳过此前一律静默 return，导致 driver 无差别 ACK：消费者在 markSuccess/markFail 前崩溃残留的 processing 标记，会被当成"已处理" 而 ACK，该消息从未成功消费却被丢弃（Critical 静默丢消息）。 - MqIdempotentStore 新增 isCompleted(key)，默认 true 保持旧行为，状态机实现须覆写 - MqConsumeDispatcher：跳过时按 isCompleted 分流——success 正常 return 让 driver ACK；仍 processing 则抛 MqMessageInProgressException（新，纯控制流信号、不带栈）让 driver 不 ACK - RedisMqIdempotentStore.isCompleted 仅 success 视为完成；RedisStreamMqDriver 捕获信号异常时留 PEL 待重投 - 补 dispatcher 两类跳过 + store isCompleted 语义单测 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-09 20:41:02 +08:00
hongawen	886c02ff96	fix(mq-starter): 修复启动 fail-fast 校验 review 发现的缺陷（relaxed binding / 逃生开关 / redis 假地址） workflow 多智能体 review 后修复： - rocketmq 探测改用 Binder 读 rocketmq.name-server，与 driver（RocketMQProperties）同源、享受 relaxed binding，修复 camelCase(rocketmq.nameServer) 被误判为「未配置」而错误 fail-fast - MqStartupChecks 逃生开关只认严格 false（忽略大小写/空白），缺省或非法值一律继续校验，不再因非法布尔值（如 enabled=disabled）抛类型转换异常炸掉启动 - redis 探测 target() 对 cluster/sentinel 回退到指向真实配置项的文案，不再假报使用者从未配过的 localhost:6379 - 补齐回归测试：camelCase 键 / 非法开关值继续校验 / host:port 提示分支 / cluster 回退（core 5、redis-stream 5、rocketmq 5，全绿） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-09 14:06:05 +08:00
hongawen	0c6d37842e	fix(mq-starter-rocketmq): 导入 spring-boot BOM 统一 spring 版本（修既有 5.2.15↔5.3 裂开，Task3 测试可运行）	2026-07-09 11:20:24 +08:00
hongawen	d7ef8b5626	feat(mq-starter-rocketmq): 启动期 TCP 探测 NameServer 连通性（①② fail-fast）	2026-07-09 11:05:00 +08:00
hongawen	46b0c13462	feat(mq-starter-redis-stream): 启动期 PING 探测 Redis 连通性（①② fail-fast）	2026-07-09 10:49:23 +08:00
hongawen	5c4a29e1e1	feat(mq-starter-core): 启动期校验 mq.type 有对应 MqDriver（③ fail-fast） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-07-09 10:34:50 +08:00
hongawen	bd2e733d98	Merge: mq-starter 消费侧通用化 + redis-stream 可放量（去重/批量/异常落库/重试 reclaim + opus review 修复）	2026-06-29 20:51:59 +08:00
hongawen	f0dc785998	fix(mq-starter): 落实 opus review 反馈（含 1 个 Critical 静默丢消息） - [Critical] dispatcher errorHandler 抛异常时仍保证 markFail + 抛原始异常（safeOnError 隔离）：否则去重标记停在 processing，reclaim 重投被去重挡掉并 ACK → 静默丢消息 - [Important] 重试耗尽回调携带完整 payload（reclaim 解码 body，不再传 null） - [Important] reclaim 默认 idle 5s→30s，避免多实例误 reclaim in-flight 消息；注释 at-least-once 语义 - [Important] idempotent=true 但无 MqIdempotentStore bean 时 start() 告警 - core 引入 slf4j-api(provided)；批量失败 / errorHandler 失败均记日志 - errorHandler bean 未找到时抛带 topic/group 的清晰错误 - 文档化已知限制（批量无 idle flush、去重 fail→reopen 非原子） - 全仓 29 测试绿（含连真实 redis 的 IT） Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>	2026-06-29 20:48:53 +08:00