Compare commits

5 Commits

Author SHA1 Message Date
0f7be86bb8 docs(CLAUDE.md): 2026-07-14 复核对齐——batchSize 软校验决策 + 三处口径补正
- 方法签名约束:记录 batchSize 与首参 List 错配只打 WARN 不阻断(刻意软校验,走不走批量只由首参决定)
- 失败落库口径:单条/批量落库均仅在配了 errorHandler 时发生(safeOnError 对 null handler 静默跳过);MqConsumeErrorHandler 标注可选、未配则任何失败仅 ACK 丢弃
- InProgress 白跑烧重试计数改记为两端共性(rocketmq reconsumeTimes / redis-stream PEL delivery),缓解同构:processing-ttl < maxRetry 轮重投累计时长
- fail-fast 校验补充:仅显式配 mq.type 时介入,缺省落 rocketmq 时三层校验不触发(已知遗留)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 04:19:36 +01:00
8c39399d09 feat(mq-starter-core): @MqListener batchSize 与首参 List 错配时打 WARN 软校验
走不走批量始终只由首参是否 List 决定,错配不阻断启动:
- batchSize>1 但首参非 List → WARN 提示 batchSize 不生效、按单条逐一回调
- 首参 List 但 batchSize<=1 → WARN 提示将按 1 条一批回调

测试:ListAppender 捕获日志,覆盖两个错配方向 + 匹配时无 WARN。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 04:19:17 +01:00
14502449aa docs(README): 承接全景手册上手教学——手册删教学 tab 后 README 成为唯一上手入口
- 发送语义: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 <noreply@anthropic.com>
2026-07-14 11:14:30 +08:00
5191f5ace9 docs(README): batchSize 能力格对齐全景手册——两端一致(core 内存攒批/no idle flush/重启丢未满批),纠正 redis 侧 伪差异
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-10 15:33:53 +08:00
3c14b0cca3 docs(README): SINGLE 失败落库口径对齐全景手册——两端一致(单条落库+重投/批量失败被吞),纠正 rocketmq 记错的 RECONSUME_LATER 伪差异
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-10 15:12:02 +08:00
4 changed files with 138 additions and 22 deletions

View File

