Compare commits
5 Commits
96e26c9bbb
...
master
| Author | SHA1 | Date | |
|---|---|---|---|
| 0f7be86bb8 | |||
| 8c39399d09 | |||
| 14502449aa | |||
| 5191f5ace9 | |||
| 3c14b0cca3 |
11
CLAUDE.md
11
CLAUDE.md
@@ -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` 会让标记卡在 processing,reclaim 重投时被去重挡掉并 ACK → 静默丢消息。
|
- **单条消费失败**:先落库(**仅当配了 `errorHandler`**;`safeOnError` 对 null handler 静默跳过、不落库)+ `store.markFail()`(把去重标记从 processing 复位),再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processing,reclaim 重投时被去重挡掉并 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 每次 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`)
|
## 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` 一次 Redis(lettuce 默认懒连接,不主动探测会拖到运行时才暴露连不上)。
|
2. `RedisStreamStartupProbeAutoConfiguration`:`mq.type=redis-stream` 时主动 `PING` 一次 Redis(lettuce 默认懒连接,不主动探测会拖到运行时才暴露连不上)。
|
||||||
3. `RocketMqStartupProbeAutoConfiguration`:`mq.type=rocketmq` 时对 NameServer 做 TCP 短超时探测(任一地址可达即通过)。
|
3. `RocketMqStartupProbeAutoConfiguration`:`mq.type=rocketmq` 时对 NameServer 做 TCP 短超时探测(任一地址可达即通过)。
|
||||||
|
|
||||||
|
|||||||
79
README.md
79
README.md
@@ -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 负责序列化(两端都是 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<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+1m≈100s),否则并发中 / 崩溃残留的消息把 `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 重投副本并行双消费);也别调得太小——每次 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<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`,改动尽量带上测试。
|
||||||
|
|
||||||
|
|||||||
@@ -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={} 但首参非 List,batchSize 不生效、将按单条逐一回调(如需批量攒批请把首参改为 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));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user