From ae0caafa63b7a79ceb8da0a57ef5b135287b041e Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 9 Jul 2026 20:41:54 +0800 Subject: [PATCH] =?UTF-8?q?docs(mq-starter):=20=E9=87=8D=E5=86=99=20README?= =?UTF-8?q?=20+=20=E6=96=B0=E5=A2=9E=20CLAUDE.md=EF=BC=8Cgitignore=20docs/?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README 重写为面向使用者的完整说明（快速开始/配置项/两 driver 能力对照/架构/TODO） - 新增 CLAUDE.md：补代码结构、装配链路、消费 ACK 语义等踩坑点，供 Claude Code 参考 - .gitignore 忽略 docs/（AI 生成的本地参考/教学资料，不纳入版本库） Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 3 + CLAUDE.md | 95 +++++++++++++++++++++ README.md | 247 +++++++++++++++++++++++++++++++++++++++++++++++------ 3 files changed, 320 insertions(+), 25 deletions(-) create mode 100644 CLAUDE.md diff --git a/.gitignore b/.gitignore index 733b6b7..7847614 100644 --- a/.gitignore +++ b/.gitignore @@ -30,3 +30,6 @@ hs_err_pid* replay_pid* .DS_Store + +# 本地文档（AI 生成的参考/教学资料，不纳入版本库） +docs/ diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..bd3dcab --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,95 @@ +# 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-redis-stream Redis Stream driver（功能完整：去重/批量/落库/reclaim 重试都接上了） +mq-starter-rocketmq RocketMQ driver（基础收发 + 原生重试/死信，去重和 RETRY_EXHAUSTED 落库尚未接入） +mq-spring-boot-starter 聚合 jar，业务方只引这一个 +``` + +> ⚠️ **两个 driver 完整度不对齐**（README 有对照表）。改到跨 driver 行为、或使用者依赖去重/失败落库时，先确认走的是哪个 driver —— 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()`。 + +## 核心执行流 + +**发送**：`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 无法精确重投整批，故落库 + 吞掉不抛。 + +Redis driver 侧对应实现在 `RedisStreamMqDriver.java`：捕获 `MqMessageInProgressException` 时不 `acknowledge()`。 + +## SPI 契约（`com.njcn.mq.spi`） + +driver 只需实现 `MqDriver`（`send` / `subscribe` / `stop`）。其余是可选扩展点，由业务方或 driver 提供 bean： + +- **`MqIdempotentStore`** —— 去重状态机 `processing / success / fail`。注意 `isCompleted(key)` 的默认实现返回 `true`（向后兼容旧行为），**有真状态机的实现必须覆写**，否则 processing 中的消息会被误判为已完成而被 ACK 丢掉。redis-stream 自带 `RedisMqIdempotentStore`。 +- **`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`，且**必须声明具体元素类型**（`List`，不能裸 `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 模块依赖时注意别破坏这个收敛。 +- **序列化两边不一致**：redis-stream 用 Jackson，rocketmq 用 fastjson。跨 driver 的序列化行为不要假设一致。 diff --git a/README.md b/README.md index 63f2e25..ff718a3 100644 --- a/README.md +++ b/README.md @@ -1,46 +1,243 @@ -# 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 + + com.njcn + mq-spring-boot-starter + 1.0.0-SNAPSHOT + +``` + +目前还没发到中央仓库，先在项目根目录 `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` 是唯一的装配开关。没配它，`MqTemplate` 和 `@MqListener` 都不会生效。 + +### 发送 + +```java +@Resource +private MqTemplate mq; + +mq.send("LN_Topic", frontData); // topic + 消息体 +mq.send("File_Topic", nodeId, fileData); // 带路由 key，映射成 tag +``` + +消息体传任意对象就行，driver 负责序列化。每条消息会自动生成一个雪花 ID 当 key，去重时会用到。 + +### 消费 + +方法上打个注解就行，方法参数写你期望的类型，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 +// 去重：重复消息不会再进业务（需要 MqIdempotentStore，redis-stream 自带了一个） +@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 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 存到异常表，后面人工或定时补偿 + } +} +``` + +## 两个 driver 不太一样 + +这点得说在前面：两套实现的完整度并不一致。Redis Stream 这套是照着 core 的设计意图做全的，去重、批量、失败落库、reclaim 重试都接上了；RocketMQ 那套目前只做了基础收发加原生的重试和死信，去重和 `RETRY_EXHAUSTED` 落库还没接进去。 + +| 能力 | Redis Stream | RocketMQ | +|---|:---:|:---:| +| 收发 / 并发消费 | ✅ | ✅ | +| 去重 `idempotent` | ✅ 内置实现 | ❌ 还没接，配了也不生效 | +| 失败落库 `RETRY_EXHAUSTED` | ✅ | ❌ 只会进原生 DLQ | +| 失败落库 `SINGLE` | ✅ | ⚠️ 异常会被转成 RECONSUME_LATER | +| 批量 `batchSize` | ✅ | ⚠️ 逐条回调，靠 core 内存攒批 | +| 重试 | ✅ PEL reclaim | ✅ 原生重投 | +| 死信队列 | ❌ 只落库 | ✅ 原生 `%DLQ%` | +| 顺序消息 | ❌ | ❌ 用的 Concurrently | +| 事务 / 延迟消息 | ❌ | ❌ 没暴露 | +| 序列化 | Jackson | fastjson | + +所以现在如果你依赖去重或者失败落库，先用 Redis Stream。RocketMQ driver 能收能发，但还没「放量」，具体见下面的 TODO。 + +## 架构 + +``` + 业务代码（只见 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-redis-stream` | Redis Stream 实现（XADD / 消费组 XREADGROUP / XACK / XPENDING+XCLAIM reclaim），另带一个默认的去重实现 | +| `mq-starter-rocketmq` | RocketMQ 实现，基于 rocketmq-spring 编程式注册 | +| `mq-spring-boot-starter` | 聚合 jar，业务方引这一个就够 | -## 装配机制 +core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测。想看更细的图文（消费链、SPI 契约、配置速查、FAQ 都在里面），打开 [`docs/mq-starter-全景手册.html`](./docs/mq-starter-全景手册.html)。 -`@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 兑现 | +| `batchSize` | 1 | 攒批大小，大于 1 要用 `List` 参数 | +| `maxRetry` | 3 | 最大重试次数 | +| `tag` | `*` | tag 过滤 | +| `idempotent` | false | 是否去重 | +| `errorHandler` | `""` | 失败落库的 bean 名 | + +Redis Stream 这边的参数（前缀 `mq.redis-stream`，都有默认值）： + +| 配置 | 默认 | 说明 | +|---|---|---| +| `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:` | 去重键前缀 | + +RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置。 + +## 已知限制 / TODO + +按优先级列，详细的在 [`docs/mq-starter-改进优化清单.md`](./docs/mq-starter-改进优化清单.md)。 + +P0，放量到生产前想补的： + +- [x] 失败落库、去重、批量消费（core 已做） +- [ ] 给 rocketmq driver 补上去重和 `RETRY_EXHAUSTED` 落库，跟 redis-stream 对齐 +- [ ] 批量消费加 idle flush，现在低流量下攒不满会卡着，进程重启还会丢内存里没攒满的那批 +- [ ] 顺序消息 +- [ ] 事务消息 + +P1：延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔（现在两边一个 Jackson 一个 fastjson）、广播消费。 + +P2：并发度语义统一、headers 完整透传、`mq.type` 错配时 fail-fast、监控告警。 + +## 构建 + +```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`）在 `mvn verify` 阶段跑。 + +有想法欢迎提 issue 或 PR，优先认领上面 P0 的几项，提交前跑一下 `mvn verify`，改动尽量带上测试。 + +## License + +[Apache License 2.0](./LICENSE)，Copyright 2026 EdmundNightmare。