100 lines
11 KiB
Markdown
100 lines
11 KiB
Markdown
# 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` 标记过期后重投。搞错就是**静默丢消息**。
|
||
- **单条消费失败**：先落库 + `store.markFail()`（把去重标记从 processing 复位），再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processing，reclaim 重投时被去重挡掉并 ACK → 静默丢消息。
|
||
- **批量消费**：无 idle flush，攒够 `batchSize` 才 flush；低流量下未满批会停滞，进程重启丢内存里没入库的那批（与 legacy 行为一致，适合高流量流）。批量失败因消息已逐条 ACK 无法精确重投整批，故落库 + 吞掉不抛。
|
||
|
||
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 丢掉。默认实现 `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）。
|
||
|
||
## `@MqListener` 方法签名约束
|
||
|
||
`MqListenerAnnotationBeanPostProcessor` 在启动扫描时强校验，违反直接抛 `IllegalStateException`：
|
||
|
||
- 参数只能是 `(payload)` 或 `(payload, MqContext)`，第二参必须是 `MqContext`。
|
||
- `batchSize > 1` 时首参要用 `List<T>`，且**必须声明具体元素类型**（`List<FrontData>`，不能裸 `List`）—— driver 仍按 `T` 逐条反序列化，dispatcher 负责攒批。
|
||
|
||
## 启动期 fail-fast 校验（近期新增，改动时留意）
|
||
|
||
三处独立校验，都受统一逃生开关 `mq.startup-check.enabled`（默认 `true`，仅显式 `false` 关闭；见 `MqStartupChecks`）控制：
|
||
|
||
1. `MqDriverPresenceAutoConfiguration`（core）：配了 `mq.type` 却没装出任何 `MqDriver`（忘引 starter 或 `mq.type` 拼错）→ 启动失败并给中文提示。放在 core 是因为只有 core 有"到底有没有 driver"的全局视角。
|
||
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。
|