docs: 去重默认开启口径落文档 + 异构发送方对接契约 + tag 用法教学

README:idempotent 默认 true 全文对齐(快速开始加 spring.redis、能力表、注解属性表、 troubleshooting 加「只有第一条能消费」条目);新增「Redis Stream 消息结构」一节 (五字段信封、enc=plain/gzip、key 逐条唯一契约、rocketmq 端 KEYS 属性补注); 新增 tag 小节(broker 过滤 vs redis-stream 全收)。 CLAUDE.md:装配链路第 2 条改为「默认调用 store、Redis 不可达在消费期暴露」口径。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-14 13:04:08 +01:00
parent 94d5de5182
commit ec5b22b392
2 changed files with 64 additions and 10 deletions
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -46,7 +46,7 @@ mq-spring-boot-starter      聚合 jar，业务方只引这一个（恒含 idemp
 各模块 `META-INF/spring.factories` 注册 AutoConfiguration。装配开关是**唯一的** `mq.type`：

 1. **driver 二选一**：`RedisStreamMqDriverAutoConfiguration` / `RocketMqDriverAutoConfiguration` 都用 `@ConditionalOnProperty(name="mq.type", havingValue=...)` 按 `mq.type` 条件装配，各自产出一个 `MqDriver` bean。没配 `mq.type` 则两个都不装，`MqTemplate`/`@MqListener` 全部不生效。
-2. **去重 store 不按 mq.type 门控**：`MqIdempotentRedisAutoConfiguration`（idempotent-redis 模块）只按 Redis 在场门控（`@ConditionalOnClass/OnBean(StringRedisTemplate)` + `@AutoConfigureAfter(RedisAutoConfiguration)`），这是 rocketmq 部署也能用上去重的关键。注意 classpath 有 lettuce 时 Spring Boot 会用 localhost 默认值自动造 `StringRedisTemplate`，所以该条件几乎恒真——真正的启动安全网是**懒连接 + core 仅在 `idempotent=true` 时才调用 store**（该装配与 store 构造都刻意不 PING Redis，与下述 fail-fast 探测相反）。逃生开关 `mq.idempotent.enabled=false`。
+2. **去重 store 不按 mq.type 门控**：`MqIdempotentRedisAutoConfiguration`（idempotent-redis 模块）只按 Redis 在场门控（`@ConditionalOnClass/OnBean(StringRedisTemplate)` + `@AutoConfigureAfter(RedisAutoConfiguration)`），这是 rocketmq 部署也能用上去重的关键。注意 classpath 有 lettuce 时 Spring Boot 会用 localhost 默认值自动造 `StringRedisTemplate`，所以该条件几乎恒真——真正的启动安全网只剩**懒连接**（该装配与 store 构造都刻意不 PING Redis，与下述 fail-fast 探测相反）——`@MqListener` 的 `idempotent` 自 2026-07-14 起**默认 `true`**，消费链默认就会调用 store，Redis 不可达在消费期（而非启动期）以连接异常暴露。逃生开关 `mq.idempotent.enabled=false`（关掉后去重不生效、启动仅打 WARN）。
 3. **core 依赖 driver**：`MqCoreAutoConfiguration` 里 `MqTemplate` 和 `MqListenerRegistry` 都是 `@ConditionalOnBean(MqDriver.class)` —— 有 driver 才有门面。driver 与 idempotent-redis 的 AutoConfiguration 都用 `@AutoConfigureBefore(MqCoreAutoConfiguration.class)` 保证自己先装。
 4. **注解扫描**：`MqListenerAnnotationBeanPostProcessor`（一个 `BeanPostProcessor`）扫描所有 bean 的方法，把 `@MqListener` 方法收集成 `MqListenerEndpoint` 列表。
 5. **启动订阅**：`MqListenerRegistry` 实现 `SmartLifecycle`（`phase = Integer.MAX_VALUE - 100`，最后启动），在 `start()` 里为每个 endpoint 组装 `MqConsumeDispatcher`，包成 `MqSubscription` 交给 `driver.subscribe()`。
--- a/README.md
+++ b/README.md
@@ -67,6 +67,10 @@ rocketmq:
  name-server: localhost:9876
  producer:
    group: my-producer-group
