Compare commits
10 Commits
ffa30a7631
...
2bcce78763
| Author | SHA1 | Date | |
|---|---|---|---|
| 2bcce78763 | |||
| 8bba71236f | |||
| 2b635a4cd7 | |||
| aa5a949cb9 | |||
| f9bbc0721c | |||
| c6e2db1275 | |||
| 69a84f2912 | |||
| 83656215f6 | |||
| ac7559151c | |||
| 1f29c71dba |
@@ -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 消费端过滤/订阅 stream 定时裁剪全接入；去重 store 依赖 idempotent-redis 模块。裁剪复用底层 redis-stream.trim.* 配置、默认开启，maxlen<=0 关，只裁订阅过的 stream——中间件自己的 trim 任务只认它的 handler，对 mq-starter 无效）
|
||||
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%`）、批量攒批语义、`concurrency` 两端都未兑现。改跨 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 模块本身不依赖它（刻意零耦合）。
|
||||
|
||||
|
||||
38
README.md
38
README.md
@@ -128,7 +128,8 @@ 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 的差别只剩过滤位置与成本：消息会先走一趟网络到消费端才被丢弃。
|
||||
- **同一 group 的所有订阅必须用同一个 tag 表达式**（含滚动发布期间的新旧实例）——要按 tag 分流，就各用各的 group。同组配不同 tag 时，消息只会投给组内某一个消费者，拿到"不归自己"消息的那个会直接 ACK 吞掉，对方永远收不到（两端皆然，rocketmq 生态称之为「订阅关系不一致」）。
|
||||
|
||||
## 进阶用法
|
||||
|
||||
@@ -172,26 +173,28 @@ public class MyErrorHandler implements MqConsumeErrorHandler {
|
||||
|
||||
## 两个 driver 不太一样
|
||||
|
||||
两套实现的能力大头已经对齐：去重、失败落库两端都接上了。剩下的差异主要在死信模型和批量语义：
|
||||
两套实现的能力大头已经对齐：去重、失败落库、批量攒批（含 idle flush）、tag 过滤两端都接上了。剩下的差异主要在死信模型，以及批量的**整批失败处理**（攒批本身两端已对齐，糙点在整批失败被吞、无法精确重投）：
|
||||
|
||||
| 能力 | Redis Stream | RocketMQ |
|
||||
|---|:---:|:---:|
|
||||
| 收发 | ✅ | ✅ |
|
||||
| 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 |
|
||||
| 失败落库 `SINGLE` | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） | ⚠️ 单条：落库+重投；批量：失败被吞（两端一致） |
|
||||
| 批量 `batchSize` | ⚠️ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） | ⚠️ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） |
|
||||
| 批量 `batchSize` | ✅ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） | ✅ core 内存攒批（driver 逐条喂）；超时/停机强制刷出，仅 kill -9 丢未满批（两端一致） |
|
||||
| 重试 | ✅ PEL reclaim | ✅ 原生重投 |
|
||||
| 死信队列 | ❌ 只落库 | ⚠️ 原生 `%DLQ%` 通常被耗尽 ACK 旁路（InProgress 白跑烧光计数时仍会进） |
|
||||
| 顺序消息 | ❌ | ❌ 用的 Concurrently |
|
||||
| 事务 / 延迟消息 | ❌ | ❌ 没暴露 |
|
||||
| 序列化 | 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）。
|
||||
> 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、不影响启动）。
|
||||
|
||||
@@ -202,7 +205,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 +269,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 名 |
|
||||
|
||||
@@ -288,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 | stream 封顶裁剪，**默认开启**：driver 对**订阅过**的 stream 每周期执行一次 `XTRIM MAXLEN ~ maxlen`（`~` 近似裁剪，实际长度可能略超 maxlen）。⚠️ 裁剪只按条数砍最老的、**不看消费进度**——堆积超过 maxlen 时未消费的老消息也会被删，容量按「可容忍的最大停机时长 × 生产速率」估算；`maxlen` 配 0/负数关闭；纯发送、不订阅的 topic 不裁（裁剪节奏归消费方） |
|
||||
| `autoclaim.*` | — | 对 mq-starter **不生效**：中间件的 autoclaim 只认自己注册的 handler 的 stream（无 handler 任务不启动），mq-starter 不走那套。mq-starter 有自己的 reclaim，配上表的 `mq.redis-stream.reclaim-min-idle` |
|
||||
|
||||
RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`name-server`、`producer.group` 等）。
|
||||
|
||||
@@ -319,10 +333,12 @@ RocketMQ 这边直接复用 rocketmq-spring 原生的 `rocketmq.*` 配置（`nam
|
||||
多半是「处理中」标记没过期挡住了重投：消费者崩溃残留标记，或单条处理耗时超过 `mq.idempotent.processing-ttl`。等 TTL 过期会自动恢复；调 `processing-ttl` 时注意配置表里的上下界约束。rocketmq 下持续重投烧光 `maxReconsumeTimes` 会进原生 `%DLQ%`（进死信，不是丢消息）。
|
||||
|
||||
**从 redis-stream 切到 rocketmq 后行为变了（多收 / 漏收、并发不一样）？**
|
||||
正常，见「两个 driver 不太一样」里的三条运行时翻转：tag 过滤、并发模型、顺序保证。
|
||||
正常，见「两个 driver 不太一样」里的两条运行时翻转：并发模型、顺序保证。
|
||||
|
||||
## 已知限制 / TODO
|
||||
|
||||
> ⚠️ **升级注意（redis-stream）**：订阅 stream 的定时裁剪**默认开启**（2026-07 起）。升级后每个订阅过的 stream 按 `redis-stream.trim.maxlen`（未配则 10 万条）封顶，**堆积超过封顶时最老的未消费消息会被直接删掉**——裁剪只按条数、不看消费进度。堆积可能超限的场景，升级前把 `trim.maxlen` 调大或配 0/负数关闭；容量按「可容忍的最大停机时长 × 生产速率」估算，详见「配置项」一节的 `redis-stream.*` 表。
|
||||
|
||||
按优先级列，详细的在 [`docs/mq-starter-改进优化清单.md`](./docs/mq-starter-改进优化清单.md)。
|
||||
|
||||
P0，放量到生产前想补的：
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
package com.njcn.mq.spi;
|
||||
|
||||
/**
|
||||
* 订阅 tag 表达式匹配，语义对齐 RocketMQ 原生 tag 表达式**子集**：
|
||||
* {@code *}/空 = 全收、单 tag 精确匹配、{@code A || B} 多选一。不支持 SQL92 / 否定 / 通配。
|
||||
*
|
||||
* <p>无服务端过滤能力的 driver（如 redis-stream）在消费端调用本类模拟过滤；
|
||||
* 语义保持单一来源，勿在 driver 内自行实现比对逻辑。rocketmq driver 无需本类（broker 原生过滤）。
|
||||
*/
|
||||
public final class MqTagMatcher {
|
||||
|
||||
private MqTagMatcher() {
|
||||
}
|
||||
|
||||
/**
|
||||
* @param subscriptionTag 订阅方声明的 tag 表达式（{@code @MqListener.tag}，缺省 {@code *}）
|
||||
* @param messageTag 消息携带的 tag（null/空 = 未打标签，只被全收表达式命中）
|
||||
* @return true = 该消息归本订阅消费
|
||||
*/
|
||||
public static boolean matches(String subscriptionTag, String messageTag) {
|
||||
if (subscriptionTag == null) {
|
||||
return true;
|
||||
}
|
||||
String expr = subscriptionTag.trim();
|
||||
if (expr.isEmpty() || "*".equals(expr)) {
|
||||
return true;
|
||||
}
|
||||
boolean hasValidPart = false;
|
||||
for (String part : expr.split("\\|\\|")) {
|
||||
String p = part.trim();
|
||||
if (p.isEmpty()) {
|
||||
continue; // 空段忽略（如 "A ||" 的尾段）
|
||||
}
|
||||
hasValidPart = true;
|
||||
if (p.equals(messageTag)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
// 全空段（如 "||"）= 没声明任何有效 tag = 视同全收；有有效段但都没命中 = 不匹配
|
||||
return !hasValidPart;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package com.njcn.mq.spi;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertFalse;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
|
||||
class MqTagMatcherTest {
|
||||
|
||||
@Test
|
||||
void catchAllExpressions_alwaysMatch() {
|
||||
assertTrue(MqTagMatcher.matches(null, "A"));
|
||||
assertTrue(MqTagMatcher.matches("", "A"));
|
||||
assertTrue(MqTagMatcher.matches(" ", "A"));
|
||||
assertTrue(MqTagMatcher.matches("*", "A"));
|
||||
assertTrue(MqTagMatcher.matches("||", "A")); // 全空段 = 空表达式 = 全收
|
||||
assertTrue(MqTagMatcher.matches("*", null)); // 全收对无 tag 消息同样放行
|
||||
assertTrue(MqTagMatcher.matches("*", ""));
|
||||
}
|
||||
|
||||
@Test
|
||||
void singleTag_exactCaseSensitiveMatch() {
|
||||
assertTrue(MqTagMatcher.matches("A", "A"));
|
||||
assertFalse(MqTagMatcher.matches("A", "B"));
|
||||
assertFalse(MqTagMatcher.matches("A", "a")); // 大小写敏感
|
||||
assertTrue(MqTagMatcher.matches(" A ", "A")); // 订阅侧逐段 trim
|
||||
assertFalse(MqTagMatcher.matches("A", "A ")); // 消息侧不 trim
|
||||
}
|
||||
|
||||
@Test
|
||||
void orExpression_matchesAnyValidPart() {
|
||||
assertTrue(MqTagMatcher.matches("A || B", "A"));
|
||||
assertTrue(MqTagMatcher.matches("A || B", "B"));
|
||||
assertFalse(MqTagMatcher.matches("A || B", "C"));
|
||||
assertTrue(MqTagMatcher.matches("A ||", "A")); // 空段忽略，不影响有效段
|
||||
assertFalse(MqTagMatcher.matches("A ||", "B"));
|
||||
}
|
||||
|
||||
@Test
|
||||
void untaggedMessage_onlyMatchedByCatchAll() {
|
||||
assertFalse(MqTagMatcher.matches("A", null));
|
||||
assertFalse(MqTagMatcher.matches("A", ""));
|
||||
assertFalse(MqTagMatcher.matches("A || B", null));
|
||||
}
|
||||
}
|
||||
@@ -8,6 +8,7 @@ import com.njcn.mq.core.MqMessage;
|
||||
import com.njcn.mq.spi.MqDriver;
|
||||
import com.njcn.mq.spi.MqMessageInProgressException;
|
||||
import com.njcn.mq.spi.MqSubscription;
|
||||
import com.njcn.mq.spi.MqTagMatcher;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.connection.stream.*;
|
||||
import org.springframework.data.domain.Range;
|
||||
@@ -26,6 +27,7 @@ public class RedisStreamMqDriver implements MqDriver {
|
||||
private final RedisStreamProperties props;
|
||||
private final long reclaimMinIdleMs;
|
||||
private final List<Thread> workers = new ArrayList<>();
|
||||
private final RedisStreamTrimScheduler trimScheduler;
|
||||
private volatile boolean running = true;
|
||||
|
||||
public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
|
||||
@@ -37,6 +39,7 @@ public class RedisStreamMqDriver implements MqDriver {
|
||||
public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) {
|
||||
this.redis = redis; this.keyBuilder = keyBuilder; this.props = props;
|
||||
this.reclaimMinIdleMs = reclaimMinIdleMs;
|
||||
this.trimScheduler = new RedisStreamTrimScheduler(redis, props);
|
||||
}
|
||||
|
||||
@Override public void send(String topic, MqMessage message) {
|
||||
@@ -56,6 +59,7 @@ public class RedisStreamMqDriver implements MqDriver {
|
||||
String stream = keyBuilder.buildStream(sub.getTopic());
|
||||
String group = sub.getGroup();
|
||||
ensureGroup(stream, group);
|
||||
trimScheduler.register(stream);
|
||||
String consumer = (props.getConsumerName() == null || props.getConsumerName().isEmpty())
|
||||
? "mq-" + UUID.randomUUID() : props.getConsumerName();
|
||||
Thread t = new Thread(() -> loop(sub, stream, group, consumer), "mq-redis-" + stream + "-" + group);
|
||||
@@ -87,12 +91,23 @@ public class RedisStreamMqDriver implements MqDriver {
|
||||
|
||||
/** @return true=ACK；false=留 PEL（重试）。 */
|
||||
private boolean dispatch(MqSubscription sub, Map<String, String> fields, int maxRetry) {
|
||||
String msgTag = fields.get(StreamMessageConstant.F_TAG);
|
||||
if (!MqTagMatcher.matches(sub.getTag(), msgTag)) {
|
||||
// tag 不匹配：不归本订阅消费，直接 ACK 丢弃（对齐 rocketmq broker 过滤语义，只是位置在消费端）。
|
||||
// 置于反序列化之前：不碰去重标记、不进攒批、不走重试，连 JSON 都不用解。
|
||||
log.debug("[mq-redis] tag 不匹配，ACK 跳过 key={} msgTag={} subTag={}",
|
||||
fields.get(StreamMessageConstant.F_KEY), msgTag, sub.getTag());
|
||||
return true;
|
||||
}
|
||||
Object payload;
|
||||
try {
|
||||
String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
|
||||
payload = JSON.parseObject(json, sub.getPayloadType());
|
||||
} catch (Exception poison) {
|
||||
log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={}", fields.get(StreamMessageConstant.F_KEY), poison);
|
||||
String key = fields.get(StreamMessageConstant.F_KEY);
|
||||
log.error("[mq-redis] 消息反序列化失败，按毒消息 ACK 丢弃 key={} type={} message={}",
|
||||
key, poison.getClass().getSimpleName(), poison.getMessage());
|
||||
log.debug("[mq-redis] 消息反序列化失败堆栈 key={}", key, poison);
|
||||
return true; // 毒消息直接 ACK，不卡队列
|
||||
}
|
||||
MqMessage msg = MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
|
||||
@@ -137,6 +152,13 @@ public class RedisStreamMqDriver implements MqDriver {
|
||||
Map<String, String> fields = new HashMap<>();
|
||||
rec.getValue().forEach((k, v) -> fields.put(
|
||||
new String(k, StandardCharsets.UTF_8), new String(v, StandardCharsets.UTF_8)));
|
||||
if (!MqTagMatcher.matches(sub.getTag(), fields.get(StreamMessageConstant.F_TAG))) {
|
||||
// tag 不匹配：不归本组消费，直接 ACK 丢弃——不得进耗尽分支以本组名义落库，也无须重投。
|
||||
redis.opsForStream().acknowledge(stream, group, rec.getId());
|
||||
log.debug("[mq-redis] tag 不匹配（reclaim），ACK 丢弃 key={} delivery={}",
|
||||
fields.get(StreamMessageConstant.F_KEY), deliveryCount);
|
||||
continue;
|
||||
}
|
||||
if (deliveryCount > maxRetry) {
|
||||
if (sub.getRetryExhaustedHandler() != null) {
|
||||
try {
|
||||
@@ -186,5 +208,5 @@ public class RedisStreamMqDriver implements MqDriver {
|
||||
}
|
||||
private static void sleep(long ms) { try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); } }
|
||||
|
||||
@Override public void stop() { running = false; for (Thread t : workers) t.interrupt(); workers.clear(); }
|
||||
@Override public void stop() { running = false; trimScheduler.stop(); for (Thread t : workers) t.interrupt(); workers.clear(); }
|
||||
}
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
package com.njcn.mq.driver.redis;
|
||||
|
||||
import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.data.redis.core.RedisCallback;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.ScheduledExecutorService;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* 订阅 stream 的定时裁剪（{@code XTRIM <stream> MAXLEN ~ <maxlen>}，近似裁剪）。
|
||||
*
|
||||
* <p>底层中间件的 StreamTrimScheduler 只裁自己注册的 handler 的 stream，而 mq-starter 的消费
|
||||
* 不走 handler 体系——driver 不自己裁的话，只经 mq-starter 收发的 topic 无界增长。配置复用中间件的
|
||||
* {@code redis-stream.trim.*}（默认开启：maxlen 10 万 / 周期 60s）。
|
||||
*
|
||||
* <p>两条语义约束：{@code maxlen<=0} 视为关闭——XTRIM 0 会清空整条 stream，绝不能发出；
|
||||
* 只裁「订阅过」的 stream——与中间件“handler 即消费者”的语义对齐，纯发送方不裁
|
||||
* （裁剪节奏该由消费方掌握，发送方不知道还有谁没消费到哪）。
|
||||
*/
|
||||
@Slf4j
|
||||
class RedisStreamTrimScheduler {
|
||||
private final StringRedisTemplate redis;
|
||||
private final RedisStreamProperties props;
|
||||
private final Set<String> streams = ConcurrentHashMap.newKeySet();
|
||||
private ScheduledExecutorService scheduler;
|
||||
|
||||
RedisStreamTrimScheduler(StringRedisTemplate redis, RedisStreamProperties props) {
|
||||
this.redis = redis;
|
||||
this.props = props;
|
||||
}
|
||||
|
||||
/** 订阅时登记 stream；首个登记且 maxlen>0 时拉起调度线程（重复登记幂等）。 */
|
||||
synchronized void register(String stream) {
|
||||
streams.add(stream);
|
||||
long maxlen = props.getTrim().getMaxlen();
|
||||
if (scheduler != null || maxlen <= 0) return;
|
||||
long interval = props.getTrim().getIntervalMs();
|
||||
scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
|
||||
Thread t = new Thread(r, "mq-redis-trim");
|
||||
t.setDaemon(true);
|
||||
return t;
|
||||
});
|
||||
scheduler.scheduleWithFixedDelay(this::trimAll, interval, interval, TimeUnit.MILLISECONDS);
|
||||
log.info("[mq-redis] stream 裁剪调度启动 interval={}ms maxlen={}（近似裁剪，只裁订阅过的 stream）", interval, maxlen);
|
||||
}
|
||||
|
||||
void trimAll() {
|
||||
long maxlen = props.getTrim().getMaxlen();
|
||||
if (maxlen <= 0) return;
|
||||
for (String stream : streams) {
|
||||
try {
|
||||
redis.execute((RedisCallback<Object>) c -> c.execute("XTRIM",
|
||||
stream.getBytes(StandardCharsets.UTF_8),
|
||||
"MAXLEN".getBytes(StandardCharsets.UTF_8),
|
||||
"~".getBytes(StandardCharsets.UTF_8),
|
||||
String.valueOf(maxlen).getBytes(StandardCharsets.UTF_8)));
|
||||
} catch (Exception e) {
|
||||
// 单条失败不得中断其余 stream 的裁剪
|
||||
log.warn("[mq-redis] stream 裁剪失败 stream={}", stream, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
synchronized void stop() {
|
||||
if (scheduler != null) {
|
||||
scheduler.shutdownNow();
|
||||
scheduler = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package com.njcn.mq.driver.redis;
|
||||
|
||||
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.middle.stream.constant.StreamMessageConstant;
|
||||
import com.njcn.mq.spi.MqSubscription;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class RedisStreamMqDriverPoisonLogTest {
|
||||
|
||||
private final ListAppender<ILoggingEvent> logAppender = new ListAppender<>();
|
||||
private Logger logger;
|
||||
private Level originalLevel;
|
||||
|
||||
@BeforeEach
|
||||
void attachAppender() {
|
||||
logger = (Logger) LoggerFactory.getLogger(RedisStreamMqDriver.class);
|
||||
originalLevel = logger.getLevel();
|
||||
logger.setLevel(Level.DEBUG);
|
||||
logAppender.start();
|
||||
logger.addAppender(logAppender);
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void detachAppender() {
|
||||
logger.detachAppender(logAppender);
|
||||
logger.setLevel(originalLevel);
|
||||
}
|
||||
|
||||
@Test
|
||||
void poisonMessage_logsErrorSummaryAndDebugStackTrace() throws Exception {
|
||||
RedisStreamMqDriver driver = new RedisStreamMqDriver(null, null, null);
|
||||
MqSubscription sub = MqSubscription.builder()
|
||||
.tag("*")
|
||||
.payloadType(String.class)
|
||||
.listener(msg -> fail("poison message must not reach listener"))
|
||||
.build();
|
||||
|
||||
Map<String, String> fields = new HashMap<>();
|
||||
fields.put(StreamMessageConstant.F_KEY, "poison-1");
|
||||
fields.put(StreamMessageConstant.F_TAG, "");
|
||||
fields.put(StreamMessageConstant.F_ENC, "plain");
|
||||
fields.put(StreamMessageConstant.F_BODY, "new");
|
||||
|
||||
Method dispatch = RedisStreamMqDriver.class.getDeclaredMethod("dispatch", MqSubscription.class, Map.class, int.class);
|
||||
dispatch.setAccessible(true);
|
||||
|
||||
Object ack = dispatch.invoke(driver, sub, fields, 3);
|
||||
|
||||
assertEquals(Boolean.TRUE, ack, "poison message should still be ACKed");
|
||||
List<ILoggingEvent> errors = events(Level.ERROR);
|
||||
assertEquals(1, errors.size());
|
||||
ILoggingEvent error = errors.get(0);
|
||||
assertNull(error.getThrowableProxy(), "ERROR summary should not print a stack trace by default");
|
||||
assertTrue(error.getFormattedMessage().contains("key=poison-1"));
|
||||
assertTrue(error.getFormattedMessage().contains("type=JSONException"));
|
||||
assertTrue(error.getFormattedMessage().contains("message="));
|
||||
|
||||
List<ILoggingEvent> debugs = events(Level.DEBUG);
|
||||
assertEquals(1, debugs.size());
|
||||
assertNotNull(debugs.get(0).getThrowableProxy(), "DEBUG log should retain the stack trace for deep diagnosis");
|
||||
}
|
||||
|
||||
private List<ILoggingEvent> events(Level level) {
|
||||
return logAppender.list.stream()
|
||||
.filter(e -> e.getLevel() == level)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,123 @@
|
||||
package com.njcn.mq.driver.redis;
|
||||
|
||||
import com.njcn.mq.core.MqMessage;
|
||||
import com.njcn.mq.spi.MqSubscription;
|
||||
import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
|
||||
import com.njcn.middle.stream.support.StreamKeyBuilder;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
import org.springframework.data.redis.connection.stream.*;
|
||||
import org.springframework.data.redis.core.RedisCallback;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||
import static org.junit.jupiter.api.Assertions.assertNotNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertNull;
|
||||
import static org.junit.jupiter.api.Assertions.assertTrue;
|
||||
import static org.junit.jupiter.api.Assumptions.assumeTrue;
|
||||
|
||||
class RedisStreamMqDriverTagFilterIT {
|
||||
|
||||
@Test
|
||||
void tagFilter_onlyMatchingDelivered_othersAckedCleanly() throws Exception {
|
||||
StringRedisTemplate redis = newRedisOrSkip();
|
||||
StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
|
||||
RedisStreamProperties props = new RedisStreamProperties();
|
||||
props.setConsumerName("it-tag");
|
||||
props.setReadCount(10);
|
||||
props.setBlockMs(300);
|
||||
props.setMaxRetry(3);
|
||||
RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
|
||||
|
||||
String topic = "MQ_TAG_IT_" + System.nanoTime();
|
||||
BlockingQueue<String> got = new LinkedBlockingQueue<>();
|
||||
driver.subscribe(MqSubscription.builder()
|
||||
.topic(topic).group("g1").tag("A").payloadType(String.class)
|
||||
.concurrency(1).batchSize(1).maxRetry(3)
|
||||
.listener(m -> got.add((String) m.getPayload()))
|
||||
.build());
|
||||
|
||||
driver.send(topic, MqMessage.of("k-a", "A", "match"));
|
||||
driver.send(topic, MqMessage.of("k-b", "B", "mismatch"));
|
||||
driver.send(topic, MqMessage.of("k-none", "", "untagged"));
|
||||
|
||||
assertEquals("match", got.poll(5, TimeUnit.SECONDS), "tag=A 的消息应正常消费");
|
||||
assertNull(got.poll(1500, TimeUnit.MILLISECONDS), "tag 不匹配/未打 tag 的消息不得进业务方法");
|
||||
|
||||
Thread.sleep(500); // 留时间让过滤 ACK 落地
|
||||
PendingMessagesSummary summary = redis.opsForStream().pending(kb.buildStream(topic), "g1");
|
||||
assertEquals(0L, summary.getTotalPendingMessages(), "被过滤的消息应 ACK 干净，PEL 无残留");
|
||||
driver.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
void reclaimExhausted_mismatchedTag_ackedWithoutExhaustedCallback() throws Exception {
|
||||
StringRedisTemplate redis = newRedisOrSkip();
|
||||
StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
|
||||
RedisStreamProperties props = new RedisStreamProperties();
|
||||
props.setConsumerName("it-tag-reclaim");
|
||||
props.setReadCount(10);
|
||||
props.setBlockMs(300);
|
||||
props.setMaxRetry(1);
|
||||
RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
|
||||
|
||||
String topic = "MQ_TAG_RECLAIM_IT_" + System.nanoTime();
|
||||
String stream = kb.buildStream(topic);
|
||||
|
||||
// 1. 手工建组（driver 尚未 subscribe，模拟历史遗留 PEL 场景）
|
||||
redis.execute((RedisCallback<Object>) c -> c.execute("XGROUP",
|
||||
"CREATE".getBytes(StandardCharsets.UTF_8), stream.getBytes(StandardCharsets.UTF_8),
|
||||
"g1".getBytes(StandardCharsets.UTF_8), "$".getBytes(StandardCharsets.UTF_8),
|
||||
"MKSTREAM".getBytes(StandardCharsets.UTF_8)));
|
||||
|
||||
// 2. 发一条 tag=B 的消息（send 不依赖 subscribe）
|
||||
driver.send(topic, MqMessage.of("k-stale", "B", "stale"));
|
||||
|
||||
// 3. 手工 XREADGROUP 不 ACK：delivery=1，模拟消费者读走后崩溃
|
||||
List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
|
||||
Consumer.from("g1", "seed"),
|
||||
StreamReadOptions.empty().count(1),
|
||||
StreamOffset.create(stream, ReadOffset.lastConsumed()));
|
||||
assertNotNull(recs);
|
||||
assertEquals(1, recs.size(), "种子消息应已进 PEL");
|
||||
RecordId staleId = recs.get(0).getId();
|
||||
|
||||
// 4. 手工 XCLAIM 一次：delivery=2 > maxRetry=1，reclaim 将走耗尽分支
|
||||
redis.execute((RedisCallback<Object>) conn -> conn.streamCommands()
|
||||
.xClaim(stream.getBytes(StandardCharsets.UTF_8), "g1", "seed2", Duration.ZERO, staleId));
|
||||
|
||||
// 5. 启动订阅 tag=A 的 driver，等 reclaim 处理这条 tag=B 的残留
|
||||
BlockingQueue<MqMessage> exhausted = new LinkedBlockingQueue<>();
|
||||
BlockingQueue<Object> delivered = new LinkedBlockingQueue<>();
|
||||
driver.subscribe(MqSubscription.builder()
|
||||
.topic(topic).group("g1").tag("A").payloadType(String.class)
|
||||
.concurrency(1).batchSize(1).maxRetry(1)
|
||||
.listener(m -> delivered.add(m.getPayload()))
|
||||
.retryExhaustedHandler(exhausted::add)
|
||||
.build());
|
||||
|
||||
assertNull(exhausted.poll(3, TimeUnit.SECONDS), "tag 不匹配的消息不得以本组名义耗尽落库");
|
||||
assertTrue(delivered.isEmpty(), "tag 不匹配的消息不得进业务方法");
|
||||
PendingMessagesSummary summary = redis.opsForStream().pending(stream, "g1");
|
||||
assertEquals(0L, summary.getTotalPendingMessages(), "应被 ACK 丢弃，PEL 清空");
|
||||
driver.stop();
|
||||
}
|
||||
|
||||
private static StringRedisTemplate newRedisOrSkip() {
|
||||
try {
|
||||
LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);
|
||||
cf.afterPropertiesSet();
|
||||
StringRedisTemplate t = new StringRedisTemplate(cf);
|
||||
t.afterPropertiesSet();
|
||||
t.hasKey("ping"); // 触发真实连接
|
||||
return t;
|
||||
} catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,181 @@
|
||||
package com.njcn.mq.driver.redis;
|
||||
|
||||
import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
|
||||
import com.njcn.middle.stream.support.StreamKeyBuilder;
|
||||
import com.njcn.mq.spi.MqSubscription;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.springframework.data.redis.connection.RedisConnection;
|
||||
import org.springframework.data.redis.core.RedisCallback;
|
||||
import org.springframework.data.redis.core.StreamOperations;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* 订阅 stream 的定时裁剪（XTRIM MAXLEN ~）。
|
||||
*
|
||||
* <p>底层中间件的 trim 任务只认自己注册的 handler，mq-starter 不走那套——
|
||||
* 所以 driver 必须自己裁订阅过的 stream，否则只经 mq-starter 收发的 topic 无界增长。
|
||||
* 配置复用中间件的 {@code redis-stream.trim.*}（maxlen 默认 10 万 / 周期默认 60s），默认开启。
|
||||
*/
|
||||
class RedisStreamMqDriverTrimTest {
|
||||
|
||||
/** 记录 connection.execute 收到的每次调用：[command, arg1, arg2, ...]（args 按 UTF-8 还原成字符串）。 */
|
||||
private final List<List<String>> executed = Collections.synchronizedList(new ArrayList<>());
|
||||
private StringRedisTemplate redis;
|
||||
|
||||
@BeforeEach
|
||||
void mockRedis() {
|
||||
redis = mock(StringRedisTemplate.class);
|
||||
RedisConnection conn = recordingConnection(null);
|
||||
when(redis.execute(any(RedisCallback.class))).thenAnswer(inv ->
|
||||
((RedisCallback<?>) inv.getArgument(0)).doInRedis(conn));
|
||||
}
|
||||
|
||||
/**
|
||||
* 记录 execute 调用的 RedisConnection——用默认 Answer 而非参数匹配，
|
||||
* 因为 execute(String, byte[]...) 是 varargs，这版 Mockito 的 any(byte[][].class) 匹配不上。
|
||||
*/
|
||||
private RedisConnection recordingConnection(String failStream) {
|
||||
return mock(RedisConnection.class, inv -> {
|
||||
if ("toString".equals(inv.getMethod().getName())) return "mock-conn";
|
||||
if (!"execute".equals(inv.getMethod().getName())) return null;
|
||||
List<String> call = new ArrayList<>();
|
||||
for (Object arg : inv.getArguments()) {
|
||||
if (arg instanceof byte[]) call.add(new String((byte[]) arg, StandardCharsets.UTF_8));
|
||||
else if (arg instanceof byte[][]) {
|
||||
for (byte[] b : (byte[][]) arg) call.add(new String(b, StandardCharsets.UTF_8));
|
||||
} else call.add(String.valueOf(arg));
|
||||
}
|
||||
if (failStream != null && call.size() > 1 && failStream.equals(call.get(1))) {
|
||||
throw new IllegalStateException("boom");
|
||||
}
|
||||
executed.add(call);
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
private static RedisStreamProperties propsWithTrim(long maxlen, long intervalMs) {
|
||||
RedisStreamProperties props = new RedisStreamProperties();
|
||||
props.getTrim().setMaxlen(maxlen);
|
||||
props.getTrim().setIntervalMs(intervalMs);
|
||||
props.setConsumerName("test-consumer");
|
||||
return props;
|
||||
}
|
||||
|
||||
private List<List<String>> xtrims() {
|
||||
synchronized (executed) {
|
||||
return executed.stream().filter(c -> "XTRIM".equals(c.get(0))).collect(Collectors.toList());
|
||||
}
|
||||
}
|
||||
|
||||
// ---- RedisStreamTrimScheduler 单元 ----
|
||||
|
||||
@Test
|
||||
void trimAll_issuesApproximateXtrimForEachRegisteredStream() {
|
||||
RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(50000, 60000));
|
||||
trim.register("order_dev");
|
||||
trim.register("pay_dev");
|
||||
|
||||
trim.trimAll();
|
||||
|
||||
List<List<String>> calls = xtrims();
|
||||
assertThat(calls).hasSize(2);
|
||||
assertThat(calls).allSatisfy(c -> {
|
||||
assertThat(c.subList(2, 5)).as("XTRIM <stream> MAXLEN ~ <maxlen>")
|
||||
.containsExactly("MAXLEN", "~", "50000");
|
||||
});
|
||||
assertThat(calls.stream().map(c -> c.get(1)))
|
||||
.containsExactlyInAnyOrder("order_dev", "pay_dev");
|
||||
trim.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
void register_sameStreamTwice_trimsOnlyOnce() {
|
||||
RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(100, 60000));
|
||||
trim.register("order_dev");
|
||||
trim.register("order_dev");
|
||||
|
||||
trim.trimAll();
|
||||
|
||||
assertThat(xtrims()).hasSize(1);
|
||||
trim.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
void maxlenNotPositive_treatedAsDisabled_neverTrims() {
|
||||
RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(0, 60000));
|
||||
trim.register("order_dev");
|
||||
|
||||
trim.trimAll();
|
||||
|
||||
assertThat(xtrims()).as("maxlen<=0 视为关闭，XTRIM 0 会清空整条 stream，绝不能发出").isEmpty();
|
||||
trim.stop();
|
||||
}
|
||||
|
||||
@Test
|
||||
void oneStreamFails_othersStillTrimmed() {
|
||||
RedisConnection conn = recordingConnection("bad_dev");
|
||||
when(redis.execute(any(RedisCallback.class))).thenAnswer(inv ->
|
||||
((RedisCallback<?>) inv.getArgument(0)).doInRedis(conn));
|
||||
|
||||
RedisStreamTrimScheduler trim = new RedisStreamTrimScheduler(redis, propsWithTrim(100, 60000));
|
||||
trim.register("bad_dev");
|
||||
trim.register("good_dev");
|
||||
|
||||
trim.trimAll(); // bad_dev 抛异常不得中断 good_dev
|
||||
|
||||
List<List<String>> calls = xtrims();
|
||||
assertThat(calls).hasSize(1);
|
||||
assertThat(calls.get(0).get(1)).isEqualTo("good_dev");
|
||||
trim.stop();
|
||||
}
|
||||
|
||||
// ---- driver 接线：订阅登记 + 周期触发 + 停机停止 ----
|
||||
|
||||
@Test
|
||||
void subscribe_startsPeriodicTrimOfSubscribedStream_andStopHaltsIt() throws Exception {
|
||||
RedisStreamProperties props = propsWithTrim(1234, 50); // 50ms 一裁，便于观察
|
||||
StreamOperations<String, Object, Object> streamOps = mock(StreamOperations.class);
|
||||
when(redis.opsForStream()).thenReturn(streamOps);
|
||||
when(streamOps.read(any(org.springframework.data.redis.connection.stream.Consumer.class),
|
||||
any(), any())).thenAnswer(inv -> { Thread.sleep(20); return Collections.emptyList(); });
|
||||
|
||||
RedisStreamMqDriver driver = new RedisStreamMqDriver(
|
||||
redis, new StreamKeyBuilder(true, "dev"), props);
|
||||
driver.subscribe(MqSubscription.builder()
|
||||
.topic("t-order").group("g1").tag("*").maxRetry(3)
|
||||
.payloadType(String.class).listener(msg -> { }).build());
|
||||
|
||||
List<List<String>> calls = awaitXtrim(2000);
|
||||
assertThat(calls).as("订阅后应周期性裁剪").isNotEmpty();
|
||||
assertThat(calls.get(0).get(1)).as("裁的是拼了环境后缀的实际 stream 名").isEqualTo("t-order_dev");
|
||||
assertThat(calls.get(0).subList(2, 5)).containsExactly("MAXLEN", "~", "1234");
|
||||
|
||||
driver.stop();
|
||||
Thread.sleep(120); // 让在途 tick 收尾
|
||||
int settled = xtrims().size();
|
||||
Thread.sleep(250);
|
||||
assertThat(xtrims()).as("stop 后不得再裁").hasSize(settled);
|
||||
}
|
||||
|
||||
private List<List<String>> awaitXtrim(long timeoutMs) throws InterruptedException {
|
||||
long deadline = System.currentTimeMillis() + timeoutMs;
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
List<List<String>> calls = xtrims();
|
||||
if (!calls.isEmpty()) return calls;
|
||||
Thread.sleep(20);
|
||||
}
|
||||
return xtrims();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user