From 14502449aabaf540de5d7f363204521b05104fec Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Tue, 14 Jul 2026 11:14:30 +0800 Subject: [PATCH] =?UTF-8?q?docs(README):=20=E6=89=BF=E6=8E=A5=E5=85=A8?= =?UTF-8?q?=E6=99=AF=E6=89=8B=E5=86=8C=E4=B8=8A=E6=89=8B=E6=95=99=E5=AD=A6?= =?UTF-8?q?=E2=80=94=E2=80=94=E6=89=8B=E5=86=8C=E5=88=A0=E6=95=99=E5=AD=A6?= =?UTF-8?q?=20tab=20=E5=90=8E=20README=20=E6=88=90=E4=B8=BA=E5=94=AF?= =?UTF-8?q?=E4=B8=80=E4=B8=8A=E6=89=8B=E5=85=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 发送语义:fastjson 序列化约束(无参构造+getter/setter)、send 同步/失败抛异常、反序列化失败两端行为(redis-stream ACK 丢弃/rocketmq 重试到耗尽) - 批量的坑:裸 List 启动报错、无 idle flush、整批失败被吞须配 errorHandler、可加 MqContext 第二参 - 新增「排错 FAQ」章节(8 问,自手册 FAQ 改写) - 配置表小补:fail-ttl 语义、redis-stream.* 底层配置来源 - 架构章节指向手册的描述改为架构图文/能力矩阵/Roadmap(配置速查与 FAQ 不再在手册) Co-Authored-By: Claude Fable 5 --- README.md | 75 ++++++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 60 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 33b5150..610203f 100644 --- a/README.md +++ b/README.md @@ -69,7 +69,7 @@ rocketmq: group: my-producer-group ``` -`mq.type` 是唯一的装配开关。没配它，`MqTemplate` 和 `@MqListener` 都不会生效。 +`mq.type` 是装配开关。注意 rocketmq driver 带 `matchIfMissing=true`：**没配 `mq.type` 会默认装配 rocketmq**（聚合坐标下 classpath 恒有 rocketmq），此时若 `rocketmq.*` 没配好通常直接启动报错；想用 redis-stream **必须显式**写 `mq.type: redis-stream`；配了两端都不认的错误值（拼错）**默认直接 fail-fast**——启动抛 `IllegalStateException` 并附中文排查提示，不是静默不装。仅当显式 `mq.startup-check.enabled=false` 关掉启动校验时，才退化为两个 driver 都不装、`MqTemplate` 和 `@MqListener` 都不生效。 ### 发送 @@ -81,7 +81,12 @@ mq.send("LN_Topic", frontData); // topic + 消息体 mq.send("File_Topic", nodeId, fileData); // 带路由 key，映射成 tag ``` -消息体传任意对象就行，driver 负责序列化。每条消息会自动生成一个雪花 ID 当 key，去重时会用到。 +消息体传任意对象就行，driver 负责序列化（两端都是 fastjson，POJO 记得留公开无参构造和 getter/setter，收发两端用同一个类型）。每条消息会自动生成一个雪花 ID 当 key，去重时会用到。 + +发送语义有两点要知道： + +- `send` 是**同步**调用、返回 `void`（不回传消息 ID）：方法正常返回就代表已入队（rocketmq 是等 broker 确认的 `syncSend`），失败直接抛异常，业务自己 try/catch 或交给上层事务回滚。 +- 消费端**反序列化失败**两端处理不一样：redis-stream 当"毒消息"直接 ACK 丢弃，rocketmq 会反复重试直到耗尽。改消息类字段时注意兼容，建议先自测能 round-trip。 ### 消费 @@ -137,23 +142,35 @@ public class MyErrorHandler implements MqConsumeErrorHandler { } ``` +> ⚠️ **不配 `errorHandler` = 失败静默丢弃**。未指定 `errorHandler` 时，单条 / 批量 / 重试耗尽的任何消费失败最终都会被 ACK、不落库、不留痕迹（ACK 终止是内建行为，与是否落库无关）。关键消费务必配上 `errorHandler` 落库，否则失败即丢、事后无从排查。 +> +> ⚠️ **批量有几个坑**：参数必须写带具体元素类型的 `List`（裸 `List` 启动扫描直接抛 `IllegalStateException`）；没有 idle flush，低流量攒不满会一直等，进程重启会丢内存里没攒满的那批；整批业务失败只回调 SINGLE 落库、不重投（所以批量务必配 `errorHandler`）。只适合高流量、能接受这些取舍的场景。批量方法同样可以加第二参 `MqContext`（拿到的是该批最后一条的元数据）。 +> +> 另外 `batchSize` 与首参是否 `List` **错配时不报错**：启动时打一条 WARN 日志、仍按首参决定走不走批量（`batchSize=50` 配单个 payload 退化为逐条；`List` 首参配 `batchSize` 缺省按 1 条一批）。见启动日志 `[mq] @MqListener batchSize=...`。 + ## 两个 driver 不太一样 两套实现的能力大头已经对齐：去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义： | 能力 | Redis Stream | RocketMQ | |---|:---:|:---:| -| 收发 / 并发消费 | ✅ | ✅ | +| 收发 | ✅ | ✅ | +| 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 | | 去重 `idempotent` | ✅（需 Redis） | ✅（需 Redis） | -| 失败落库 `RETRY_EXHAUSTED` | ✅ | ✅ 落库并 ACK | +| 失败落库 `RETRY_EXHAUSTED`（需配 `errorHandler`，否则仅 ACK 丢弃） | ✅ | ✅ 落库+ACK | | 失败落库 `SINGLE` | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） | | 批量 `batchSize` | ⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致） | ⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致） | | 重试 | ✅ PEL reclaim | ✅ 原生重投 | -| 死信队列 | ❌ 只落库 | ✅ 原生 `%DLQ%` | +| 死信队列 | ❌ 只落库 | ⚠️ 原生 `%DLQ%` 通常被耗尽 ACK 旁路（InProgress 白跑烧光计数时仍会进） | | 顺序消息 | ❌ | ❌ 用的 Concurrently | | 事务 / 延迟消息 | ❌ | ❌ 没暴露 | | 序列化 | fastjson | fastjson | +> ⚠️ **切换 `mq.type` 业务代码零改动，但三处运行时行为会跟着翻转**，切换前按这三条过一遍消费方假设： +> 1. **tag 过滤**：redis-stream 消费端不过滤（全收）、rocketmq 走 broker 服务端按 tag 过滤——依赖 tag 分流的业务切 rocketmq 会"只收匹配的"、切回 redis 会"收到全部"。 +> 2. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。 +> 3. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。 + 去重的状态存在 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 没配好而影响启动。 ## 架构 @@ -183,7 +200,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler { | `mq-starter-rocketmq` | RocketMQ 实现，基于 rocketmq-spring 编程式注册 | | `mq-spring-boot-starter` | 聚合 jar，业务方引这一个就够（恒含 idempotent-redis） | -core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测。想看更细的图文（消费链、SPI 契约、配置速查、FAQ 都在里面），打开 [`docs/mq-starter-全景手册.html`](./docs/mq-starter-全景手册.html)。 +core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测。想看更细的架构图文（消费链、SPI 契约、driver 能力矩阵、改进 Roadmap），打开 [`docs/mq-starter-全景手册.html`](./docs/mq-starter-全景手册.html)；上手和排错看本 README 就够了。 ## 配置项 @@ -192,9 +209,9 @@ core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测 | 属性 | 默认 | 说明 | |---|---|---| | `topic` / `group` | 必填 | 主题 / 消费组 | -| `concurrency` | 1 | 并发消费线程数，实际由 driver 兑现 | -| `batchSize` | 1 | 攒批大小，大于 1 要用 `List` 参数 | -| `maxRetry` | 3 | 最大重试次数 | +| `concurrency` | 1 | 并发消费线程数，**当前两 driver 均未按此字段兑现**（redis 单线程 poll、rocketmq 用原生线程池默认值） | +| `batchSize` | 1 | 攒批大小，大于 1 要用 `List` 参数；与首参是否 `List` 错配时启动打 WARN、仍按首参跑 | +| `maxRetry` | 3 | 最大重试次数（redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker `maxReconsumeTimes`；被去重标记挡回的 InProgress 白跑**两端都计入**） | | `tag` | `*` | tag 过滤 | | `idempotent` | false | 是否去重 | | `errorHandler` | `""` | 失败落库的 bean 名 | @@ -204,9 +221,9 @@ core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测 | 配置 | 默认 | 说明 | |---|---|---| | `enabled` | true | 逃生开关，显式 false 关闭去重 store 装配 | -| `processing-ttl` | 60s | 「处理中」标记存多久，建议大于单条处理耗时；rocketmq 下还建议小于业务重试节奏的累计时长，否则并发中的消息可能反复重投直至进原生 `%DLQ%` | +| `processing-ttl` | 60s | 「处理中」标记存多久，必须大于最长单条处理耗时，且**两端都有上界**（被标记挡回的 InProgress 白跑会烧重试计数）：redis-stream 下需 **≥ `reclaim-min-idle`** 且 **< `maxRetry` × `reclaim-min-idle`**（下界防慢消费双消费；标记活过全部 reclaim 轮次则 delivery 烧光、业务没重跑就判耗尽——未配 `errorHandler` 即静默丢弃），默认 30 ≤ 60 < 90 自洽；rocketmq 下需小于前 `maxRetry` 轮 broker 重投累计时长（默认 10s+30s+1m≈100s），否则并发中 / 崩溃残留的消息把 `reconsumeTimes` 烧光、进原生 `%DLQ%` | | `success-ttl` | 300s | 「已完成」标记存多久（去重窗口） | -| `fail-ttl` | 300s | 失败标记保留多久 | +| `fail-ttl` | 300s | 失败标记保留多久（fail 态不挡重投，重投后允许重新消费） | | `key-prefix` | `mq:idem:` | 去重键前缀 | > 旧键 `mq.redis-stream.idempotent.*` 仍向后兼容（会打一条弃用告警），中立键和旧键同时配置时以 `mq.idempotent.*` 为准。建议尽快迁移。 @@ -215,9 +232,37 @@ Redis Stream 这边的参数（前缀 `mq.redis-stream`）： | 配置 | 默认 | 说明 | |---|---|---| -| `reclaim-min-idle` | 30s | 消息投出去多久没 ack 才 reclaim 重投，要大于单条处理耗时（且建议大于 processing-ttl） | +| `reclaim-min-idle` | 30s | 消息投出去多久没 ack 才 reclaim 重投，要大于单条处理耗时；**不要调到比 `processing-ttl` 还大**（需保持 `processing-ttl ≥ reclaim-min-idle`，否则存活的慢消费者会被 reclaim 重投副本并行双消费）；也别调得太小——每次 reclaim（xClaim）都使 PEL delivery +1，被 processing 挡回的白跑同样计入，需保持 `maxRetry × reclaim-min-idle > processing-ttl`，否则崩溃残留会在业务没重跑的情况下烧光重试直接判耗尽 | -RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置。 +Redis 连接走标准 `spring.redis.*`；更底层的 `redis-stream.*`（readCount / blockMs / consumerName 等）来自内部 stream 中间件（`redis-stream-springboot-starter`），一般不用动。 + +RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`name-server`、`producer.group` 等）。 + +## 排错 FAQ + +**启动找不到 `MqTemplate` bean / 一启动就抛 MQ 相关异常？** +先看 `mq.type`：拼错了默认直接 fail-fast（`IllegalStateException` + 中文排查提示，检查依赖和拼写）；完全没配则默认装 rocketmq，`rocketmq.*` 没配好通常直接启动报错。想用 redis-stream 必须显式写 `mq.type: redis-stream`。详见上文「配置」一节。 + +**`@MqListener` 方法不触发？** +依次查：① 所在类是不是 Spring bean；② `topic` / `group` 和发送端是否一致；③ 方法签名是否合法（单条 `(payload)` 或 `(payload, MqContext)`，批量首参 `List` 带具体元素类型）。 + +**`idempotent = true` 在 rocketmq 下不生效？** +依次查：① Redis 是否可达（标准 `spring.redis.*` 配置）；② classpath 有没有 `mq-starter-idempotent-redis`（聚合坐标已含，单独引 rocketmq 模块要自己加）；③ 是否配了 `mq.idempotent.enabled=false`。开了去重但 Redis 连不上时，消费会抛连接异常走重投，日志可见，不是静默丢。 + +**消息重复消费？** +开 `idempotent = true`，两个 driver 都支持；确认去重 store 已装配（默认实现随 `mq-starter-idempotent-redis` 自动注册，不看 `mq.type`）。 + +**批量消费收不到 / 停滞 / 丢消息？** +没有 idle flush，攒不满 `batchSize` 不触发；整批失败会被吞掉不重投。低流量把 `batchSize` 调小或改单条消费。 + +**消费失败后消息不见了、事后查不到？** +多半是没配 `errorHandler`：没配时单条 / 批量 / 重试耗尽的失败一律只 ACK 丢弃、不留痕迹。关键消费用 `@MqListener(errorHandler = "beanName")` 落库，见「进阶用法」。 + +**开了去重后某条消息迟迟不消费 / 反复重投？** +多半是「处理中」标记没过期挡住了重投：消费者崩溃残留标记，或单条处理耗时超过 `mq.idempotent.processing-ttl`。等 TTL 过期会自动恢复；调 `processing-ttl` 时注意配置表里的上下界约束。rocketmq 下持续重投烧光 `maxReconsumeTimes` 会进原生 `%DLQ%`（进死信，不是丢消息）。 + +**从 redis-stream 切到 rocketmq 后行为变了（多收 / 漏收、并发不一样）？** +正常，见「两个 driver 不太一样」里的三条运行时翻转：tag 过滤、并发模型、顺序保证。 ## 已知限制 / TODO @@ -233,7 +278,7 @@ P0，放量到生产前想补的： P1：延迟消息、死信队列统一抽象、消息轨迹 / 指标、序列化可插拔（两端都是 fastjson，但硬编码换不掉）、广播消费。 -P2：并发度语义统一、headers 完整透传、`mq.type` 错配时 fail-fast、监控告警。 +P2：并发度语义统一、headers 完整透传、监控告警。（启动期 fail-fast 自检 —— `mq.type` 错配 / 中间件连不上时启动报错 —— 已于 2026-07-09 实现，逃生开关 `mq.startup-check.enabled=false`；详见全景手册 Roadmap 13。） ## 构建 @@ -243,7 +288,7 @@ cd mq-starter mvn clean install ``` -需要 JDK 1.8+ 和 Maven 3.6+。单元测试 `mvn test` 不依赖外部中间件；带真实 Redis 的集成测试（`*IT`）在 `mvn verify` 阶段跑。 +需要 JDK 1.8+ 和 Maven 3.6+。单元测试 `mvn test` 不依赖外部中间件；带真实 Redis 的集成测试（`*IT`，连 `localhost:6379`）在 `mvn verify` 阶段跑，测试内部会自检连接、连不上 Redis 自动跳过而非失败。 有想法欢迎提 issue 或 PR，优先认领上面 P0 的几项，提交前跑一下 `mvn verify`，改动尽量带上测试。