+spring:
+  redis:             # 去重默认开启、状态存 Redis，所以 rocketmq 部署也要配 Redis；
+    host: localhost  # 确实不用去重时可去掉，并配 mq.idempotent.enabled=false 全局关闭
+    port: 6379
 ```

 `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` 都不生效。
@@ -111,13 +115,29 @@ public class FrontDataConsumer {

 方法签名只认这两种：`(payload)` 或 `(payload, MqContext)`。

+### tag：同一个 topic 下再分小类
+
+topic 是大频道，tag 是频道内给消息贴的标签——多个下游订同一个 topic、各自只认自己的标签，就不用为每个小类单独开 topic。发送方打标签、消费方声明要哪类，就这两步：
+
+```java
+mq.send("File_Topic", "node-01", fileData);   // 三参 send 的第二参就是 tag
+
+@MqListener(topic = "File_Topic", group = "file_consumer", tag = "node-01")
+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` 时的行为翻转点之一。
+
 ## 进阶用法

 去重、批量、失败落库都是 `@MqListener` 上的属性，业务方法本身不用改：

 ```java
-// 去重：重复消息不会再进业务（去重状态存 Redis，两个 driver 都支持，见下方说明）
-@MqListener(topic = "LN_Topic", group = "g", idempotent = true)
+// 去重默认开启：重复消息不会再进业务（去重状态存 Redis，两个 driver 都支持，见下方说明）
+// 确实不需要去重的监听器显式关掉：
+@MqListener(topic = "LN_Topic", group = "g", idempotent = false)
 public void onData(FrontData msg) { }

 // 批量：攒够 batchSize 条一次给你，参数要用 List
@@ -156,7 +176,7 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
 |---|:---:|:---:|
 | 收发 | ✅ | ✅ |
 | 并发消费 `concurrency` | ⚠️ 单线程 poll，未兑现 | ⚠️ 原生线程池，不受 concurrency 控制 |
-| 去重 `idempotent` | ✅（需 Redis） | ✅（需 Redis） |
+| 去重 `idempotent`（默认开启） | ✅（需 Redis） | ✅（需 Redis） |
 | 失败落库 `RETRY_EXHAUSTED`（需配 `errorHandler`，否则仅 ACK 丢弃） | ✅ | ✅ 落库+ACK |
 | 失败落库 `SINGLE` | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） |
 | 批量 `batchSize` | ⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致） | ⚠️ core 内存攒批（driver 逐条喂）；无 idle flush，低频停滞/重启丢未满批（两端一致） |
@@ -171,7 +191,37 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
 > 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 没配好而影响启动。
+去重按**消息 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、不影响启动）。
+
+## Redis Stream 消息结构（异构发送方对接）
+
+redis-stream 模式下，消息在 Redis 里**不是裸 JSON**，而是固定五字段的信封（XADD 的 field），契约与内部 `redis-stream-springboot-starter` 一致（该格式当初就是和 C++ 端对死的）。不经 mq-starter 的发送方（C++ / 脚本 / 其他语言）直接 XADD 进对应 stream，`@MqListener` 一样能收，只要按这个结构写：
+
+| field | 内容 |
+|---|---|
+| `key` | 消息业务 key，**每条必须唯一**——消费端去重按它判重（`idempotent` 默认开启），key 重复或恒定会把不同消息误判成重复而丢弃。经 `MqTemplate` 发送时自动生成雪花 ID，无需关心 |
+| `tag` | 路由 / 过滤标记，可为空串（注意 redis-stream 消费端当前不按 tag 过滤，全收） |
+| `enc` | **body 的编码方式，压缩标识就放这里**：`plain`（明文 JSON）或 `gzip`（压缩，见下） |
+| `body` | 消息体：`enc=plain` 时为 JSON 明文；`enc=gzip` 时为 `base64(gzip(json))`——先标准 gzip（RFC1952，带 gzip 头），再 base64（基础编码、不带换行） |
+| `ts` | 发送时间戳（毫秒） |
+
+发送方需要压缩时（大报文省 Redis 内存）：`enc` 填 `gzip`，`body` 填「JSON → gzip → base64」的结果；不压缩就 `enc` 填 `plain`、`body` 放 JSON 明文。**只有 `body` 参与压缩，其余四个字段恒明文**。消费端按 `enc` 自动解码，两种都能收。`enc` 别自造取值：未知值会被兜底当明文处理，body 若实际是压缩数据就会解析失败、按毒消息 ACK 丢弃。
+
+```bash
+# 明文
+XADD LN_Topic * key 1815728470893350912 tag "" enc plain body '{"id":1,"name":"..."}' ts 1752480000000
+# 压缩：body 为 base64(gzip(json))
+XADD LN_Topic * key 1815728470893350913 tag "" enc gzip body 'H4sIAAAAAAAA...' ts 1752480000000
+```
+
+两个注意点：
+
+- **stream 名**：默认就是 topic；若开了中间件的环境隔离（`redis-stream.env-isolation`），实际 stream 名是 `topic_环境后缀`（如 `LN_Topic_test`），异构发送方要写到带后缀的那个。
+- **方向差**：mq-starter 自己**发送恒为 plain**（`enc` 照写、恒填 `plain`），压缩解码能力只在消费侧。需要 mq-starter 发送侧也压缩目前做不到，要改 driver。
+
+RocketMQ 模式没有这层信封也没有 `enc`——消息体就是裸 JSON，大报文由 RocketMQ 客户端自动透明压缩（发送端超约 4KB 自动压、消费端自动解），应用层不感知。
+
+但信封里 `key` 的角色在 RocketMQ 模式下由**报文头 KEYS 属性**承担（消费端去重取的是报文头 KEYS，**不读消息体里的任何字段**），异构发送方要遵守同一条契约：**KEYS 逐条唯一、同一条消息重投时保持不变**（发送时 `setKeys(唯一id)`，复用消息体里已有的唯一业务 id 即可）。注意 KEYS 在原生 RocketMQ 里只是可选的运维检索字段，不填也收发一切正常、多年无症状——但接入 mq-starter 消费后它就是去重钥匙：**为空或固定值时所有消息共用一把钥匙，第一条成功后去重窗口内其余全部被判重复、静默 ACK 丢弃**。发送方暂时无法保证 KEYS 唯一的，对应消费端必须显式 `@MqListener(idempotent = false)`。

 ## 架构

