diff --git a/CLAUDE.md b/CLAUDE.md index 2f92e8a..29e1cf7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -32,12 +32,12 @@ mvn -pl mq-starter-core test -Dtest=MqConsumeDispatcherTest#someMethod # 单个 ``` 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-redis-stream Redis Stream driver（去重/批量/落库/reclaim 重试/tag 消费端过滤全接入；去重 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%`）与批量**整批失败**的处理（逐条 ACK 后无法精确重投整批，故落库+吞掉）；真正两端都未兑现的只有 `concurrency`。批量攒批本身（含 idle flush 超时/停机强刷，2026-07 已落地）两端已对齐。改跨 driver 行为前先对照 README 能力表。 +> ⚠️ **两个 driver 仍有差异**（README 有对照表）：去重、失败落库、tag 过滤三者两端已对齐（2026-07 起 rocketmq `idempotent=true` 生效、需 Redis 可达；tag：rocketmq broker 原生 / redis-stream 消费端模拟，语义子集见 core 的 `MqTagMatcher`，dispatch 顶部与 reclaim 耗尽分支前两处调用）；剩余差距是死信模型（redis 只落库 / rocketmq 原生 `%DLQ%`）与批量**整批失败**的处理（逐条 ACK 后无法精确重投整批，故落库+吞掉）；真正两端都未兑现的只有 `concurrency`。批量攒批本身（含 idle flush 超时/停机强刷，2026-07 已落地）两端已对齐。改跨 driver 行为前先对照 README 能力表。 > > **单独只引 `mq-starter-rocketmq`（不走聚合坐标）时去重需自行加 `mq-starter-idempotent-redis` 依赖**——rocketmq 模块本身不依赖它（刻意零耦合）。 diff --git a/README.md b/README.md index 602fa35..a872249 100644 --- a/README.md +++ b/README.md @@ -128,7 +128,7 @@ public void onFile(FileData msg) { } // 只收 tag=node-01 的消息， - 消费端不写 `tag` 默认 `*`，整个 topic 全收；发送端用两参 `send` 则消息不带 tag——**对端若按 tag 过滤，不带标签的消息它收不到**，跨系统对接时这是个经典坑。 - rocketmq 模式：过滤发生在 **broker 服务端**，不匹配的消息根本不会投到消费者；tag 表达式直接透传 RocketMQ 原生语法，`"A || B"` 多选一也支持。 -- redis-stream 模式：**当前不过滤、全收**（见能力表），tag 只是随消息带到消费端，可从 `MqContext.getTag()` 读到，需要分流的话业务自己判断——这也是切换 `mq.type` 时的行为翻转点之一。 +- redis-stream 模式：**消费端模拟过滤**——driver 收到消息后按订阅 tag 比对，不匹配的直接 ACK 丢弃、不进消费链（不碰去重、不算重试、不落库），支持与 rocketmq 相同的写法子集（`*` 全收、单 tag 精确、`A || B` 多选一）。与 rocketmq 的差别只剩过滤位置与成本：消息会先走一趟网络到消费端才被丢弃。 ## 进阶用法 @@ -172,11 +172,12 @@ public class MyErrorHandler implements MqConsumeErrorHandler { ## 两个 driver 不太一样 -两套实现的能力大头已经对齐：去重、失败落库、批量攒批（含 idle flush）两端都接上了。剩下的差异主要在死信模型，以及批量的**整批失败处理**（攒批本身两端已对齐，糙点在整批失败被吞、无法精确重投）： +两套实现的能力大头已经对齐：去重、失败落库、批量攒批（含 idle flush）、tag 过滤两端都接上了。剩下的差异主要在死信模型，以及批量的**整批失败处理**（攒批本身两端已对齐，糙点在整批失败被吞、无法精确重投）： | 能力 | Redis Stream | RocketMQ | |---|:---:|:---:| | 收发 | ✅ | ✅ | +| tag 过滤 | ✅ 消费端模拟（不匹配 ACK 丢弃，写法子集同右） | ✅ broker 服务端原生 | | 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 | | 去重 `idempotent`（默认开启） | ✅（需 Redis） | ✅（需 Redis） | | 失败落库 `RETRY_EXHAUSTED`（需配 `errorHandler`，否则仅 ACK 丢弃） | ✅ | ✅ 落库+ACK | @@ -188,10 +189,9 @@ public class MyErrorHandler implements MqConsumeErrorHandler { | 事务 / 延迟消息 | ❌ | ❌ 没暴露 | | 序列化 | fastjson | fastjson | -> ⚠️ **切换 `mq.type` 业务代码零改动，但三处运行时行为会跟着翻转**，切换前按这三条过一遍消费方假设： -> 1. **tag 过滤**：redis-stream 消费端不过滤（全收）、rocketmq 走 broker 服务端按 tag 过滤——依赖 tag 分流的业务切 rocketmq 会"只收匹配的"、切回 redis 会"收到全部"。 -> 2. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。 -> 3. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。 +> ⚠️ **切换 `mq.type` 业务代码零改动，但两处运行时行为会跟着翻转**，切换前按这两条过一遍消费方假设： +> 1. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。 +> 2. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。 去重按**消息 key** 判重：经 `MqTemplate` 发送时自动生成雪花 ID、天然唯一，业务无感；**不经 `MqTemplate` 的异构发送方必须自己保证 key 逐条唯一**（redis-stream = 信封 `key` field，rocketmq = 报文头 KEYS 属性；为空或固定值会把不同消息误判成重复而静默丢弃，详见「Redis Stream 消息结构」一节及其末尾的 RocketMQ 补注）。去重的状态存在 Redis（`processing / success / fail` 三态 + TTL），由独立模块 `mq-starter-idempotent-redis` 提供，**不看 `mq.type`**——所以 RocketMQ 部署只要能连上 Redis（标准 `spring.redis.*` 配置），去重一样生效。引聚合坐标的话这个模块已经打在里面；单独只引 `mq-starter-rocketmq` 的话需要自己再加一个 `mq-starter-idempotent-redis` 依赖。**去重默认开启**，消费链路会读写 Redis，务必保证 Redis 可达——连不上时消费抛连接异常走重投（日志可见、不是静默丢，但业务会一直消费不进去）。确实不需要去重的监听器显式写 `idempotent = false`；或配 `mq.idempotent.enabled=false` 全局关掉 store（关掉后去重不生效、启动打 WARN 提示，完全不碰 Redis、不影响启动）。 @@ -202,7 +202,7 @@ redis-stream 模式下，消息在 Redis 里**不是裸 JSON**，而是固定五 | field | 内容 | |---|---| | `key` | 消息业务 key，**每条必须唯一**——消费端去重按它判重（`idempotent` 默认开启），key 重复或恒定会把不同消息误判成重复而丢弃。经 `MqTemplate` 发送时自动生成雪花 ID，无需关心 | -| `tag` | 路由 / 过滤标记，可为空串（注意 redis-stream 消费端当前不按 tag 过滤，全收） | +| `tag` | 路由 / 过滤标记，可为空串——消费端按订阅 tag 过滤，不带 tag 的消息只有订 `*`（缺省）的监听器收得到 | | `enc` | **body 的编码方式，压缩标识就放这里**：`plain`（明文 JSON）或 `gzip`（压缩，见下） | | `body` | 消息体：`enc=plain` 时为 JSON 明文；`enc=gzip` 时为 `base64(gzip(json))`——先标准 gzip（RFC1952，带 gzip 头），再 base64（基础编码、不带换行） | | `ts` | 发送时间戳（毫秒） | @@ -266,7 +266,7 @@ core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测 | `batchSize` | 1 | 攒批大小，大于 1 要用 `List` 参数；与首参是否 `List` 错配时启动打 WARN、仍按首参跑 | | `batchTimeoutMs` | 60000 | 批量攒批最长等待毫秒数（从本批第一条算起，到点未满也强制刷出）；0/负数关闭；仅批量监听器生效 | | `maxRetry` | 3 | 最大重试次数（redis-stream 按 PEL delivery 计数判耗尽 / rocketmq 写入 broker `maxReconsumeTimes`；被去重标记挡回的 InProgress 白跑**两端都计入**） | -| `tag` | `*` | 只收打了指定 tag 的消息（发送端三参 `send` 的第二参打 tag）。rocketmq 为 broker 服务端过滤、支持原生表达式（如 `A \|\| B`）；redis-stream 当前不过滤、全收。详见「消费」一节的 tag 小节 | +| `tag` | `*` | 只收打了指定 tag 的消息（发送端三参 `send` 的第二参打 tag）。两端都生效：rocketmq 为 broker 服务端过滤（原生表达式透传）；redis-stream 为消费端模拟过滤，支持相同写法子集（`*`、单 tag、`A \|\| B`）。详见「消费」一节的 tag 小节 | | `idempotent` | true | 是否去重（默认开启，需 Redis 可达；单个监听器关闭写 `idempotent = false`，全局关闭配 `mq.idempotent.enabled=false`） | | `errorHandler` | `""` | 失败落库的 bean 名 | @@ -319,7 +319,7 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`nam 多半是「处理中」标记没过期挡住了重投：消费者崩溃残留标记，或单条处理耗时超过 `mq.idempotent.processing-ttl`。等 TTL 过期会自动恢复；调 `processing-ttl` 时注意配置表里的上下界约束。rocketmq 下持续重投烧光 `maxReconsumeTimes` 会进原生 `%DLQ%`（进死信，不是丢消息）。 **从 redis-stream 切到 rocketmq 后行为变了（多收 / 漏收、并发不一样）？** -正常，见「两个 driver 不太一样」里的三条运行时翻转：tag 过滤、并发模型、顺序保证。 +正常，见「两个 driver 不太一样」里的两条运行时翻转：并发模型、顺序保证。 ## 已知限制 / TODO