From 2b635a4cd7497d1811cdeacf0556727f910291af Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 16 Jul 2026 10:24:27 +0100 Subject: [PATCH] =?UTF-8?q?docs:=20=E8=A1=A5=E5=A4=9A=E7=8E=AF=E5=A2=83?= =?UTF-8?q?=E9=9A=94=E7=A6=BB=E8=B7=A8=20driver=20=E5=B7=AE=E5=BC=82?= =?UTF-8?q?=E4=B8=8E=20redis-stream.*=20=E5=BA=95=E5=B1=82=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E8=A1=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - README 能力矩阵新增「多环境隔离」行：redis-stream 经底层 starter 拼 topic_env 后缀， rocketmq 无对应机制；切换 mq.type 须知从两条扩为三条（隔离静默消失）。 - 配置项一节把「redis-stream.* 一般不用动」一句扩成完整表格（env-isolation/env/ read-count/block-ms/consumer-name/max-retry/trim/autoclaim），并标明 trim/autoclaim 只认中间件自己的 handler、对 mq-starter 不生效。 - CLAUDE.md 驱动差异清单同步。 Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 2 +- README.md | 15 ++++++++++++++- 2 files changed, 15 insertions(+), 2 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 29e1cf7..46d0f90 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -37,7 +37,7 @@ mq-starter-rocketmq RocketMQ driver（收发 + 原生重试/死信 + RET mq-spring-boot-starter 聚合 jar，业务方只引这一个（恒含 idempotent-redis） ``` -> ⚠️ **两个 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 能力表。 +> ⚠️ **两个 driver 仍有差异**（README 有对照表）：去重、失败落库、tag 过滤三者两端已对齐（2026-07 起 rocketmq `idempotent=true` 生效、需 Redis 可达；tag：rocketmq broker 原生 / redis-stream 消费端模拟，语义子集见 core 的 `MqTagMatcher`，dispatch 顶部与 reclaim 耗尽分支前两处调用）；剩余差距是死信模型（redis 只落库 / rocketmq 原生 `%DLQ%`）、批量**整批失败**的处理（逐条 ACK 后无法精确重投整批，故落库+吞掉）与**多环境隔离**（redis-stream 经底层 starter 的 `redis-stream.env-isolation`/`env` 拼 `topic_env` 后缀；rocketmq 无对应机制，切换后隔离静默消失）；真正两端都未兑现的只有 `concurrency`。批量攒批本身（含 idle flush 超时/停机强刷，2026-07 已落地）两端已对齐。改跨 driver 行为前先对照 README 能力表。 > > **单独只引 `mq-starter-rocketmq`（不走聚合坐标）时去重需自行加 `mq-starter-idempotent-redis` 依赖**——rocketmq 模块本身不依赖它（刻意零耦合）。 diff --git a/README.md b/README.md index 025ad9b..24211cf 100644 --- a/README.md +++ b/README.md @@ -179,6 +179,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler { |---|:---:|:---:| | 收发 | ✅ | ✅ | | tag 过滤 | ✅ 消费端模拟（不匹配 ACK 丢弃，写法子集同右） | ✅ broker 服务端原生 | +| 多环境隔离（topic 拼环境后缀） | ✅ 底层 starter 能力（非 mq-starter 实现）：配 `redis-stream.env-isolation=true` + `redis-stream.env=dev` 后实际 stream 名为 `topic_dev`，各环境共用一套 Redis 互不串流 | ❌ topic 原样使用，多环境共用集群需自行隔离（独立集群 / topic 命名约定） | | 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 | | 去重 `idempotent`（默认开启） | ✅（需 Redis） | ✅（需 Redis） | | 失败落库 `RETRY_EXHAUSTED`（需配 `errorHandler`，否则仅 ACK 丢弃） | ✅ | ✅ 落库+ACK | @@ -193,6 +194,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler { > ⚠️ **切换 `mq.type` 业务代码零改动，但两处运行时行为会跟着翻转**，切换前按这两条过一遍消费方假设： > 1. **并发模型**：redis-stream 单线程串行、rocketmq 原生线程池并发——消费逻辑若有非线程安全假设，切到 rocketmq 可能暴露。 > 2. **顺序**：两端都不保证严格有序（rocketmq 用的 Concurrently 而非 Orderly）。 +> 3. **多环境隔离**：redis-stream 的环境后缀是底层 starter 拼的，rocketmq 没有对应机制——切过去后隔离**静默消失**，多环境共用一套 RocketMQ 时同名 topic+group 会被当成同一消费组互相抢消息，切换前先规划好隔离方式。 去重按**消息 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、不影响启动）。 @@ -289,7 +291,18 @@ Redis Stream 这边的参数（前缀 `mq.redis-stream`）： |---|---|---| | `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`，否则崩溃残留会在业务没重跑的情况下烧光重试直接判耗尽 | -Redis 连接走标准 `spring.redis.*`；更底层的 `redis-stream.*`（readCount / blockMs / consumerName 等）来自内部 stream 中间件（`redis-stream-springboot-starter`），一般不用动。 +Redis 连接走标准 `spring.redis.*`。另有一组更底层的参数（前缀 `redis-stream`，来自内部 stream 中间件 `redis-stream-springboot-starter`，mq-starter 的 driver 复用其中一部分）： + +| 配置 | 默认 | 说明 | +|---|---|---| +| `env-isolation` | false | 多环境隔离开关：开启且 `env` 非空时，实际 stream 名 = `topic_环境值`（如 `order_dev`），各环境共用一套 Redis 互不串流。发送、订阅两侧都生效；跨应用收发时两边的 `env` 配置必须一致，否则流名对不上收不到 | +| `env` | 空 | 环境标识（`dev` / `test` / `prod`…），配合上面的开关用 | +| `read-count` | 10 | 单次 XREADGROUP 拉取条数。大堆积追赶想快些可调大（如 100），代价是单批占内存多、批内逐条处理时间拉长 | +| `block-ms` | 2000 | 拉取无消息时的阻塞等待毫秒数，到点空手返回进下一轮 | +| `consumer-name` | 空 | 组内消费者名。留空则每次启动生成随机名（推荐，重启遗留的 PEL 由 reclaim 拿回）；写死会让多实例同名、PEL 归属混在一起 | +| `max-retry` | 3 | 兜底重试次数，仅 `@MqListener` 的 `maxRetry` 配成 0/负数时才用到 | +| `trim.maxlen` / `trim.interval-ms` | 100000 / 60s | ⚠️ **对 mq-starter 消费的 stream 不生效**：中间件的定时裁剪只认自己注册的 handler 的 stream（无 handler 时任务不启动），mq-starter 不走那套 handler。只经 mq-starter 收发的 topic **没有任何裁剪、无界增长**，要封顶得靠直连中间件的服务碰巧订同一 topic，或运维层面自行 XTRIM | +| `autoclaim.*` | — | 同上对 mq-starter 不生效（也是按 handler 找 stream）。mq-starter 有自己的 reclaim，配上表的 `mq.redis-stream.reclaim-min-idle` | RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`name-server`、`producer.group` 等）。