@@ -208,12 +258,13 @@ core 是中立的，编排逻辑不依赖 Spring 和 Redis，可以直接单测

 | 属性 | 默认 | 说明 |
 |---|---|---|
-| `topic` / `group` | 必填 | 主题 / 消费组 |
+| `topic` | 必填 | 订阅的主题，和发送端 `send(topic, ...)` 传的 topic 一致才收得到 |
+| `group` | 必填 | 消费组名。同一组的多个实例**分摊**消费（一条消息只被组内一个实例消费），不同组**各收一份**全量。消费进度跟着组走：换个组名就是新组，两端都**从最新消息开始收、不回放历史**（rocketmq 新组默认最新位点，redis-stream 建组用 `$`） |
 | `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 | 是否去重 |
+| `tag` | `*` | 只收打了指定 tag 的消息（发送端三参 `send` 的第二参打 tag）。rocketmq 为 broker 服务端过滤、支持原生表达式（如 `A \|\| B`）；redis-stream 当前不过滤、全收。详见「消费」一节的 tag 小节 |
+| `idempotent` | true | 是否去重（默认开启，需 Redis 可达；单个监听器关闭写 `idempotent = false`，全局关闭配 `mq.idempotent.enabled=false`） |
 | `errorHandler` | `""` | 失败落库的 bean 名 |

 去重参数（前缀 `mq.idempotent`，driver 中立，都有默认值）：
@@ -246,11 +297,14 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`nam
 **`@MqListener` 方法不触发？**
 依次查：① 所在类是不是 Spring bean；② `topic` / `group` 和发送端是否一致；③ 方法签名是否合法（单条 `(payload)` 或 `(payload, MqContext)`，批量首参 `List<T>` 带具体元素类型）。

-**`idempotent = true` 在 rocketmq 下不生效？**
+**去重在 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`）。
+去重默认开启（`idempotent` 默认 true），正常不该重复。依次查：① 是否显式关掉了（监听器上 `idempotent = false`，或全局 `mq.idempotent.enabled=false`）；② 去重 store 是否已装配、Redis 是否可达（默认实现随 `mq-starter-idempotent-redis` 自动注册，不看 `mq.type`）。
+
+**对接异构发送方后，只有第一条消息能消费、后面成片丢失？**
+几乎必是去重 key 撞车：发送方没填或写死了 key（redis-stream 信封 `key` field / rocketmq 报文头 KEYS 属性），所有消息共用一把去重钥匙，第一条成功后其余全被判重复静默 ACK。让发送方保证 key 逐条唯一（见「Redis Stream 消息结构」一节），兑现前先在对应监听器上显式 `idempotent = false`。

 **批量消费收不到 / 停滞 / 丢消息？**
 没有 idle flush，攒不满 `batchSize` 不触发；整批失败会被吞掉不重投。低流量把 `batchSize` 调小或改单条消费。