docs: README/CLAUDE.md 对齐 rocketmq 消费去重落地——能力表 、新模块 mq-starter-idempotent-redis、mq.idempotent.* 配置、纠正序列化过时表述

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
2026-07-10 13:18:09 +08:00
parent e2a956edb1
commit 96e26c9bbb
2 changed files with 41 additions and 28 deletions

View File

@@ -30,22 +30,26 @@ mvn -pl mq-starter-core test -Dtest=MqConsumeDispatcherTest#someMethod # 单个
## 模块结构
```
mq-starter-core 门面 + 注解扫描 + 消费链 + SPI 契约(中立,不依赖 Spring/Redis 的核心编排逻辑,可纯单测)
mq-starter-redis-stream Redis Stream driver功能完整去重/批量/落库/reclaim 重试都接上了
mq-starter-rocketmq RocketMQ driver基础收发 + 原生重试/死信,去重和 RETRY_EXHAUSTED 落库尚未接入
mq-spring-boot-starter 聚合 jar业务方只引这一个
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 有对照表)。改到跨 driver 行为、或使用者依赖去重/失败落库时,先确认走的是哪个 driver —— rocketmq 那侧很多能力配了也不生效
> ⚠️ **两个 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. **core 依赖 driver**`MqCoreAutoConfiguration``MqTemplate``MqListenerRegistry` 都是 `@ConditionalOnBean(MqDriver.class)` —— 有 driver 才有门面。driver 的 AutoConfiguration 用 `@AutoConfigureBefore(MqCoreAutoConfiguration.class)` 保证自己先装
3. **注解扫描**`MqListenerAnnotationBeanPostProcessor`(一个 `BeanPostProcessor`)扫描所有 bean 的方法,把 `@MqListener` 方法收集成 `MqListenerEndpoint` 列表
4. **启动订阅**`MqListenerRegistry` 实现 `SmartLifecycle``phase = Integer.MAX_VALUE - 100`,最后启动),在 `start()` 里为每个 endpoint 组装 `MqConsumeDispatcher`,包成 `MqSubscription` 交给 `driver.subscribe()`
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()`
## 核心执行流
@@ -61,13 +65,13 @@ mq-spring-boot-starter 聚合 jar业务方只引这一个
- **单条消费失败**:先落库 + `store.markFail()`(把去重标记从 processing 复位),再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processingreclaim 重投时被去重挡掉并 ACK → 静默丢消息。
- **批量消费**:无 idle flush攒够 `batchSize` 才 flush低流量下未满批会停滞进程重启丢内存里没入库的那批与 legacy 行为一致,适合高流量流)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库 + 吞掉不抛。
Redis driver 侧对应实现在 `RedisStreamMqDriver.java`捕获 `MqMessageInProgressException` 时不 `acknowledge()`
driver 侧对 `MqMessageInProgressException` 的对应处理:`RedisStreamMqDriver.java` 捕获后不 `acknowledge()`(留 PEL 待 reclaim`RocketMqDriver.java` 在通用 `catch(Exception)` **之前**单独捕获并返回 `RECONSUME_LATER`绝不进耗尽落库分支——catch 顺序不能动,它是 `Exception` 子类。边缘rocketmq 下 InProgress 反复重投超过 `maxReconsumeTimes` 会进原生 `%DLQ%`(进死信、非丢消息);缓解靠 `mq.idempotent.processing-ttl` < 业务重试节奏累计时长
## SPI 契约(`com.njcn.mq.spi`
driver 只需实现 `MqDriver``send` / `subscribe` / `stop`)。其余是可选扩展点由业务方或 driver 提供 bean
- **`MqIdempotentStore`** —— 去重状态机 `processing / success / fail`注意 `isCompleted(key)` 的默认实现返回 `true`向后兼容旧行为),**有真状态机的实现必须覆写**否则 processing 中的消息会被误判为已完成而被 ACK 丢掉。redis-stream 自带 `RedisMqIdempotentStore`
- **`MqIdempotentStore`** —— 去重状态机 `processing / success / fail`注意 `isCompleted(key)` 的默认实现返回 `true`向后兼容旧行为**有真状态机的实现必须覆写**否则 processing 中的消息会被误判为已完成而被 ACK 丢掉默认实现 `RedisMqIdempotentStore` `mq-starter-idempotent-redis` 模块 `com.njcn.mq.idempotent.redis`配置前缀 `mq.idempotent.*`TTL/前缀旧键 `mq.redis-stream.idempotent.*` Binder 逐键回退向后兼容用到会打一条聚合弃用 WARN)。用户自带 store bean 会让自动装配让位`@ConditionalOnMissingBean`)。
- **`MqConsumeErrorHandler`** —— 失败落库回调通过 `@MqListener(errorHandler="beanName")` bean 名解析`MqErrorIdentity` 区分 `SINGLE`单次失败 `RETRY_EXHAUSTED`重试耗尽 driver 在超过 `maxRetry` 时回调 `MqSubscription.retryExhaustedHandler`)。
- **`MqSubscription`** —— core 传给 driver 的订阅描述topic/group/tag/payloadType/concurrency/batchSize/maxRetry/listener/retryExhaustedHandler)。
@@ -92,4 +96,4 @@ driver 只需实现 `MqDriver``send` / `subscribe` / `stop`)。其余是可
- `pom.xml` 统一管理版本Spring Boot 2.3.12fastjson 1.2.83hutoollombok)。
- `mq-starter-rocketmq` 额外 `import` `spring-boot-dependencies` BOM —— 用于收敛 `rocketmq-spring-boot-starter` 传递进来的 Spring 版本曾出现 5.2.15 5.3 裂开导致测试跑不起来)。 rocketmq 模块依赖时注意别破坏这个收敛
- **序列化两边不一致**redis-stream 用 Jacksonrocketmq 用 fastjson。跨 driver 的序列化行为不要假设一致。
- **序列化两端都是 fastjson**`RocketMqDriver` `RedisStreamMqDriver` 均直接调 `JSON.toJSONString/parseObject`)。历史文档曾说"redis-stream Jackson"——那是错的Jackson 只出现在无关的上游 middleware不在 driver 路径但序列化目前**硬编码不可插拔**替换需改两个 driver