@@ -62,17 +62,17 @@ mq-spring-boot-starter 聚合 jar业务方只引这一个恒含 idemp
`mq-starter-core/.../container/MqConsumeDispatcher.java` 是消费链的心脏,纯逻辑无外部依赖。有几处**不能想当然改动**的正确性约束(都写在注释里): `mq-starter-core/.../container/MqConsumeDispatcher.java` 是消费链的心脏,纯逻辑无外部依赖。有几处**不能想当然改动**的正确性约束(都写在注释里):
- **去重的两类跳过**决定 driver 能否 ACK`success` → 正常返回让 driver ACK 挡掉重复;仍 `processing`(并发中或消费者崩溃残留标记)→ 抛 `MqMessageInProgressException` 让 driver **不要 ACK**、保留消息待 `processing` 标记过期后重投。搞错就是**静默丢消息**。 - **去重的两类跳过**决定 driver 能否 ACK`success` → 正常返回让 driver ACK 挡掉重复;仍 `processing`(并发中或消费者崩溃残留标记)→ 抛 `MqMessageInProgressException` 让 driver **不要 ACK**、保留消息待 `processing` 标记过期后重投。搞错就是**静默丢消息**。
- **单条消费失败**:先落库 + `store.markFail()`(把去重标记从 processing 复位),再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processingreclaim 重投时被去重挡掉并 ACK → 静默丢消息。 - **单条消费失败**:先落库**仅当配了 `errorHandler`**`safeOnError` 对 null handler 静默跳过、不落库)+ `store.markFail()`(把去重标记从 processing 复位),再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processingreclaim 重投时被去重挡掉并 ACK → 静默丢消息。
- **批量消费**:无 idle flush攒够 `batchSize` 才 flush低流量下未满批会停滞进程重启丢内存里没入库的那批与 legacy 行为一致,适合高流量流)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库 + 吞掉不抛。 - **批量消费**:无 idle flush攒够 `batchSize` 才 flush低流量下未满批会停滞进程重启丢内存里没入库的那批与 legacy 行为一致,适合高流量流)。批量失败因消息已逐条 ACK 无法精确重投整批,故落库(同样只在配了 `errorHandler` 时)+ 吞掉不抛。
driver 侧对 `MqMessageInProgressException` 的对应处理:`RedisStreamMqDriver.java` 捕获后不 `acknowledge()`(留 PEL 待 reclaim`RocketMqDriver.java` 在通用 `catch(Exception)` **之前**单独捕获并返回 `RECONSUME_LATER`绝不进耗尽落库分支——catch 顺序不能动,它是 `Exception` 子类)。边缘rocketmq 下 InProgress 反复重投超过 `maxReconsumeTimes` 进原生 `%DLQ%`(进死信、非丢消息);缓解靠 `mq.idempotent.processing-ttl` < 业务重试节奏累计时长 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 每次 reclaimxClaim使 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` ## SPI 契约(`com.njcn.mq.spi`
driver 只需实现 `MqDriver``send` / `subscribe` / `stop`)。其余是可选扩展点,由业务方或 driver 提供 bean 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`)。 - **`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`)。 - **`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 - **`MqSubscription`** —— core 传给 driver 的订阅描述topic/group/tag/payloadType/concurrency/batchSize/maxRetry/listener/retryExhaustedHandler
## `@MqListener` 方法签名约束 ## `@MqListener` 方法签名约束
@@ -81,12 +81,13 @@ driver 只需实现 `MqDriver``send` / `subscribe` / `stop`)。其余是可
- 参数只能是 `(payload)``(payload, MqContext)`,第二参必须是 `MqContext` - 参数只能是 `(payload)``(payload, MqContext)`,第二参必须是 `MqContext`
- `batchSize > 1` 时首参要用 `List<T>`,且**必须声明具体元素类型**`List<FrontData>`,不能裸 `List`)—— driver 仍按 `T` 逐条反序列化dispatcher 负责攒批。 - `batchSize > 1` 时首参要用 `List<T>`,且**必须声明具体元素类型**`List<FrontData>`,不能裸 `List`)—— driver 仍按 `T` 逐条反序列化dispatcher 负责攒批。
- `batchSize` 与首参是否 `List` 的**错配只打 WARN、不阻断启动**2026-07-14 决策,刻意软校验):走不走批量始终只由首参决定——`batchSize>1` 配单 payload 按单条跑,`List` 首参配 `batchSize<=1` 按 1 条一批跑。
## 启动期 fail-fast 校验(近期新增,改动时留意) ## 启动期 fail-fast 校验(近期新增,改动时留意)
三处独立校验,都受统一逃生开关 `mq.startup-check.enabled`(默认 `true`,仅显式 `false` 关闭;见 `MqStartupChecks`)控制: 三处独立校验,都受统一逃生开关 `mq.startup-check.enabled`(默认 `true`,仅显式 `false` 关闭;见 `MqStartupChecks`)控制:
1. `MqDriverPresenceAutoConfiguration`core配了 `mq.type` 却没装出任何 `MqDriver`忘引 starter `mq.type` 拼错)→ 启动失败并给中文提示放在 core 是因为只有 core "到底有没有 driver"的全局视角 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` 一次 Redislettuce 默认懒连接,不主动探测会拖到运行时才暴露连不上)。 2. `RedisStreamStartupProbeAutoConfiguration``mq.type=redis-stream` 时主动 `PING` 一次 Redislettuce 默认懒连接,不主动探测会拖到运行时才暴露连不上)。
3. `RocketMqStartupProbeAutoConfiguration``mq.type=rocketmq` 时对 NameServer 做 TCP 短超时探测(任一地址可达即通过)。 3. `RocketMqStartupProbeAutoConfiguration``mq.type=rocketmq` 时对 NameServer 做 TCP 短超时探测(任一地址可达即通过)。

