diff --git a/CLAUDE.md b/CLAUDE.md index bd3dcab..583dcc0 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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` 会让标记卡在 processing,reclaim 重投时被去重挡掉并 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.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 模块依赖时注意别破坏这个收敛。 -- **序列化两边不一致**:redis-stream 用 Jackson,rocketmq 用 fastjson。跨 driver 的序列化行为不要假设一致。 +- **序列化两端都是 fastjson**(`RocketMqDriver` 与 `RedisStreamMqDriver` 均直接调 `JSON.toJSONString/parseObject`)。历史文档曾说"redis-stream 用 Jackson"——那是错的,Jackson 只出现在无关的上游 middleware,不在 driver 路径。但序列化目前**硬编码不可插拔**,替换需改两个 driver。 diff --git a/README.md b/README.md index ff718a3..3d6162e 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,7 @@ public class FrontDataConsumer { 去重、批量、失败落库都是 `@MqListener` 上的属性,业务方法本身不用改: ```java -// 去重:重复消息不会再进业务(需要 MqIdempotentStore,redis-stream 自带了一个) +// 去重:重复消息不会再进业务(去重状态存 Redis,两个 driver 都支持,见下方说明) @MqListener(topic = "LN_Topic", group = "g", idempotent = true) public void onData(FrontData msg) { } @@ -139,22 +139,22 @@ public class MyErrorHandler implements MqConsumeErrorHandler { ## 两个 driver 不太一样 -这点得说在前面:两套实现的完整度并不一致。Redis Stream 这套是照着 core 的设计意图做全的,去重、批量、失败落库、reclaim 重试都接上了;RocketMQ 那套目前只做了基础收发加原生的重试和死信,去重和 `RETRY_EXHAUSTED` 落库还没接进去。 +两套实现的能力大头已经对齐:去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义: | 能力 | Redis Stream | RocketMQ | |---|:---:|:---:| | 收发 / 并发消费 | ✅ | ✅ | -| 去重 `idempotent` | ✅ 内置实现 | ❌ 还没接,配了也不生效 | -| 失败落库 `RETRY_EXHAUSTED` | ✅ | ❌ 只会进原生 DLQ | +| 去重 `idempotent` | ✅(需 Redis) | ✅(需 Redis) | +| 失败落库 `RETRY_EXHAUSTED` | ✅ | ✅ 落库并 ACK | | 失败落库 `SINGLE` | ✅ | ⚠️ 异常会被转成 RECONSUME_LATER | | 批量 `batchSize` | ✅ | ⚠️ 逐条回调,靠 core 内存攒批 | | 重试 | ✅ PEL reclaim | ✅ 原生重投 | | 死信队列 | ❌ 只落库 | ✅ 原生 `%DLQ%` | | 顺序消息 | ❌ | ❌ 用的 Concurrently | | 事务 / 延迟消息 | ❌ | ❌ 没暴露 | -| 序列化 | Jackson | fastjson | +| 序列化 | fastjson | fastjson | -所以现在如果你依赖去重或者失败落库,先用 Redis Stream。RocketMQ driver 能收能发,但还没「放量」,具体见下面的 TODO。 +去重的状态存在 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 没配好而影响启动。 ## 架构 @@ -173,14 +173,15 @@ public class MyErrorHandler implements MqConsumeErrorHandler { └─────────────────────┘ └─────────────────────┘ ``` -四个模块: +五个模块: | 模块 | 干什么 | |---|---| | `mq-starter-core` | 门面 `MqTemplate`、`@MqListener` 扫描注册、中立信封 `MqMessage`、消费处理链、几个 SPI,还有基于 `SmartLifecycle` 的启停 | -| `mq-starter-redis-stream` | Redis Stream 实现(XADD / 消费组 XREADGROUP / XACK / XPENDING+XCLAIM reclaim),另带一个默认的去重实现 | +| `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,业务方引这一个就够 | +| `mq-spring-boot-starter` | 聚合 jar,业务方引这一个就够(恒含 idempotent-redis) | core 是中立的,编排逻辑不依赖 Spring 和 Redis,可以直接单测。想看更细的图文(消费链、SPI 契约、配置速查、FAQ 都在里面),打开 [`docs/mq-starter-全景手册.html`](./docs/mq-starter-全景手册.html)。 @@ -198,15 +199,23 @@ core 是中立的,编排逻辑不依赖 Spring 和 Redis,可以直接单测 | `idempotent` | false | 是否去重 | | `errorHandler` | `""` | 失败落库的 bean 名 | -Redis Stream 这边的参数(前缀 `mq.redis-stream`,都有默认值): +去重参数(前缀 `mq.idempotent`,driver 中立,都有默认值): | 配置 | 默认 | 说明 | |---|---|---| -| `reclaim-min-idle` | 30s | 消息投出去多久没 ack 才 reclaim 重投,要大于单条处理耗时 | -| `idempotent.processing-ttl` | 60s | 「处理中」标记存多久,建议大于处理耗时、小于 reclaim-min-idle | -| `idempotent.success-ttl` | 300s | 「已完成」标记存多久(去重窗口) | -| `idempotent.fail-ttl` | 300s | 失败标记保留多久 | -| `idempotent.key-prefix` | `mq:idem:` | 去重键前缀 | +| `enabled` | true | 逃生开关,显式 false 关闭去重 store 装配 | +| `processing-ttl` | 60s | 「处理中」标记存多久,建议大于单条处理耗时;rocketmq 下还建议小于业务重试节奏的累计时长,否则并发中的消息可能反复重投直至进原生 `%DLQ%` | +| `success-ttl` | 300s | 「已完成」标记存多久(去重窗口) | +| `fail-ttl` | 300s | 失败标记保留多久 | +| `key-prefix` | `mq:idem:` | 去重键前缀 | + +> 旧键 `mq.redis-stream.idempotent.*` 仍向后兼容(会打一条弃用告警),中立键和旧键同时配置时以 `mq.idempotent.*` 为准。建议尽快迁移。 + +Redis Stream 这边的参数(前缀 `mq.redis-stream`): + +| 配置 | 默认 | 说明 | +|---|---|---| +| `reclaim-min-idle` | 30s | 消息投出去多久没 ack 才 reclaim 重投,要大于单条处理耗时(且建议大于 processing-ttl) | RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置。 @@ -217,12 +226,12 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置。 P0,放量到生产前想补的: - [x] 失败落库、去重、批量消费(core 已做) -- [ ] 给 rocketmq driver 补上去重和 `RETRY_EXHAUSTED` 落库,跟 redis-stream 对齐 +- [x] 给 rocketmq driver 补上去重和 `RETRY_EXHAUSTED` 落库,跟 redis-stream 对齐(去重 store 已抽成 driver 中立的 `mq-starter-idempotent-redis`) - [ ] 批量消费加 idle flush,现在低流量下攒不满会卡着,进程重启还会丢内存里没攒满的那批 - [ ] 顺序消息 - [ ] 事务消息 -P1:延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔(现在两边一个 Jackson 一个 fastjson)、广播消费。 +P1:延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔(两端都是 fastjson,但硬编码换不掉)、广播消费。 P2:并发度语义统一、headers 完整透传、`mq.type` 错配时 fail-fast、监控告警。