mq-starter/CLAUDE.md

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 命令在仓库根目录（多模块聚合）执行。

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 重试/tag 消费端过滤/订阅 stream 定时裁剪全接入；去重 store 依赖 idempotent-redis 模块。裁剪复用底层 redis-stream.trim.* 配置、默认开启，maxlen<=0 关，只裁订阅过的 stream——中间件自己的 trim 任务只认它的 handler，对 mq-starter 无效）
mq-starter-rocketmq         RocketMQ driver（收发 + 原生重试/死信 + RETRY_EXHAUSTED 落库 + 去重信号处理；去重 store 经聚合坐标间接获得）
mq-spring-boot-starter      聚合 jar，业务方只引这一个（恒含 idempotent-redis）

⚠️ 两个 driver 仍有差异（README 有对照表）：去重、失败落库、tag 过滤三者两端已对齐（2026-07 起 rocketmq idempotent=true 生效、需 Redis 可达；tag：rocketmq broker 原生 / redis-stream 消费端模拟，语义子集见 core 的 MqTagMatcher，dispatch 顶部与 reclaim 耗尽分支前两处调用）；剩余差距是死信模型（redis 只落库 / rocketmq 原生 %DLQ%）、批量整批失败的处理（逐条 ACK 后无法精确重投整批，故落库+吞掉）与多环境隔离（redis-stream 经底层 starter 的 redis-stream.env-isolation/env 拼 topic_env 后缀；rocketmq 无对应机制，切换后隔离静默消失）；真正两端都未兑现的只有 concurrency。批量攒批本身（含 idle flush 超时/停机强刷，2026-07 已落地）两端已对齐。改跨 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，所以该条件几乎恒真——真正的启动安全网只剩懒连接（该装配与 store 构造都刻意不 PING Redis，与下述 fail-fast 探测相反）——@MqListener 的 idempotent 自 2026-07-14 起默认 true，消费链默认就会调用 store，Redis 不可达在消费期（而非启动期）以连接异常暴露。逃生开关 mq.idempotent.enabled=false（关掉后去重不生效、启动仅打 WARN）。
3. core 依赖 driver：MqCoreAutoConfiguration 里 MqTemplate 和 MqListenerRegistry 都是 @ConditionalOnBean(MqDriver.class) —— 有 driver 才有门面。driver 与 idempotent-redis 的 AutoConfiguration 都用 @AutoConfigureBefore(MqCoreAutoConfiguration.class) 保证自己先装。
4. 注解扫描：MqListenerAnnotationBeanPostProcessor（一个 BeanPostProcessor）扫描所有 bean 的方法，把 @MqListener 方法收集成 MqListenerEndpoint 列表。
5. 启动订阅：MqListenerRegistry 实现 SmartLifecycle（phase = Integer.MAX_VALUE - 100，最后启动），在 start() 里为每个 endpoint 组装 MqConsumeDispatcher，包成 MqSubscription 交给 driver.subscribe()。

核心执行流

发送：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 → 静默丢消息。
• 批量消费：满批 / 超时（batchTimeoutMs 缺省 60s，从本批首条算起；registry 全局单线程 mq-batch-flush 每 1s 扫）/ 停机（stop 时 forceFlush）三路径刷出（forceFlush 同时打烊：停机后迟到的消息拒收抛 InProgress、不 ACK、下次启动重投，堵住 driver.stop 异步返回的缝隙；已过 tryAcquire 才撞上打烊的拒收会先 markFail 复位标记再抛，避免重投一直撞 stale processing 白烧重试计数），统一走 doFlush（flushLock 串行化，业务批量方法任一时刻至多一个调用）；kill -9 仍丢内存未刷批（上界 = 一个超时窗口）。批量失败因消息已逐条 ACK 无法精确重投整批，故落库（同样只在配了 errorHandler 时）+ 吞掉不抛。
• 停机顺序（MqListenerRegistry.stop()，2026-07-15 review 修复后）：driver 停（异常吞掉打 error，不得中断后续强刷）→ 定时器 shutdown() + awaitTermination(30s) 等在途的超时刷批跑完（超时快照已把批移出 buffer，forceFlush 看不见它，不等就是停机丢批）→ 逐监听器 forceFlush → running=false。改停机代码时别破坏这条链的任何一环。
• 已知语义 / 结构注意（2026-07-15 review 逐条拍板"记录不修"）：① 批间不保序——超时批快照后、抢到 flushLock 前可能被新满批插队（MQ 重投本身不保序，框架不承诺批间有序，README 已写给业务方）；② 停机强刷无超时——forceFlush → doFlush 排队 flushLock 无上界，挂死的业务方法会卡住 stop（与满批路径同风险，已接受，README 已写）；③ start 中途失败无回滚——先 dispatchers.add 后 driver.subscribe，中途抛异常再重试 start 会重复订阅先成功的 endpoint（既有行为，Spring 实际几乎不重试 start）；④ "锁内快照→清空→锁外 doFlush"样板在三处重复——dispatch 满批分支 / flushIfTimeout / forceFlush，仅 drain 条件不同，改其中一处务必对照另外两处，防静默分叉。

driver 侧对 MqMessageInProgressException 的对应处理：RedisStreamMqDriver.java 捕获后不 acknowledge()（留 PEL 待 reclaim）；RocketMqDriver.java 在通用 catch(Exception) 之前单独捕获并返回 RECONSUME_LATER（绝不进耗尽落库分支——catch 顺序不能动，它是 Exception 子类）。边缘（两端共性，2026-07-14 复核）：InProgress 白跑会烧重试计数——rocketmq 每次 RECONSUME_LATER 使 reconsumeTimes +1，烧光进原生 %DLQ%（进死信、非丢消息）；redis-stream 每次 reclaim（xClaim）使 PEL delivery +1，deliveryCount > maxRetry 时不再 dispatch 直接判耗尽（配了 errorHandler 落库，没配则静默丢弃——业务可能一次都没重跑）。缓解同构：mq.idempotent.processing-ttl 小于 maxRetry 轮重投累计时长（rocketmq 默认 ~100s / redis-stream 为 maxRetry × reclaim-min-idle 默认 90s）。

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 条一批跑。
• batchTimeoutMs 仅批量监听器生效；batchSize<=1 且显式配了非缺省值时同样只打 WARN 软校验、不阻断（恰好显式写 60000 检测不出，已知可接受）。
• 同 group 订阅 tag 一致性强校验（2026-07-16）：同一 group 声明不等价的 tag 表达式直接抛 IllegalStateException（订阅关系不一致 = 静默丢消息；等价性按 MqTagMatcher.validParts 有效 tag 集合比较，*/空/病态式同为全收不算冲突）。只覆盖单应用内，跨应用/滚动发布窗口的不一致防不住。病态表达式（写了内容却解析不出有效 tag，如 "||"、纯空白）只打 WARN 不阻断——redis 端全收 vs rocketmq 端行为不同的分叉本身维持现状（2026-07-16 拍板记录不修）。

启动期 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。