View File

@@ -69,7 +69,7 @@ rocketmq:
group: my-producer-group 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 mq.send("File_Topic", nodeId, fileData); // 带路由 key映射成 tag
``` ```
消息体传任意对象就行driver 负责序列化。每条消息会自动生成一个雪花 ID 当 key去重时会用到。 消息体传任意对象就行driver 负责序列化(两端都是 fastjsonPOJO 记得留公开无参构造和 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<FrontData>`(裸 `List` 启动扫描直接抛 `IllegalStateException`);没有 idle flush低流量攒不满会一直等进程重启会丢内存里没攒满的那批整批业务失败只回调 SINGLE 落库、不重投(所以批量务必配 `errorHandler`)。只适合高流量、能接受这些取舍的场景。批量方法同样可以加第二参 `MqContext`(拿到的是该批最后一条的元数据)。
>
> 另外 `batchSize` 与首参是否 `List` **错配时不报错**:启动时打一条 WARN 日志、仍按首参决定走不走批量(`batchSize=50` 配单个 payload 退化为逐条;`List` 首参配 `batchSize` 缺省按 1 条一批)。见启动日志 `[mq] @MqListener batchSize=...`。
## 两个 driver 不太一样 ## 两个 driver 不太一样
两套实现的能力大头已经对齐:去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义: 两套实现的能力大头已经对齐:去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义:
| 能力 | Redis Stream | RocketMQ | | 能力 | Redis Stream | RocketMQ |
|---|:---:|:---:| |---|:---:|:---:|
| 收发 / 并发消费 | ✅ | ✅ | | 收发 | ✅ | ✅ |
| 并发消费 `concurrency` | ⚠️ 单线程 poll未兑现 | ⚠️ 原生线程池,不受 concurrency 控制 |
| 去重 `idempotent` | ✅(需 Redis | ✅(需 Redis | | 去重 `idempotent` | ✅(需 Redis | ✅(需 Redis |
| 失败落库 `RETRY_EXHAUSTED` | ✅ | ✅ 落库ACK | | 失败落库 `RETRY_EXHAUSTED`(需配 `errorHandler`,否则仅 ACK 丢弃) | ✅ | ✅ 落库+ACK |
| 失败落库 `SINGLE` | ✅ | ⚠️ 异常会被转成 RECONSUME_LATER | | 失败落库 `SINGLE` | ⚠️ 单条:落库+重投;批量:失败被吞(两端一致) | ⚠️ 单条:落库+重投;批量:失败被吞(两端一致) |
| 批量 `batchSize` | ✅ | ⚠️ 逐条回调,靠 core 内存攒批 | | 批量 `batchSize` | ⚠️ core 内存攒批driver 逐条喂);无 idle flush低频停滞/重启丢未满批(两端一致) | ⚠️ core 内存攒批driver 逐条喂);无 idle flush低频停滞/重启丢未满批(两端一致) |
| 重试 | ✅ PEL reclaim | ✅ 原生重投 | | 重试 | ✅ PEL reclaim | ✅ 原生重投 |
| 死信队列 | ❌ 只落库 | 原生 `%DLQ%` | | 死信队列 | ❌ 只落库 | ⚠️ 原生 `%DLQ%` 通常被耗尽 ACK 旁路InProgress 白跑烧光计数时仍会进) |
| 顺序消息 | ❌ | ❌ 用的 Concurrently | | 顺序消息 | ❌ | ❌ 用的 Concurrently |
| 事务 / 延迟消息 | ❌ | ❌ 没暴露 | | 事务 / 延迟消息 | ❌ | ❌ 没暴露 |
| 序列化 | fastjson | fastjson | | 序列化 | 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 没配好而影响启动。 去重的状态存在 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-starter-rocketmq` | RocketMQ 实现,基于 rocketmq-spring 编程式注册 |
| `mq-spring-boot-starter` | 聚合 jar业务方引这一个就够恒含 idempotent-redis | | `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` | 必填 | 主题 / 消费组 | | `topic` / `group` | 必填 | 主题 / 消费组 |
| `concurrency` | 1 | 并发消费线程数,实际由 driver 兑现 | | `concurrency` | 1 | 并发消费线程数,**当前两 driver 均未按此字段兑现**redis 单线程 poll、rocketmq 用原生线程池默认值) |
| `batchSize` | 1 | 攒批大小,大于 1 要用 `List` 参数 | | `batchSize` | 1 | 攒批大小,大于 1 要用 `List` 参数;与首参是否 `List` 错配时启动打 WARN、仍按首参跑 |
| `maxRetry` | 3 | 最大重试次数 | | `maxRetry` | 3 | 最大重试次数redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker `maxReconsumeTimes`;被去重标记挡回的 InProgress 白跑**两端都计入** |
| `tag` | `*` | tag 过滤 | | `tag` | `*` | tag 过滤 |
| `idempotent` | false | 是否去重 | | `idempotent` | false | 是否去重 |
| `errorHandler` | `""` | 失败落库的 bean 名 | | `errorHandler` | `""` | 失败落库的 bean 名 |
@@ -204,9 +221,9 @@ core 是中立的,编排逻辑不依赖 Spring 和 Redis可以直接单测
| 配置 | 默认 | 说明 | | 配置 | 默认 | 说明 |
|---|---|---| |---|---|---|
| `enabled` | true | 逃生开关,显式 false 关闭去重 store 装配 | | `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+1m100s否则并发中 / 崩溃残留的消息把 `reconsumeTimes` 烧光进原生 `%DLQ%` |
| `success-ttl` | 300s | 已完成标记存多久去重窗口 | | `success-ttl` | 300s | 已完成标记存多久去重窗口 |
| `fail-ttl` | 300s | 失败标记保留多久 | | `fail-ttl` | 300s | 失败标记保留多久fail 态不挡重投重投后允许重新消费 |
| `key-prefix` | `mq:idem:` | 去重键前缀 | | `key-prefix` | `mq:idem:` | 去重键前缀 |
> 旧键 `mq.redis-stream.idempotent.*` 仍向后兼容(会打一条弃用告警),中立键和旧键同时配置时以 `mq.idempotent.*` 为准。建议尽快迁移。 > 旧键 `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 重投副本并行双消费也别调得太小——每次 reclaimxClaim都使 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<T>` 带具体元素类型)。
**`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 ## 已知限制 / TODO
@@ -233,7 +278,7 @@ P0放量到生产前想补的
P1延迟消息死信队列统一抽象消息轨迹 / 指标序列化可插拔两端都是 fastjson但硬编码换不掉)、广播消费 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 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`改动尽量带上测试 有想法欢迎提 issue PR优先认领上面 P0 的几项提交前跑一下 `mvn verify`改动尽量带上测试

View File

@@ -1,6 +1,7 @@
package com.njcn.mq.container; package com.njcn.mq.container;
import com.njcn.mq.annotation.MqListener; import com.njcn.mq.annotation.MqListener;
import com.njcn.mq.core.MqContext; import com.njcn.mq.core.MqContext;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeansException; import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.core.annotation.AnnotatedElementUtils;
@@ -8,6 +9,7 @@ import java.lang.reflect.Method;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
@Slf4j
public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor { public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor {
private final List<MqListenerEndpoint> endpoints = new ArrayList<>(); private final List<MqListenerEndpoint> endpoints = new ArrayList<>();
@@ -39,6 +41,12 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor
} else { } else {
payloadType = first; payloadType = first;
} }
// 走不走批量只由首参是否 List 决定batchSize 与之错配时静默按首参跑,这里 WARN 提示而不阻断启动。
if (!batch && meta.batchSize() > 1) {
log.warn("[mq] @MqListener batchSize={} 但首参非 ListbatchSize 不生效、将按单条逐一回调(如需批量攒批请把首参改为 List<T>: {}", meta.batchSize(), m);
} else if (batch && meta.batchSize() <= 1) {
log.warn("[mq] @MqListener 首参为 List 但 batchSize={},将按 1 条一批回调(如需攒批请设置 batchSize>1: {}", meta.batchSize(), m);
}
m.setAccessible(true); m.setAccessible(true);
endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch)); endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch));
} }

View File

@@ -1,11 +1,35 @@
package com.njcn.mq.container; package com.njcn.mq.container;
import ch.qos.logback.classic.Level;
import ch.qos.logback.classic.Logger;
import ch.qos.logback.classic.spi.ILoggingEvent;
import ch.qos.logback.core.read.ListAppender;
import com.njcn.mq.annotation.MqListener; import com.njcn.mq.annotation.MqListener;
import com.njcn.mq.core.MqContext; import com.njcn.mq.core.MqContext;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Test;
import org.slf4j.LoggerFactory;
import java.util.List; import java.util.List;
import static org.junit.jupiter.api.Assertions.*; import static org.junit.jupiter.api.Assertions.*;
class MqListenerScanTest { class MqListenerScanTest {
private final ListAppender<ILoggingEvent> logAppender = new ListAppender<>();
@BeforeEach
void attachAppender() {
logAppender.start();
((Logger) LoggerFactory.getLogger(MqListenerAnnotationBeanPostProcessor.class)).addAppender(logAppender);
}
@AfterEach
void detachAppender() {
((Logger) LoggerFactory.getLogger(MqListenerAnnotationBeanPostProcessor.class)).detachAppender(logAppender);
}
private long warnCount() {
return logAppender.list.stream().filter(e -> e.getLevel() == Level.WARN).count();
}
static class Biz { static class Biz {
@MqListener(topic = "LN_Topic", group = "ln_consumer") @MqListener(topic = "LN_Topic", group = "ln_consumer")
public void onData(String msg) { } public void onData(String msg) { }
@@ -39,4 +63,42 @@ class MqListenerScanTest {
assertTrue(ep.isBatch(), "List 参数应识别为批量"); assertTrue(ep.isBatch(), "List 参数应识别为批量");
assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T"); assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T");
} }
static class BatchSizeButSinglePayload {
@MqListener(topic = "W1", group = "wg1", batchSize = 50)
public void onOne(String msg) { }
}
static class ListButBatchSizeOne {
@MqListener(topic = "W2", group = "wg2")
public void onList(List<String> msgs) { }
}
@Test
void scan_batchSizeGtOneButSinglePayload_warnsAndRunsAsSingle() {
MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
bpp.postProcessAfterInitialization(new BatchSizeButSinglePayload(), "biz");
MqListenerEndpoint ep = bpp.getEndpoints().get(0);
assertFalse(ep.isBatch(), "首参非 List 应按单条跑");
assertEquals(1, warnCount(), "batchSize>1 但首参非 List 应打一条 WARN");
assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchSize"));
}
@Test
void scan_listParamButBatchSizeOne_warnsAndRunsAsBatch() {
MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
bpp.postProcessAfterInitialization(new ListButBatchSizeOne(), "biz");
MqListenerEndpoint ep = bpp.getEndpoints().get(0);
assertTrue(ep.isBatch(), "首参 List 仍按批量跑");
assertEquals(1, warnCount(), "首参 List 但 batchSize<=1 应打一条 WARN");
assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchSize"));
}
@Test
void scan_matchedSignatures_noWarn() {
MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
bpp.postProcessAfterInitialization(new Biz(), "biz");
bpp.postProcessAfterInitialization(new BatchBiz(), "batchBiz");
assertEquals(0, warnCount(), "batchSize 与首参匹配时不应有 WARN");
}
} }