From 74f68c9680c7c5d87635b8c5e8c231e2b1b3ae3b Mon Sep 17 00:00:00 2001 From: root Date: Wed, 24 Jun 2026 20:55:12 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20Redis=20Stream=20Spring=20Boot=20starte?= =?UTF-8?q?r=20=E9=A6=96=E6=AC=A1=E6=8F=90=E4=BA=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 封装 Redis Streams 收发,零业务依赖。含自动装配、发送模板、监听容器、 autoclaim 崩溃恢复+死信淘汰、stream 定时裁剪、gzip 编解码、环境隔离。 本次随提交的改进: - per-handler 重试次数:getMaxRetryTimes() 覆写生效,全局 redis-stream.max-retry 兜底 - 死信回调传入反序列化后的业务消息(saveExceptionFromFields) - pom 升级 maven-surefire-plugin 至 2.22.2,使 JUnit 5 测试真正执行 - 集成测试连接信息改为 -D 注入(默认 localhost:6379),凭据不入代码 - lombok 改为 optional,不污染下游依赖 - 新增 README 使用文档 Co-Authored-By: Claude Opus 4.8 (1M context) --- .gitignore | 32 +++ README.md | 257 ++++++++++++++++++ pom.xml | 129 +++++++++ .../RedisStreamAutoConfiguration.java | 78 ++++++ .../autoconfig/RedisStreamProperties.java | 53 ++++ .../middle/stream/codec/MessageCodec.java | 92 +++++++ .../constant/StreamMessageConstant.java | 43 +++ .../stream/container/StreamAutoClaimer.java | 227 ++++++++++++++++ .../container/StreamListenerContainer.java | 172 ++++++++++++ .../stream/domain/StreamBaseMessage.java | 35 +++ .../middle/stream/handler/ConsumeResult.java | 14 + .../handler/EnhanceStreamConsumerHandler.java | 134 +++++++++ .../stream/support/StreamKeyBuilder.java | 33 +++ .../stream/support/StreamTrimScheduler.java | 102 +++++++ .../template/RedisStreamEnhanceTemplate.java | 56 ++++ src/main/resources/META-INF/spring.factories | 2 + .../autoconfig/RedisStreamPropertiesTest.java | 37 +++ .../middle/stream/codec/MessageCodecTest.java | 68 +++++ .../StreamAutoClaimerMaxRetryTest.java | 96 +++++++ .../stream/domain/StreamBaseMessageTest.java | 33 +++ .../EnhanceStreamConsumerHandlerTest.java | 223 +++++++++++++++ .../StreamAutoClaimDeadLetterTest.java | 197 ++++++++++++++ .../integration/StreamRoundTripTest.java | 130 +++++++++ .../stream/support/StreamKeyBuilderTest.java | 29 ++ .../RedisStreamEnhanceTemplateTest.java | 93 +++++++ 25 files changed, 2365 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 pom.xml create mode 100644 src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java create mode 100644 src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamProperties.java create mode 100644 src/main/java/com/njcn/middle/stream/codec/MessageCodec.java create mode 100644 src/main/java/com/njcn/middle/stream/constant/StreamMessageConstant.java create mode 100644 src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java create mode 100644 src/main/java/com/njcn/middle/stream/container/StreamListenerContainer.java create mode 100644 src/main/java/com/njcn/middle/stream/domain/StreamBaseMessage.java create mode 100644 src/main/java/com/njcn/middle/stream/handler/ConsumeResult.java create mode 100644 src/main/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandler.java create mode 100644 src/main/java/com/njcn/middle/stream/support/StreamKeyBuilder.java create mode 100644 src/main/java/com/njcn/middle/stream/support/StreamTrimScheduler.java create mode 100644 src/main/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplate.java create mode 100644 src/main/resources/META-INF/spring.factories create mode 100644 src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamPropertiesTest.java create mode 100644 src/test/java/com/njcn/middle/stream/codec/MessageCodecTest.java create mode 100644 src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerMaxRetryTest.java create mode 100644 src/test/java/com/njcn/middle/stream/domain/StreamBaseMessageTest.java create mode 100644 src/test/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandlerTest.java create mode 100644 src/test/java/com/njcn/middle/stream/integration/StreamAutoClaimDeadLetterTest.java create mode 100644 src/test/java/com/njcn/middle/stream/integration/StreamRoundTripTest.java create mode 100644 src/test/java/com/njcn/middle/stream/support/StreamKeyBuilderTest.java create mode 100644 src/test/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplateTest.java diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..733b6b7 --- /dev/null +++ b/.gitignore @@ -0,0 +1,32 @@ +# Compiled class file +*.class + +# Log file +*.log +logs/ + +# Package Files +*.jar +*.war +*.ear +*.tar.gz +*.rar + +# Maven +target/ + +# IDE +*.iml +*.ipr +*.iws +.idea/ +.classpath +.project +.settings/ +.factorypath + +# virtual machine crash logs +hs_err_pid* +replay_pid* + +.DS_Store diff --git a/README.md b/README.md new file mode 100644 index 0000000..4bb92eb --- /dev/null +++ b/README.md @@ -0,0 +1,257 @@ +# redis-stream-springboot-starter + +基于 Redis Streams 的消息收发 SDK,对标 `rocket-mq-springboot-starter`,**零业务依赖**、开箱即用。 + +引入依赖即自动装配:配好 Redis 连接,注入 `RedisStreamEnhanceTemplate` 即可发送;注册一个 `EnhanceStreamConsumerHandler` Bean 即可消费。框架负责消费组创建、阻塞拉取、ACK、崩溃恢复(autoclaim)、死信淘汰、Stream 定时裁剪。 + +- Spring Boot `2.3.12.RELEASE` / JDK `1.8` +- 底层 `spring-data-redis 2.3.x` + Lettuce +- 消息体可选 gzip 压缩;XADD 字段契约固定为 `key/tag/enc/body/ts`(可与 C++ 等异构端互通) + +--- + +## 一、引入依赖 + +```xml + + com.njcn + redis-stream-springboot-starter + 1.0.0-SNAPSHOT + +``` + +制品发布在内网 Nexus,若拉不到依赖请确认 `settings.xml` / `pom.xml` 配置了仓库: + +```xml + + nexus-snapshots + http://192.168.1.22:8001/nexus/content/repositories/snapshots/ + +``` + +> 无需 `@EnableXxx` 注解。starter 通过 `spring.factories` 自动装配,且在 `RedisAutoConfiguration` 之后生效(确保 `StringRedisTemplate` 已就绪)。 + +--- + +## 二、最小配置 + +连接信息走标准 Spring Redis 配置,starter 自身配置前缀为 `redis-stream`: + +```yaml +spring: + redis: + host: 127.0.0.1 + port: 6379 + # password: ... + # database: 0 + +redis-stream: + # 以下均为可选,默认值见「配置项」一节 + read-count: 10 + block-ms: 2000 + max-retry: 3 +``` + +--- + +## 三、定义消息 + +业务消息继承 `StreamBaseMessage`(starter 自带,**不依赖任何业务 api**): + +```java +import com.njcn.middle.stream.domain.StreamBaseMessage; +import lombok.Data; +import lombok.EqualsAndHashCode; + +@Data +@EqualsAndHashCode(callSuper = true) +public class OrderMessage extends StreamBaseMessage { + private Long orderId; + private String status; +} +``` + +基类已内置以下字段,业务无需自己维护: + +| 字段 | 默认值 | 说明 | +|---|---|---| +| `key` | 雪花 id | 消息业务 key,全局单例 Snowflake 生成,保证唯一 | +| `tag` | `""` | 下行路由 / 过滤标识 | +| `source` | `""` | 消息来源标识 | +| `sendTime` | `now()` | 发送时间 | +| `retryTimes` | `0` | 预留字段(重试计数实际由 Redis PEL 维护,见下文) | + +--- + +## 四、发送消息 + +注入 `RedisStreamEnhanceTemplate`,调用 `send(topic, message, compress)`: + +```java +import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.stereotype.Service; + +@Service +public class OrderProducer { + + private final RedisStreamEnhanceTemplate streamTemplate; + + public OrderProducer(RedisStreamEnhanceTemplate streamTemplate) { + this.streamTemplate = streamTemplate; + } + + public void sendOrder(Long orderId) { + OrderMessage msg = new OrderMessage(); + msg.setOrderId(orderId); + msg.setStatus("CREATED"); + // 需要按 tag 路由时,发送前先 setTag(...) + msg.setTag("node-1"); + + // 第三参 compress=true 时对 body 做 gzip+base64;false 为明文 json + RecordId id = streamTemplate.send("order-topic", msg, false); + } +} +``` + +要点: + +- `topic` 会经 `StreamKeyBuilder` 转为实际 stream 名(开启环境隔离时追加 `_env` 后缀)。 +- `key` / `tag` **从 message 对象取**,不单独传参。需要自定义 tag 时,**发送前** `message.setTag(...)`。 +- 返回 `RecordId` 即 XADD 生成的消息 id。 + +--- + +## 五、消费消息 + +继承 `EnhanceStreamConsumerHandler` 并**注册为 Spring Bean**,框架会为每个 handler 起一条消费线程: + +```java +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import org.springframework.stereotype.Component; + +@Component +public class OrderConsumer extends EnhanceStreamConsumerHandler { + + @Override + public String topic() { + return "order-topic"; // 与发送方一致 + } + + @Override + public String group() { + return "order-consumer-group"; // 消费组名,启动时自动 XGROUP CREATE + } + + @Override + public Class messageType() { + return OrderMessage.class; // 用于反序列化 + } + + @Override + public void handleMessage(OrderMessage message) throws Exception { + // 业务处理。抛异常时按 isRetry() 决定 RETRY 或落异常 + System.out.println("consume order = " + message.getOrderId()); + } + + @Override + public boolean isRetry() { + return true; // 失败是否重试(留 PEL 由 autoclaim 重投) + } + + @Override + public boolean throwException() { + return false; // Redis 模式下为 no-op,固定返回 false 即可 + } +} +``` + +### 必须实现的方法 + +| 方法 | 说明 | +|---|---| +| `topic()` | 监听的 topic | +| `group()` | 消费组名(自动幂等建组) | +| `messageType()` | 消息类型,用于反序列化 | +| `handleMessage(T)` | 业务处理逻辑 | +| `isRetry()` | 处理失败是否重试 | +| `throwException()` | **Redis 模式 no-op**,返回 `false` 即可(仅为与 rocket starter 共用接口而保留) | + +### 可选覆写的钩子 + +| 方法 | 默认 | 说明 | +|---|---|---| +| `filter(T)` | `false` | 返回 `true` 则直接 ACK 丢弃,不进入 `handleMessage` | +| `consumeSuccess(T)` | 空 | 消费成功回调 | +| `saveExceptionMsgLog(T message, body, e)` | 空 | 异常落库 / 告警回调(毒消息 / 不重试失败 / 死信触发);**`message` 可能为 `null`,实现需判空** | +| `getMaxRetryTimes()` | 沿用全局 | 覆写为正整数即**只对本 handler** 单独设定最大投递次数,优先级高于全局 `redis-stream.max-retry` | + +> 💡 不覆写 `getMaxRetryTimes()` 时,该 handler 沿用全局配置 `redis-stream.max-retry`;需要某个 handler 单独控制重试次数时,覆写它返回一个正整数即可(如重投较多的关键 topic 返回 `10`)。 +> +> ⚠️ `saveExceptionMsgLog` 的 `message` 入参:**不重试失败**与**死信**(autoclaim 超 `max-retry`)场景下为**反序列化后的业务对象**;仅**毒消息**(`body` 本身无法解析)场景下为 `null`(此时只能用原始 `body`)。回调实现务必对 `message == null` 做判空,避免 NPE。 + +--- + +## 六、消费语义:成功 / 重试 / 毒消息 / 死信 + +一条消息进入 `handleMessage` 后的处理结果: + +``` +解码(enc,body) → 反序列化 → filter → handleMessage → consumeSuccess → XACK +``` + +| 场景 | 结果 | +|---|---| +| 处理成功 | `XACK` 确认 | +| `filter()` 返回 true | `XACK` 丢弃,不处理 | +| 解码 / 反序列化失败(**毒消息**) | `saveExceptionMsgLog` + `XACK`,**不卡队列** | +| `handleMessage` 抛异常 且 `isRetry()=true` | **RETRY**:不确认,留 PEL | +| `handleMessage` 抛异常 且 `isRetry()=false` | `saveExceptionMsgLog` + `XACK` | + +**重试与死信(崩溃恢复)** 由后台 `StreamAutoClaimer` 兜底: + +1. RETRY 的消息留在 PEL(Pending Entries List)中未确认。 +2. autoclaim 周期性扫描 PEL,对 idle 超过 `autoclaim.min-idle-ms` 的消息执行 `XCLAIM` 认领并重投。 +3. 每次 `XCLAIM` 使投递计数 +1;当投递次数达到 `max-retry`(含首投)仍失败 → 触发 `saveExceptionMsgLog`(可拿到反序列化后的业务消息,异常信息含 `max retry exceeded`)+ 强制 `XACK`,**作为死信淘汰**,避免无限滞留。 + +> 即:`max-retry=3` ⇒ 最多投递 3 次(首投 1 次 + autoclaim 重投 2 次),第 3 次仍失败即落死信。 +> 单个 handler 可覆写 `getMaxRetryTimes()` 返回正整数,**对该 handler 单独设定**最大投递次数,优先级高于全局 `redis-stream.max-retry`。 +> 实例宕机导致的 PEL 残留消息,也会被其它存活实例的 autoclaim 接管重投。 + +--- + +## 七、配置项 + +| 配置项 | 默认值 | 说明 | +|---|---|---| +| `redis-stream.env-isolation` | `false` | 是否按环境隔离 stream 名 | +| `redis-stream.env` | `""` | 环境标识,隔离开启且非空时作为 stream 名后缀(`topic_env`) | +| `redis-stream.consumer-name` | `""` | 消费者名;为空时自动生成唯一名(**见下方注意事项**) | +| `redis-stream.read-count` | `10` | 单次 `XREADGROUP` 拉取条数 | +| `redis-stream.block-ms` | `2000` | `XREADGROUP` 阻塞毫秒 | +| `redis-stream.max-retry` | `3` | **全局**最大投递次数(含首投),超限落死信;可被 handler 的 `getMaxRetryTimes()` 单独覆写 | +| `redis-stream.autoclaim.min-idle-ms` | `60000` | 消息 idle 多久后可被认领重投 | +| `redis-stream.autoclaim.interval-ms` | `30000` | autoclaim 扫描周期 | +| `redis-stream.trim.interval-ms` | `60000` | `XTRIM` 执行周期 | +| `redis-stream.trim.maxlen` | `100000` | 近似保留的最大 stream 长度(`MAXLEN ~`) | + +--- + +## 八、注意事项 + +- **多实例部署不要写死 `consumer-name`**。留空时框架会为每个实例生成唯一消费者名(`consumer-xxx` / `claimer-xxx`),符合消费组负载均衡模型。若多个实例配置了相同的固定 `consumer-name`,它们会被视为同一个消费者而互相争抢,破坏负载均衡 —— `consumer-name` 仅建议用于单实例或本地调试。 +- **环境隔离**:多环境(dev/test/prod)共用一套 Redis 时,开启 `env-isolation` 并设置 `env`,各环境 stream 名互不串扰。**收发双方必须配置一致**,否则 topic 对不上。 +- **压缩契约**:仅 `body` 字段可压缩。`compress=true` ⇒ `enc=gzip`、`body=base64(gzip(json))`;否则 `enc=plain`、`body` 为明文 json。消费侧仅按 `enc` 解码,与 topic 无关,可与 C++ 等异构端互通。 +- **Stream 裁剪**:框架按 `trim.maxlen` 近似裁剪 stream 防止无界增长。请确保 `maxlen` 足够大,避免重试中的消息在达到 `max-retry` 前被裁掉(被裁掉的 PEL 消息会被静默强制 ACK)。 +- **无 handler 时容器空转**:未注册任何 `EnhanceStreamConsumerHandler` Bean 时,监听/认领/裁剪线程均不启动,仅 `RedisStreamEnhanceTemplate` 可用于纯发送场景。 + +--- + +## 九、本地验证(可选) + +集成测试需要本地 Redis;不可达时整类自动 skip(不影响 `mvn test` 全绿): + +```bash +docker run -d -p 6379:6379 redis:6.2 +mvn test +``` diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..8046186 --- /dev/null +++ b/pom.xml @@ -0,0 +1,129 @@ + + + 4.0.0 + com.njcn + redis-stream-springboot-starter + 1.0.0-SNAPSHOT + jar + redis stream中间件二次开发sdk + 封装 Redis Streams 收发，对标 rocket-mq-springboot-starter，零业务依赖 + + + + nexus-releases + Nexus Release Repository + http://192.168.1.22:8001/nexus/content/repositories/releases/ + + + nexus-snapshots + Nexus Snapshot Repository + http://192.168.1.22:8001/nexus/content/repositories/snapshots/ + + + + + 1.8 + 8 + 8 + UTF-8 + UTF-8 + 2.3.12.RELEASE + + 2.3.12.RELEASE + 1.2.83 + 5.7.9 + 1.18.18 + + + + + + org.springframework.boot + spring-boot-starter-data-redis + ${springboot.version} + + + + org.springframework.boot + spring-boot-autoconfigure + ${autoconfigure.version} + + + + org.springframework.boot + spring-boot-configuration-processor + true + ${springboot.version} + + + + com.alibaba + fastjson + ${fastjson.version} + + + + cn.hutool + hutool-all + ${hutool.version} + + + + org.projectlombok + lombok + ${lombok.version} + true + + + + org.springframework.boot + spring-boot-starter-test + ${springboot.version} + test + + + + + + + org.apache.maven.plugins + maven-compiler-plugin + 3.8.1 + + 1.8 + 1.8 + UTF-8 + + + + + org.apache.maven.plugins + maven-surefire-plugin + 2.22.2 + + + org.apache.maven.plugins + maven-source-plugin + 2.2.1 + + + attach-sources + + jar-no-fork + + + + + + + + diff --git a/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java b/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java new file mode 100644 index 0000000..677ba3f --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java @@ -0,0 +1,78 @@ +package com.njcn.middle.stream.autoconfig; + +import com.njcn.middle.stream.container.StreamAutoClaimer; +import com.njcn.middle.stream.container.StreamListenerContainer; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import com.njcn.middle.stream.support.StreamTrimScheduler; +import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.util.List; +import java.util.stream.Collectors; + +/** + * Redis Stream starter 自动装配。 + * + *

handler 列表用 {@link ObjectProvider} 收集——Phase 1 缺省为空（容器空转）， + * Phase 2 由 message-boot 注入 Processor 即生效。装配在 {@link RedisAutoConfiguration} 之后， + * 确保 {@link StringRedisTemplate} 已就绪。 + */ +@Configuration +@ConditionalOnClass(StringRedisTemplate.class) +@AutoConfigureAfter(RedisAutoConfiguration.class) +@EnableConfigurationProperties(RedisStreamProperties.class) +public class RedisStreamAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public StreamKeyBuilder streamKeyBuilder(RedisStreamProperties props) { + return new StreamKeyBuilder(props.isEnvIsolation(), props.getEnv()); + } + + @Bean + @ConditionalOnMissingBean + public RedisStreamEnhanceTemplate redisStreamEnhanceTemplate(StringRedisTemplate redis, + StreamKeyBuilder streamKeyBuilder) { + return new RedisStreamEnhanceTemplate(redis, streamKeyBuilder); + } + + @Bean + @ConditionalOnMissingBean + public StreamListenerContainer streamListenerContainer(StringRedisTemplate redis, + StreamKeyBuilder streamKeyBuilder, + RedisStreamProperties props, + ObjectProvider> handlers) { + return new StreamListenerContainer(redis, streamKeyBuilder, props, collect(handlers)); + } + + @Bean + @ConditionalOnMissingBean + public StreamAutoClaimer streamAutoClaimer(StringRedisTemplate redis, + StreamKeyBuilder streamKeyBuilder, + RedisStreamProperties props, + ObjectProvider> handlers) { + return new StreamAutoClaimer(redis, streamKeyBuilder, props, collect(handlers)); + } + + @Bean + @ConditionalOnMissingBean + public StreamTrimScheduler streamTrimScheduler(StringRedisTemplate redis, + StreamKeyBuilder streamKeyBuilder, + RedisStreamProperties props, + ObjectProvider> handlers) { + return new StreamTrimScheduler(redis, streamKeyBuilder, props, collect(handlers)); + } + + private static List> collect(ObjectProvider> handlers) { + return handlers.orderedStream().collect(Collectors.toList()); + } +} diff --git a/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamProperties.java b/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamProperties.java new file mode 100644 index 0000000..1d1d051 --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamProperties.java @@ -0,0 +1,53 @@ +package com.njcn.middle.stream.autoconfig; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +/** + * Redis Stream starter 配置属性，前缀 {@code redis-stream}。 + * + *

env / env-isolation 归 starter（非 message.mq.*），整体开关 message.mq.type 留 message-boot。 + */ +@Data +@ConfigurationProperties("redis-stream") +public class RedisStreamProperties { + + /** 是否按环境隔离 stream 名（追加 {@code _env} 后缀）。 */ + private boolean envIsolation = false; + + /** 环境标识（隔离开启时作为 stream 名后缀）。 */ + private String env = ""; + + /** 消费者名；为空时由容器生成。 */ + private String consumerName = ""; + + /** 单次 XREADGROUP 拉取条数。 */ + private int readCount = 10; + + /** XREADGROUP 阻塞毫秒。 */ + private long blockMs = 2000; + + /** 最大重试次数（autoclaim 维护）。 */ + private int maxRetry = 3; + + private Autoclaim autoclaim = new Autoclaim(); + private Trim trim = new Trim(); + + /** 崩溃恢复：PEL 中 idle 超时消息的自动认领。 */ + @Data + public static class Autoclaim { + /** 消息 idle 多久（ms）后可被其它消费者认领。 */ + private long minIdleMs = 60000; + /** autoclaim 扫描周期（ms）。 */ + private long intervalMs = 30000; + } + + /** 定时裁剪 stream 长度。 */ + @Data + public static class Trim { + /** XTRIM 执行周期（ms）。 */ + private long intervalMs = 60000; + /** 近似保留的最大长度（MAXLEN ~）。 */ + private long maxlen = 100000; + } +} diff --git a/src/main/java/com/njcn/middle/stream/codec/MessageCodec.java b/src/main/java/com/njcn/middle/stream/codec/MessageCodec.java new file mode 100644 index 0000000..1a98e0d --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/codec/MessageCodec.java @@ -0,0 +1,92 @@ +package com.njcn.middle.stream.codec; + +import com.njcn.middle.stream.constant.StreamMessageConstant; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.zip.GZIPInputStream; +import java.util.zip.GZIPOutputStream; + +/** + * 消息体编解码器。 + * + *

XADD 字段契约对死 C++：仅 {@code body} 可压缩。 + * {@code enc=gzip} → base64(gzip(json))（RFC1952 gzip 头 + 无换行 base64）； + * {@code enc=plain} → 原 json 明文。Java 1.8：{@link Base64}（默认无换行）+ {@link GZIPOutputStream}。 + */ +public final class MessageCodec { + + private MessageCodec() { + } + + /** + * 编码 body。 + * + * @param json 原始 json + * @param compress 是否压缩；true → gzip+base64，false → 明文 + * @return 编码结果（enc + body） + */ + public static Encoded encodeBody(String json, boolean compress) { + if (!compress) { + return new Encoded(StreamMessageConstant.ENC_PLAIN, json); + } + byte[] raw = json.getBytes(StandardCharsets.UTF_8); + ByteArrayOutputStream bos = new ByteArrayOutputStream(Math.max(32, raw.length / 2)); + try (GZIPOutputStream gzip = new GZIPOutputStream(bos)) { + gzip.write(raw); + } catch (IOException ex) { + throw new IllegalStateException("gzip encode failed", ex); + } + // Base64.getEncoder() 为基础编码器，不插入换行（区别于 getMimeEncoder） + String body = Base64.getEncoder().encodeToString(bos.toByteArray()); + return new Encoded(StreamMessageConstant.ENC_GZIP, body); + } + + /** + * 解码 body。仅按 {@code enc} 解码，不 hardcode topic；未知/明文一律按原文返回。 + * + * @param enc 编码方式 + * @param body body 内容 + * @return 原始 json + */ + public static String decodeBody(String enc, String body) { + if (!StreamMessageConstant.ENC_GZIP.equals(enc)) { + // plain 及未知 enc：兜底按明文返回 + return body; + } + byte[] compressed = Base64.getDecoder().decode(body); + ByteArrayOutputStream bos = new ByteArrayOutputStream(compressed.length * 2); + try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(compressed))) { + byte[] buf = new byte[1024]; + int n; + while ((n = gzip.read(buf)) != -1) { + bos.write(buf, 0, n); + } + } catch (IOException ex) { + throw new IllegalStateException("gzip decode failed", ex); + } + return new String(bos.toByteArray(), StandardCharsets.UTF_8); + } + + /** 编码结果值对象。 */ + public static class Encoded { + private final String enc; + private final String body; + + public Encoded(String enc, String body) { + this.enc = enc; + this.body = body; + } + + public String getEnc() { + return enc; + } + + public String getBody() { + return body; + } + } +} diff --git a/src/main/java/com/njcn/middle/stream/constant/StreamMessageConstant.java b/src/main/java/com/njcn/middle/stream/constant/StreamMessageConstant.java new file mode 100644 index 0000000..3875da0 --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/constant/StreamMessageConstant.java @@ -0,0 +1,43 @@ +package com.njcn.middle.stream.constant; + +/** + * Redis Stream 消息相关常量。 + * + *

XADD 字段契约对死 C++ 端：每条消息固定 5 个字段 {@code key/tag/enc/body/ts}， + * 仅 {@code body} 可压缩（{@code enc=gzip} → base64(gzip(json))），其余明文。 + */ +public final class StreamMessageConstant { + + private StreamMessageConstant() { + } + + /** 单条消息标识。 */ + public static final String IDENTITY_SINGLE = "single"; + + /** 重试消息标识。 */ + public static final String IDENTITY_RETRY = "retry"; + + /** 重试相关 key 前缀。 */ + public static final String RETRY_PREFIX = "retry_"; + + /** XADD 字段名：消息业务 key。 */ + public static final String F_KEY = "key"; + + /** XADD 字段名：消息 tag（下行路由/过滤用）。 */ + public static final String F_TAG = "tag"; + + /** XADD 字段名：body 编码方式（{@link #ENC_GZIP} / {@link #ENC_PLAIN}）。 */ + public static final String F_ENC = "enc"; + + /** XADD 字段名：消息体（json 明文或 base64(gzip(json))）。 */ + public static final String F_BODY = "body"; + + /** XADD 字段名：发送时间戳（毫秒）。 */ + public static final String F_TS = "ts"; + + /** 编码方式：gzip 压缩后 base64。 */ + public static final String ENC_GZIP = "gzip"; + + /** 编码方式：明文 json。 */ + public static final String ENC_PLAIN = "plain"; +} diff --git a/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java b/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java new file mode 100644 index 0000000..89406f1 --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java @@ -0,0 +1,227 @@ +package com.njcn.middle.stream.container; + +import cn.hutool.core.util.IdUtil; +import cn.hutool.core.util.StrUtil; +import com.njcn.middle.stream.autoconfig.RedisStreamProperties; +import com.njcn.middle.stream.handler.ConsumeResult; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.SmartLifecycle; +import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessage; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * 崩溃恢复：自动认领 PEL 中长时间未确认（消费者宕机）的消息。 + * + *

spring-data-redis 2.3.x 无 {@code XAUTOCLAIM}（需 2.6+），故走 + * {@code pending} + 原生 {@code XCLAIM ... JUSTID} + {@code range} 取内容 + 重新 dispatch。 + * 属崩溃恢复兜底、非主链路（主链路重试由消费线程 RETRY 留 PEL 触发本认领）。 + * + *

投递次数超过 {@code maxRetry} 仍失败 → 落异常 + 强制 ACK，避免死信无限滞留。 + * 正确性由 C11 集成测试覆盖。 + */ +@Slf4j +public class StreamAutoClaimer implements SmartLifecycle { + + private final StringRedisTemplate redis; + private final StreamKeyBuilder keyBuilder; + private final RedisStreamProperties props; + private final List> handlers; + + private ScheduledExecutorService scheduler; + private volatile boolean running = false; + private String claimConsumer; + + public StreamAutoClaimer(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, + RedisStreamProperties props, + List> handlers) { + this.redis = redis; + this.keyBuilder = keyBuilder; + this.props = props; + this.handlers = handlers == null ? Collections.emptyList() : handlers; + } + + @Override + public synchronized void start() { + if (running) { + return; + } + running = true; + if (handlers.isEmpty()) { + return; + } + claimConsumer = resolveConsumerName(); + long interval = props.getAutoclaim().getIntervalMs(); + scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "redis-stream-autoclaim"); + t.setDaemon(true); + return t; + }); + scheduler.scheduleWithFixedDelay(this::claimAll, interval, interval, TimeUnit.MILLISECONDS); + log.info("[redis-stream] autoclaimer started, interval={}ms minIdle={}ms", + interval, props.getAutoclaim().getMinIdleMs()); + } + + @Override + public synchronized void stop() { + running = false; + if (scheduler != null) { + scheduler.shutdownNow(); + scheduler = null; + } + } + + @Override + public boolean isRunning() { + return running; + } + + private String resolveConsumerName() { + if (StrUtil.isNotBlank(props.getConsumerName())) { + return props.getConsumerName(); + } + return "claimer-" + IdUtil.fastSimpleUUID(); + } + + private void claimAll() { + long minIdle = props.getAutoclaim().getMinIdleMs(); + for (EnhanceStreamConsumerHandler handler : handlers) { + String stream = keyBuilder.buildStream(handler.topic()); + String group = handler.group(); + int maxRetry = resolveMaxRetry(handler); + try { + claimStream(handler, stream, group, minIdle, maxRetry); + } catch (Exception e) { + log.warn("[redis-stream] autoclaim failed stream={} group={}", stream, group, e); + } + } + } + + /** + * 解析本 handler 生效的最大投递次数:优先用 handler 自配的 + * {@link EnhanceStreamConsumerHandler#getMaxRetryTimes()}(正整数);未配 + * (返回 {@link EnhanceStreamConsumerHandler#INHERIT_GLOBAL_MAX_RETRY} 等 ≤0 值)时 + * 沿用全局 {@code redis-stream.max-retry}。 + */ + int resolveMaxRetry(EnhanceStreamConsumerHandler handler) { + int perHandler = handler.getMaxRetryTimes(); + return perHandler > 0 ? perHandler : props.getMaxRetry(); + } + + private void claimStream(EnhanceStreamConsumerHandler handler, String stream, String group, + long minIdle, int maxRetry) { + int readCount = props.getReadCount(); + // 按 id 游标翻页扫描整个 PEL，避免只取最旧 readCount 条导致长 PEL 下尾部消息饥饿。 + // XPENDING start 为包含语义（兼容 Redis 5.0），故每页首条会与上页游标重复，跳过即可。 + String cursor = null; + while (running) { + Range range = cursor == null + ? Range.unbounded() + : Range.rightUnbounded(Range.Bound.inclusive(cursor)); + int fetch = cursor == null ? readCount : readCount + 1; + PendingMessages pending = redis.opsForStream().pending(stream, group, range, fetch); + if (pending == null || pending.isEmpty()) { + return; + } + int fresh = 0; + String lastId = cursor; + for (PendingMessage pm : pending) { + String id = pm.getIdAsString(); + if (id.equals(cursor)) { + // 与上一页游标重复，已处理过，跳过 + continue; + } + fresh++; + lastId = id; + if (pm.getElapsedTimeSinceLastDelivery().toMillis() < minIdle) { + continue; + } + claimOne(handler, stream, group, minIdle, maxRetry, id, pm.getTotalDeliveryCount()); + } + if (fresh < readCount) { + // 本页新条目不足一批 → 已扫到 PEL 尾部 + return; + } + if (lastId == null || lastId.equals(cursor)) { + // 防御：游标无法推进时退出，避免死循环 + return; + } + cursor = lastId; + } + } + + /** + * 处理单条 PEL 消息：取内容 → 死信判定前置 → 认领重投。 + * + *

{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数；因 {@link #claim} 用不带 JUSTID + * 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。 + * {@code maxRetry} 语义 = 最多投递 maxRetry 次（含首投），第 maxRetry 次仍 RETRY 即落死信。 + * 死信判定前置于 dispatch：达上限直接落异常 + 强制 ACK，不再多跑一次业务副作用。 + */ + private void claimOne(EnhanceStreamConsumerHandler handler, String stream, String group, + long minIdle, int maxRetry, String id, long deliveries) { + List> records = + redis.opsForStream().range(stream, Range.closed(id, id)); + if (records == null || records.isEmpty()) { + // 内容已不存在（可能被 trim）→ 强制 ACK 清理 PEL + redis.opsForStream().acknowledge(stream, group, RecordId.of(id)); + return; + } + MapRecord rec = records.get(0); + Map fields = toStringMap(rec.getValue()); + + if (deliveries >= maxRetry) { + // 投递计数已达上限 → 落异常 + 强制 ACK（死信，不再无限滞留 PEL） + // 由 handler 内部反序列化后回调,使死信回调能拿到业务消息对象（解析失败兜底 null） + handler.saveExceptionFromFields(fields, + new IllegalStateException("max retry exceeded, deliveries=" + deliveries)); + redis.opsForStream().acknowledge(stream, group, rec.getId()); + return; + } + + // 认领到本 claimer（同时把投递计数 +1、重置 idle），再重投 + claim(stream, group, minIdle, id); + ConsumeResult result = handler.dispatchMessage(fields); + if (result == ConsumeResult.ACK) { + redis.opsForStream().acknowledge(stream, group, rec.getId()); + } + // 否则 RETRY 且未超限：留 PEL，下一轮（idle 到点后）再认领重试 + } + + /** + * {@code XCLAIM stream group consumer minIdle id}（不带 JUSTID）。 + * + *

不带 JUSTID 时 XCLAIM 会转移归属、重置 idle 并把投递计数 +1——后者是 maxRetry 死信判定的 + * 依据：listener 用 {@code XREADGROUP lastConsumed()} 永不重读 PEL，若此处用 JUSTID 则投递计数 + * 永久冻结在首投值 1，{@code deliveries >= maxRetry} 恒不成立、死信分支永不触发。 + */ + private void claim(String stream, String group, long minIdle, String id) { + redis.execute((RedisCallback) connection -> connection.execute("XCLAIM", + stream.getBytes(StandardCharsets.UTF_8), + group.getBytes(StandardCharsets.UTF_8), + claimConsumer.getBytes(StandardCharsets.UTF_8), + String.valueOf(minIdle).getBytes(StandardCharsets.UTF_8), + id.getBytes(StandardCharsets.UTF_8))); + } + + private static Map toStringMap(Map raw) { + Map m = new HashMap<>(Math.max(8, raw.size() * 2)); + raw.forEach((k, v) -> m.put(String.valueOf(k), v == null ? null : String.valueOf(v))); + return m; + } +} diff --git a/src/main/java/com/njcn/middle/stream/container/StreamListenerContainer.java b/src/main/java/com/njcn/middle/stream/container/StreamListenerContainer.java new file mode 100644 index 0000000..8fe4a18 --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/container/StreamListenerContainer.java @@ -0,0 +1,172 @@ +package com.njcn.middle.stream.container; + +import cn.hutool.core.util.IdUtil; +import cn.hutool.core.util.StrUtil; +import com.njcn.middle.stream.autoconfig.RedisStreamProperties; +import com.njcn.middle.stream.handler.ConsumeResult; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.SmartLifecycle; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.connection.stream.StreamReadOptions; +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.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Stream 监听容器：每个 handler 一条消费组消费线程。 + * + *

{@code XREADGROUP} 阻塞读 → {@link EnhanceStreamConsumerHandler#dispatchMessage} → + * 结果 ACK 则 {@code XACK}，RETRY 则不确认（留 PEL 由 autoclaim 兜）。 + * + *

无 handler 时容器空转（Phase 2 才注入 handler）。正确性由 C11 集成测试覆盖。 + */ +@Slf4j +public class StreamListenerContainer implements SmartLifecycle { + + private final StringRedisTemplate redis; + private final StreamKeyBuilder keyBuilder; + private final RedisStreamProperties props; + private final List> handlers; + + private final List workers = new ArrayList<>(); + private volatile boolean running = false; + + public StreamListenerContainer(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, + RedisStreamProperties props, + List> handlers) { + this.redis = redis; + this.keyBuilder = keyBuilder; + this.props = props; + this.handlers = handlers == null ? Collections.emptyList() : handlers; + } + + @Override + public synchronized void start() { + if (running) { + return; + } + running = true; + if (handlers.isEmpty()) { + log.info("[redis-stream] no handler registered, listener container idle"); + return; + } + String consumer = resolveConsumerName(); + for (EnhanceStreamConsumerHandler handler : handlers) { + String stream = keyBuilder.buildStream(handler.topic()); + String group = handler.group(); + ensureGroup(stream, group); + Thread t = new Thread(() -> loop(handler, stream, group, consumer), + "redis-stream-" + stream + "-" + group); + t.setDaemon(true); + workers.add(t); + t.start(); + } + log.info("[redis-stream] listener container started, {} listener(s), consumer={}", workers.size(), consumer); + } + + @Override + public synchronized void stop() { + running = false; + for (Thread t : workers) { + t.interrupt(); + } + workers.clear(); + } + + @Override + public boolean isRunning() { + return running; + } + + private String resolveConsumerName() { + if (StrUtil.isNotBlank(props.getConsumerName())) { + return props.getConsumerName(); + } + return "consumer-" + IdUtil.fastSimpleUUID(); + } + + /** {@code XGROUP CREATE stream group $ MKSTREAM}，吞 BUSYGROUP（组已存在）。 */ + private void ensureGroup(String stream, String group) { + try { + redis.execute((RedisCallback) connection -> connection.execute("XGROUP", + "CREATE".getBytes(StandardCharsets.UTF_8), + stream.getBytes(StandardCharsets.UTF_8), + group.getBytes(StandardCharsets.UTF_8), + "$".getBytes(StandardCharsets.UTF_8), + "MKSTREAM".getBytes(StandardCharsets.UTF_8))); + log.info("[redis-stream] group ensured stream={} group={}", stream, group); + } catch (RuntimeException e) { + if (isBusyGroup(e)) { + // 消费组已存在，幂等忽略 + return; + } + throw e; + } + } + + private void loop(EnhanceStreamConsumerHandler handler, String stream, String group, String consumer) { + while (running && !Thread.currentThread().isInterrupted()) { + try { + List> records = redis.opsForStream().read( + Consumer.from(group, consumer), + StreamReadOptions.empty() + .count(props.getReadCount()) + .block(Duration.ofMillis(props.getBlockMs())), + StreamOffset.create(stream, ReadOffset.lastConsumed())); + if (records == null || records.isEmpty()) { + continue; + } + for (MapRecord rec : records) { + Map fields = toStringMap(rec.getValue()); + ConsumeResult result = handler.dispatchMessage(fields); + if (result == ConsumeResult.ACK) { + redis.opsForStream().acknowledge(stream, group, rec.getId()); + } + // RETRY：不确认，留 PEL，由 StreamAutoClaimer 后续认领重投 + } + } catch (Exception e) { + if (!running || Thread.currentThread().isInterrupted()) { + break; + } + log.error("[redis-stream] read loop error stream={} group={}, backing off", stream, group, e); + sleepQuietly(props.getBlockMs()); + } + } + log.info("[redis-stream] listener loop exit stream={} group={}", stream, group); + } + + private static Map toStringMap(Map raw) { + Map m = new HashMap<>(Math.max(8, raw.size() * 2)); + raw.forEach((k, v) -> m.put(String.valueOf(k), v == null ? null : String.valueOf(v))); + return m; + } + + private static void sleepQuietly(long ms) { + try { + Thread.sleep(ms); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } + } + + private static boolean isBusyGroup(Throwable e) { + for (Throwable t = e; t != null; t = t.getCause()) { + if (t.getMessage() != null && t.getMessage().contains("BUSYGROUP")) { + return true; + } + } + return false; + } +} diff --git a/src/main/java/com/njcn/middle/stream/domain/StreamBaseMessage.java b/src/main/java/com/njcn/middle/stream/domain/StreamBaseMessage.java new file mode 100644 index 0000000..ebec4ad --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/domain/StreamBaseMessage.java @@ -0,0 +1,35 @@ +package com.njcn.middle.stream.domain; + +import cn.hutool.core.util.IdUtil; +import lombok.Data; + +import java.time.LocalDateTime; + +/** + * Redis Stream 消息基类，业务消息继承此类。 + * + *

自带基类、不依赖 message-api，保证 starter 零业务耦合。 + */ +@Data +public class StreamBaseMessage { + + /** + * 消息业务 key，默认雪花 id。 + * + *

{@link IdUtil#getSnowflake()} 无参经 {@code Singleton} 返回全局单例 Snowflake， + * 多次调用 sequence 连续递增、保证唯一；不可换成每次 {@code new Snowflake}（同毫秒不同实例会撞号）。 + */ + private String key = IdUtil.getSnowflake().nextIdStr(); + + /** 消息 tag（下行路由/过滤用），默认空串。 */ + private String tag = ""; + + /** 消息来源标识，默认空串。 */ + private String source = ""; + + /** 发送时间，默认 now()。 */ + private LocalDateTime sendTime = LocalDateTime.now(); + + /** 已重试次数，默认 0。 */ + private Integer retryTimes = 0; +} diff --git a/src/main/java/com/njcn/middle/stream/handler/ConsumeResult.java b/src/main/java/com/njcn/middle/stream/handler/ConsumeResult.java new file mode 100644 index 0000000..ade1494 --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/handler/ConsumeResult.java @@ -0,0 +1,14 @@ +package com.njcn.middle.stream.handler; + +/** + * 消费结果。 + * + *

    + *
  • {@link #ACK}：确认消费（含成功、被 filter 过滤、毒消息丢弃、不重试的失败）； + *
  • {@link #RETRY}：不确认，留在 PEL，由 autoclaim 后续重投。 + *
+ */ +public enum ConsumeResult { + ACK, + RETRY +} diff --git a/src/main/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandler.java b/src/main/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandler.java new file mode 100644 index 0000000..f5acbce --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandler.java @@ -0,0 +1,134 @@ +package com.njcn.middle.stream.handler; + +import com.alibaba.fastjson.JSON; +import com.njcn.middle.stream.codec.MessageCodec; +import com.njcn.middle.stream.constant.StreamMessageConstant; +import com.njcn.middle.stream.domain.StreamBaseMessage; + +import java.util.Map; + +/** + * 增强型 Stream 消费处理器基类（骨架，待 TDD 实现 dispatchMessage）。 + * + *

对标 rocket starter 的 {@code EnhanceConsumerMessageHandler}，但接收的是 + * 反序列化前的 {@code Map}（Redis Stream 字段）。 + */ +public abstract class EnhanceStreamConsumerHandler { + + /** + * {@link #getMaxRetryTimes()} 返回此哨兵值(或任意 ≤0 值)表示:本 handler 不单独配置重试次数, + * 沿用全局配置 {@code redis-stream.max-retry}。 + */ + public static final int INHERIT_GLOBAL_MAX_RETRY = -1; + + /** 监听的 topic。 */ + public abstract String topic(); + + /** 消费组。 */ + public abstract String group(); + + /** 消息类型（用于反序列化）。 */ + public abstract Class messageType(); + + /** 业务处理。 */ + public abstract void handleMessage(T message) throws Exception; + + /** 失败是否重试。 */ + public abstract boolean isRetry(); + + /** + * 是否向上抛异常。 + * + *

Redis 模式 no-op：Redis Stream 无 broker 重投语义，异常一律走 {@link #isRetry()} + * → RETRY/落异常；此钩子仅为与 rocket starter 共用一套接口而保留，Phase 2 复用。 + */ + public abstract boolean throwException(); + + /** 过滤：返回 true 则直接 ACK 丢弃，不处理。默认 false。 */ + public boolean filter(T message) { + return false; + } + + /** 消费成功回调，默认空。 */ + public void consumeSuccess(T message) { + } + + /** 落异常日志，默认空。 */ + public void saveExceptionMsgLog(T message, String body, Exception e) { + } + + /** + * 死信落异常:先尝试反序列化 {@code fields} 得到业务消息,再回调 {@link #saveExceptionMsgLog}。 + * + *

由 autoclaim 超 maxRetry 时调用。相比直接传 {@code null},本方法让死信回调能拿到解析后的业务 + * 对象(与 {@link #dispatchMessage} 非重试失败路径一致)。仅做反序列化、不调用 {@link #handleMessage}, + * 不触发任何业务副作用;反序列化失败(理论上不会——毒消息在首投即 ACK,不入此路径)时 message 兜底为 null。 + * + * @param fields stream 字段（含 enc/body） + * @param cause 死信原因 + */ + public void saveExceptionFromFields(Map fields, Exception cause) { + String body = fields.get(StreamMessageConstant.F_BODY); + T message = null; + try { + String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), body); + message = JSON.parseObject(json, messageType()); + } catch (Exception ignore) { + // 解析失败:保持 message=null,仍把原始 body 透传给回调 + } + saveExceptionMsgLog(message, body, cause); + } + + /** + * 本 handler 的最大投递次数(含首投),达到即落死信,由 autoclaim 维护。 + * + *

默认返回 {@link #INHERIT_GLOBAL_MAX_RETRY},表示沿用全局配置 {@code redis-stream.max-retry}; + * 覆写为正整数(≥1)即对本 handler 单独生效,优先级高于全局配置。 + */ + public int getMaxRetryTimes() { + return INHERIT_GLOBAL_MAX_RETRY; + } + + /** + * 分发一条消息（已从 stream 读出的字段 map）。 + * + *

流程：decode(enc,body) → 反序列化 → filter → handleMessage → consumeSuccess → ACK。 + * 解码/反序列化失败（毒消息）落异常并 ACK（不卡队列）；handleMessage 异常时 {@link #isRetry()} + * 真则 RETRY（留 PEL 由 autoclaim 兜），否则落异常 ACK。 + * + *

注：{@link StreamBaseMessage#getRetryTimes()} 主链路不读不判——重试计数与超限落异常由 + * autoclaim 维护（区别于 rocket 在 dispatch 内做 retryTimes 早退）。 + * + * @param fields stream 字段（含 enc/body） + * @return 消费结果 + */ + public ConsumeResult dispatchMessage(Map fields) { + String body = fields.get(StreamMessageConstant.F_BODY); + T message; + try { + String enc = fields.get(StreamMessageConstant.F_ENC); + String json = MessageCodec.decodeBody(enc, body); + message = JSON.parseObject(json, messageType()); + } catch (Exception e) { + // 毒消息：解码/反序列化失败 → 落异常 + ACK，避免反复重投卡死队列 + saveExceptionMsgLog(null, body, e); + return ConsumeResult.ACK; + } + + if (filter(message)) { + return ConsumeResult.ACK; + } + + try { + handleMessage(message); + consumeSuccess(message); + return ConsumeResult.ACK; + } catch (Exception e) { + if (isRetry()) { + return ConsumeResult.RETRY; + } + saveExceptionMsgLog(message, body, e); + return ConsumeResult.ACK; + } + } +} diff --git a/src/main/java/com/njcn/middle/stream/support/StreamKeyBuilder.java b/src/main/java/com/njcn/middle/stream/support/StreamKeyBuilder.java new file mode 100644 index 0000000..1154d17 --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/support/StreamKeyBuilder.java @@ -0,0 +1,33 @@ +package com.njcn.middle.stream.support; + +import cn.hutool.core.util.StrUtil; + +/** + * Stream 名（key）构造器。 + * + *

开启环境隔离且 env 非空时，对 topic 追加 {@code _env} 后缀，使各环境（如 dev/test/prod） + * 共用一套 Redis 时互不串流；否则原样返回 topic。 + */ +public class StreamKeyBuilder { + + private final boolean envIsolation; + private final String env; + + public StreamKeyBuilder(boolean envIsolation, String env) { + this.envIsolation = envIsolation; + this.env = env; + } + + /** + * 由 topic 构造实际 stream 名。 + * + * @param topic 业务 topic + * @return 开隔离且 env 非空 → {@code topic_env}；否则 {@code topic} + */ + public String buildStream(String topic) { + if (envIsolation && StrUtil.isNotBlank(env)) { + return topic + "_" + env; + } + return topic; + } +} diff --git a/src/main/java/com/njcn/middle/stream/support/StreamTrimScheduler.java b/src/main/java/com/njcn/middle/stream/support/StreamTrimScheduler.java new file mode 100644 index 0000000..dc3b41d --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/support/StreamTrimScheduler.java @@ -0,0 +1,102 @@ +package com.njcn.middle.stream.support; + +import com.njcn.middle.stream.autoconfig.RedisStreamProperties; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.SmartLifecycle; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * 定时裁剪 stream 长度，防止无界增长。 + * + *

按 {@code trim.intervalMs} 周期对每个 handler 的 stream 执行 + * {@code XTRIM MAXLEN ~ }（{@code ~} 近似裁剪，性能更好）。 + * + *

无 handler 时不启动。正确性由 C11 集成测试覆盖。 + */ +@Slf4j +public class StreamTrimScheduler implements SmartLifecycle { + + private final StringRedisTemplate redis; + private final StreamKeyBuilder keyBuilder; + private final RedisStreamProperties props; + private final List> handlers; + + private ScheduledExecutorService scheduler; + private volatile boolean running = false; + + public StreamTrimScheduler(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, + RedisStreamProperties props, + List> handlers) { + this.redis = redis; + this.keyBuilder = keyBuilder; + this.props = props; + this.handlers = handlers == null ? Collections.emptyList() : handlers; + } + + @Override + public synchronized void start() { + if (running) { + return; + } + running = true; + if (handlers.isEmpty()) { + return; + } + long interval = props.getTrim().getIntervalMs(); + scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "redis-stream-trim"); + t.setDaemon(true); + return t; + }); + scheduler.scheduleWithFixedDelay(this::trimAll, interval, interval, TimeUnit.MILLISECONDS); + log.info("[redis-stream] trim scheduler started, interval={}ms maxlen={}", interval, props.getTrim().getMaxlen()); + } + + @Override + public synchronized void stop() { + running = false; + if (scheduler != null) { + scheduler.shutdownNow(); + scheduler = null; + } + } + + @Override + public boolean isRunning() { + return running; + } + + private void trimAll() { + long maxlen = props.getTrim().getMaxlen(); + for (String stream : distinctStreams()) { + try { + redis.execute((RedisCallback) connection -> connection.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) { + log.warn("[redis-stream] trim failed stream={}", stream, e); + } + } + } + + private Set distinctStreams() { + Set streams = new LinkedHashSet<>(); + for (EnhanceStreamConsumerHandler handler : handlers) { + streams.add(keyBuilder.buildStream(handler.topic())); + } + return streams; + } +} diff --git a/src/main/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplate.java b/src/main/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplate.java new file mode 100644 index 0000000..c96affc --- /dev/null +++ b/src/main/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplate.java @@ -0,0 +1,56 @@ +package com.njcn.middle.stream.template; + +import com.alibaba.fastjson.JSONObject; +import com.njcn.middle.stream.codec.MessageCodec; +import com.njcn.middle.stream.constant.StreamMessageConstant; +import com.njcn.middle.stream.domain.StreamBaseMessage; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.util.HashMap; +import java.util.Map; + +/** + * Redis Stream 发送模板。 + * + *

组装 XADD 字段 {@code key/tag/enc/body/ts}（对死 C++ 契约），仅 body 按 compress 压缩； + * key/tag 从 message 取，不单独传参（故 Phase 2 风格B 须先 {@code message.setTag(nodeId)}）。 + */ +public class RedisStreamEnhanceTemplate { + + private final StringRedisTemplate redis; + private final StreamKeyBuilder keyBuilder; + + public RedisStreamEnhanceTemplate(StringRedisTemplate redis, StreamKeyBuilder keyBuilder) { + this.redis = redis; + this.keyBuilder = keyBuilder; + } + + /** + * 发送一条消息到 topic 对应的 stream。 + * + * @param topic 业务 topic（经 keyBuilder 转实际 stream 名） + * @param message 消息（key/tag 从其取） + * @param compress 是否压缩 body + * @return XADD 返回的 RecordId + */ + public RecordId send(String topic, T message, boolean compress) { + String stream = keyBuilder.buildStream(topic); + String json = JSONObject.toJSONString(message); + MessageCodec.Encoded encoded = MessageCodec.encodeBody(json, compress); + + Map fields = new HashMap<>(8); + fields.put(StreamMessageConstant.F_KEY, message.getKey()); + fields.put(StreamMessageConstant.F_TAG, message.getTag()); + fields.put(StreamMessageConstant.F_ENC, encoded.getEnc()); + fields.put(StreamMessageConstant.F_BODY, encoded.getBody()); + fields.put(StreamMessageConstant.F_TS, String.valueOf(System.currentTimeMillis())); + + MapRecord record = + StreamRecords.newRecord().in(stream).ofMap(fields); + return redis.opsForStream().add(record); + } +} diff --git a/src/main/resources/META-INF/spring.factories b/src/main/resources/META-INF/spring.factories new file mode 100644 index 0000000..fc231e6 --- /dev/null +++ b/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ +com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration diff --git a/src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamPropertiesTest.java b/src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamPropertiesTest.java new file mode 100644 index 0000000..d271477 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamPropertiesTest.java @@ -0,0 +1,37 @@ +package com.njcn.middle.stream.autoconfig; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +class RedisStreamPropertiesTest { + + @Test + void hasSaneDefaults() { + RedisStreamProperties p = new RedisStreamProperties(); + assertFalse(p.isEnvIsolation(), "envIsolation 默认 false"); + assertEquals("", p.getEnv(), "env 默认空串"); + assertEquals("", p.getConsumerName(), "consumerName 默认空串"); + assertEquals(10, p.getReadCount(), "readCount 默认 10"); + assertEquals(2000L, p.getBlockMs(), "blockMs 默认 2000"); + assertEquals(3, p.getMaxRetry(), "maxRetry 默认 3"); + } + + @Test + void autoclaimHasDefaults() { + RedisStreamProperties.Autoclaim a = new RedisStreamProperties().getAutoclaim(); + assertNotNull(a, "autoclaim 默认非 null"); + assertEquals(60000L, a.getMinIdleMs(), "autoclaim.minIdleMs 默认 60000"); + assertEquals(30000L, a.getIntervalMs(), "autoclaim.intervalMs 默认 30000"); + } + + @Test + void trimHasDefaults() { + RedisStreamProperties.Trim t = new RedisStreamProperties().getTrim(); + assertNotNull(t, "trim 默认非 null"); + assertEquals(60000L, t.getIntervalMs(), "trim.intervalMs 默认 60000"); + assertEquals(100000L, t.getMaxlen(), "trim.maxlen 默认 100000"); + } +} diff --git a/src/test/java/com/njcn/middle/stream/codec/MessageCodecTest.java b/src/test/java/com/njcn/middle/stream/codec/MessageCodecTest.java new file mode 100644 index 0000000..712163c --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/codec/MessageCodecTest.java @@ -0,0 +1,68 @@ +package com.njcn.middle.stream.codec; + +import cn.hutool.core.util.StrUtil; +import com.njcn.middle.stream.codec.MessageCodec.Encoded; +import com.njcn.middle.stream.constant.StreamMessageConstant; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class MessageCodecTest { + + /** 构造一段足够大且高度可压缩的 json（重复内容）。 */ + private static String bigCompressibleJson() { + return "{\"data\":\"" + StrUtil.repeat("abcdefgh", 200) + "\"}"; + } + + @Test + void plainRoundTrip() { + String json = "{\"a\":1,\"name\":\"hello\"}"; + Encoded e = MessageCodec.encodeBody(json, false); + assertEquals(StreamMessageConstant.ENC_PLAIN, e.getEnc(), "不压缩时 enc=plain"); + assertEquals(json, e.getBody(), "明文 body 即原 json"); + assertEquals(json, MessageCodec.decodeBody(e.getEnc(), e.getBody()), "plain 往返原样"); + } + + @Test + void gzipRoundTrip() { + String json = bigCompressibleJson(); + Encoded e = MessageCodec.encodeBody(json, true); + assertEquals(StreamMessageConstant.ENC_GZIP, e.getEnc(), "压缩时 enc=gzip"); + assertEquals(json, MessageCodec.decodeBody(e.getEnc(), e.getBody()), "gzip 往返逐字节一致"); + } + + @Test + void gzipBodyIsShorterForCompressible() { + String json = bigCompressibleJson(); + Encoded e = MessageCodec.encodeBody(json, true); + assertTrue(e.getBody().length() < json.length(), + "高可压缩内容压缩后 base64 应短于原 json：" + e.getBody().length() + " < " + json.length()); + } + + @Test + void gzipBodyHasNoNewlines() { + String json = bigCompressibleJson(); + Encoded e = MessageCodec.encodeBody(json, true); + assertFalse(e.getBody().contains("\n"), "base64 不应含 \\n"); + assertFalse(e.getBody().contains("\r"), "base64 不应含 \\r"); + } + + @Test + void unicodeGzipRoundTrip() { + // UTF-8 多字节字符（中文 + emoji）往返不丢 + String json = "{\"msg\":\"中文测试🚀\"}"; + Encoded e = MessageCodec.encodeBody(json, true); + assertEquals(json, MessageCodec.decodeBody(StreamMessageConstant.ENC_GZIP, e.getBody()), + "gzip 路径 UTF-8 中文/emoji 往返不丢"); + } + + @Test + void unknownEncFallsBackToPlain() { + // 消费侧只按 enc 解码；未知/缺省 enc 兜底按明文返回，不抛异常 + String body = "{\"x\":1}"; + assertEquals(body, MessageCodec.decodeBody("", body), "空 enc 兜底明文"); + assertEquals(body, MessageCodec.decodeBody(StreamMessageConstant.ENC_PLAIN, body), "plain 直接返回"); + } +} diff --git a/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerMaxRetryTest.java b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerMaxRetryTest.java new file mode 100644 index 0000000..0866059 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerMaxRetryTest.java @@ -0,0 +1,96 @@ +package com.njcn.middle.stream.container; + +import com.njcn.middle.stream.autoconfig.RedisStreamProperties; +import com.njcn.middle.stream.domain.StreamBaseMessage; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import org.junit.jupiter.api.Test; + +import java.util.Collections; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * 单元测试:per-handler 重试次数解析（{@link StreamAutoClaimer#resolveMaxRetry}）。 + * + *

不依赖 Redis：仅校验「handler 自配优先、否则沿用全局 {@code redis-stream.max-retry}」的优先级。 + */ +class StreamAutoClaimerMaxRetryTest { + + private StreamAutoClaimer claimerWithGlobal(int globalMaxRetry) { + RedisStreamProperties props = new RedisStreamProperties(); + props.setMaxRetry(globalMaxRetry); + // resolveMaxRetry 只用到 props 与 handler，redis/keyBuilder 传 null 即可 + return new StreamAutoClaimer(null, null, props, Collections.emptyList()); + } + + @Test + void handlerInheritsGlobalWhenNotOverridden() { + StreamAutoClaimer claimer = claimerWithGlobal(5); + // 默认 handler 返回哨兵 INHERIT_GLOBAL_MAX_RETRY → 沿用全局 5 + assertEquals(5, claimer.resolveMaxRetry( + new StubHandler(EnhanceStreamConsumerHandler.INHERIT_GLOBAL_MAX_RETRY)), + "未覆写时应沿用全局 max-retry"); + } + + @Test + void perHandlerOverrideWins() { + StreamAutoClaimer claimer = claimerWithGlobal(5); + assertEquals(10, claimer.resolveMaxRetry(new StubHandler(10)), + "handler 自配的正整数应优先于全局"); + assertEquals(2, claimer.resolveMaxRetry(new StubHandler(2)), + "handler 可配置比全局更小的次数"); + } + + @Test + void nonPositiveHandlerValueFallsBackToGlobal() { + StreamAutoClaimer claimer = claimerWithGlobal(3); + assertEquals(3, claimer.resolveMaxRetry(new StubHandler(0)), "0 视为未配 → 全局"); + assertEquals(3, claimer.resolveMaxRetry(new StubHandler(-7)), "负值视为未配 → 全局"); + } + + static class TestMsg extends StreamBaseMessage { + } + + /** 可指定 getMaxRetryTimes() 返回值的测试桩。 */ + static class StubHandler extends EnhanceStreamConsumerHandler { + private final int maxRetry; + + StubHandler(int maxRetry) { + this.maxRetry = maxRetry; + } + + @Override + public String topic() { + return "T"; + } + + @Override + public String group() { + return "G"; + } + + @Override + public Class messageType() { + return TestMsg.class; + } + + @Override + public void handleMessage(TestMsg message) { + } + + @Override + public boolean isRetry() { + return true; + } + + @Override + public boolean throwException() { + return false; + } + + @Override + public int getMaxRetryTimes() { + return maxRetry; + } + } +} diff --git a/src/test/java/com/njcn/middle/stream/domain/StreamBaseMessageTest.java b/src/test/java/com/njcn/middle/stream/domain/StreamBaseMessageTest.java new file mode 100644 index 0000000..a217c59 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/domain/StreamBaseMessageTest.java @@ -0,0 +1,33 @@ +package com.njcn.middle.stream.domain; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +class StreamBaseMessageTest { + + @Test + void newMessageHasDefaults() { + StreamBaseMessage m = new StreamBaseMessage(); + // key 默认雪花 id，非空 + assertNotNull(m.getKey(), "key 应有雪花默认值"); + assertFalse(m.getKey().isEmpty(), "key 不应为空串"); + // tag/source 默认空串（非 null） + assertEquals("", m.getTag(), "tag 默认空串"); + assertEquals("", m.getSource(), "source 默认空串"); + // sendTime 默认 now() + assertNotNull(m.getSendTime(), "sendTime 默认 now()"); + // retryTimes 默认 0 + assertEquals(0, m.getRetryTimes(), "retryTimes 默认 0"); + } + + @Test + void keysAreUnique() { + StreamBaseMessage a = new StreamBaseMessage(); + StreamBaseMessage b = new StreamBaseMessage(); + assertNotEquals(a.getKey(), b.getKey(), "不同实例 key 必须唯一"); + } +} diff --git a/src/test/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandlerTest.java b/src/test/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandlerTest.java new file mode 100644 index 0000000..6967e99 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandlerTest.java @@ -0,0 +1,223 @@ +package com.njcn.middle.stream.handler; + +import com.alibaba.fastjson.JSON; +import com.njcn.middle.stream.codec.MessageCodec; +import com.njcn.middle.stream.domain.StreamBaseMessage; +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.Map; + +import static com.njcn.middle.stream.constant.StreamMessageConstant.ENC_PLAIN; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_BODY; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_ENC; +import static com.njcn.middle.stream.handler.ConsumeResult.ACK; +import static com.njcn.middle.stream.handler.ConsumeResult.RETRY; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; + +class EnhanceStreamConsumerHandlerTest { + + @Data + @EqualsAndHashCode(callSuper = true) + static class Foo extends StreamBaseMessage { + private String name; + } + + /** 可配置行为、记录调用次数的测试桩。 */ + static class FooHandler extends EnhanceStreamConsumerHandler { + boolean retry = false; + boolean doFilter = false; + boolean throwOnHandle = false; + int handled = 0; + int succeeded = 0; + int exceptionLogged = 0; + Foo lastHandled; + Foo lastExceptionMsg; + String lastExceptionBody; + + @Override + public String topic() { + return "T"; + } + + @Override + public String group() { + return "G"; + } + + @Override + public Class messageType() { + return Foo.class; + } + + @Override + public void handleMessage(Foo message) throws Exception { + if (throwOnHandle) { + throw new RuntimeException("boom"); + } + handled++; + lastHandled = message; + } + + @Override + public boolean isRetry() { + return retry; + } + + @Override + public boolean throwException() { + return false; + } + + @Override + public boolean filter(Foo message) { + return doFilter; + } + + @Override + public void consumeSuccess(Foo message) { + succeeded++; + } + + @Override + public void saveExceptionMsgLog(Foo message, String body, Exception e) { + exceptionLogged++; + lastExceptionMsg = message; + lastExceptionBody = body; + } + } + + private static Map fieldsOf(Foo foo, boolean compress) { + String json = JSON.toJSONString(foo); + MessageCodec.Encoded e = MessageCodec.encodeBody(json, compress); + Map m = new HashMap<>(); + m.put(F_ENC, e.getEnc()); + m.put(F_BODY, e.getBody()); + return m; + } + + @Test + void happyPathAcksAndHandles() { + FooHandler h = new FooHandler(); + Foo foo = new Foo(); + foo.setName("hi"); + + ConsumeResult r = h.dispatchMessage(fieldsOf(foo, false)); + + assertEquals(ACK, r); + assertEquals(1, h.handled, "handleMessage 被调用"); + assertEquals("hi", h.lastHandled.getName(), "反序列化内容正确"); + assertEquals(1, h.succeeded, "consumeSuccess 被调用"); + assertEquals(0, h.exceptionLogged); + } + + @Test + void gzipBodyDecodesAndHandles() { + FooHandler h = new FooHandler(); + Foo foo = new Foo(); + foo.setName("hi"); + + ConsumeResult r = h.dispatchMessage(fieldsOf(foo, true)); + + assertEquals(ACK, r); + assertEquals(1, h.handled, "gzip body 能解码后处理"); + assertEquals("hi", h.lastHandled.getName()); + } + + @Test + void filterTrueAcksWithoutHandling() { + FooHandler h = new FooHandler(); + h.doFilter = true; + + ConsumeResult r = h.dispatchMessage(fieldsOf(new Foo(), false)); + + assertEquals(ACK, r); + assertEquals(0, h.handled, "filter 真则不处理"); + assertEquals(0, h.succeeded); + } + + @Test + void handleExceptionWithRetryReturnsRetry() { + FooHandler h = new FooHandler(); + h.throwOnHandle = true; + h.retry = true; + + ConsumeResult r = h.dispatchMessage(fieldsOf(new Foo(), false)); + + assertEquals(RETRY, r, "异常 + isRetry → RETRY"); + assertEquals(0, h.succeeded); + assertEquals(0, h.exceptionLogged, "retry 路径不落异常（留 PEL 由 autoclaim 兜）"); + } + + @Test + void handleExceptionWithoutRetryAcksAndLogs() { + FooHandler h = new FooHandler(); + h.throwOnHandle = true; + h.retry = false; + + ConsumeResult r = h.dispatchMessage(fieldsOf(new Foo(), false)); + + assertEquals(ACK, r, "异常 + 不 retry → ACK"); + assertEquals(1, h.exceptionLogged, "落异常日志"); + assertEquals(0, h.succeeded); + } + + @Test + void poisonMessageAcksAndLogs() { + FooHandler h = new FooHandler(); + Map bad = new HashMap<>(); + bad.put(F_ENC, ENC_PLAIN); + bad.put(F_BODY, "{not valid json"); + + ConsumeResult r = h.dispatchMessage(bad); + + assertEquals(ACK, r, "毒消息 ACK（不卡队列）"); + assertEquals(0, h.handled, "毒消息不进 handleMessage"); + assertEquals(1, h.exceptionLogged, "毒消息落异常"); + } + + @Test + void saveExceptionFromFieldsParsesMessageForDeadLetter() { + FooHandler h = new FooHandler(); + Foo foo = new Foo(); + foo.setName("dead"); + Map fields = fieldsOf(foo, false); + + h.saveExceptionFromFields(fields, new IllegalStateException("max retry exceeded")); + + assertEquals(1, h.exceptionLogged); + assertNotNull(h.lastExceptionMsg, "死信回调应拿到反序列化后的消息"); + assertEquals("dead", h.lastExceptionMsg.getName(), "消息内容正确"); + assertEquals(fields.get(F_BODY), h.lastExceptionBody, "body 原样透传"); + } + + @Test + void saveExceptionFromFieldsParsesGzipBody() { + FooHandler h = new FooHandler(); + Foo foo = new Foo(); + foo.setName("dead-gzip"); + + h.saveExceptionFromFields(fieldsOf(foo, true), new IllegalStateException("x")); + + assertNotNull(h.lastExceptionMsg); + assertEquals("dead-gzip", h.lastExceptionMsg.getName(), "gzip body 也能解析后回调"); + } + + @Test + void saveExceptionFromFieldsFallsBackToNullOnParseFailure() { + FooHandler h = new FooHandler(); + Map bad = new HashMap<>(); + bad.put(F_ENC, ENC_PLAIN); + bad.put(F_BODY, "{not valid json"); + + h.saveExceptionFromFields(bad, new IllegalStateException("x")); + + assertEquals(1, h.exceptionLogged); + assertNull(h.lastExceptionMsg, "解析失败时 message 兜底为 null"); + assertEquals("{not valid json", h.lastExceptionBody, "body 仍透传"); + } +} diff --git a/src/test/java/com/njcn/middle/stream/integration/StreamAutoClaimDeadLetterTest.java b/src/test/java/com/njcn/middle/stream/integration/StreamAutoClaimDeadLetterTest.java new file mode 100644 index 0000000..69a9e8e --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/integration/StreamAutoClaimDeadLetterTest.java @@ -0,0 +1,197 @@ +package com.njcn.middle.stream.integration; + +import com.njcn.middle.stream.domain.StreamBaseMessage; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate; +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ConditionEvaluationResult; +import org.junit.jupiter.api.extension.ExecutionCondition; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * 集成测试（Gate）：autoclaim 对「永久失败 + isRetry」消息的死信淘汰必须在有限轮内触发并清空 PEL。 + * + *

覆盖 review 发现的 Major：{@code XCLAIM ... JUSTID} 不递增投递计数、且 listener 用 + * {@code lastConsumed()} 永不重读 PEL，会让 {@code getTotalDeliveryCount()} 永久冻结在首投值 1， + * 使 {@code deliveries >= maxRetry} 恒 false、死信分支永不执行 → 消息被无限重投、PEL 不收敛。 + * + *

本用例配 {@code maxRetry=2}（须 ≥2 才能暴露冻结，=1 时 1>=1 会被掩盖）、{@code min-idle-ms=0}、 + * 小 {@code interval-ms}，断言：有限时间内 handler 收到 max-retry 落异常回调，且该 group 的 PEL 清零。 + * + *

Redis 不可达时整类禁用（skipped）。连接默认 {@code 127.0.0.1:6379} 无密码，指向其它 Redis 用 + * {@code -Dspring.redis.host=… -Dspring.redis.port=… -Dspring.redis.password=…} 覆盖（凭据不入代码）。 + * 本地起 Redis：{@code docker run -d -p 6379:6379 redis:6.2}。 + */ +@ExtendWith(StreamAutoClaimDeadLetterTest.RedisReachableCondition.class) +@SpringBootTest( + classes = StreamAutoClaimDeadLetterTest.TestApp.class, + properties = { + "redis-stream.env-isolation=false", + "redis-stream.block-ms=300", + "redis-stream.read-count=10", + "redis-stream.max-retry=2", + "redis-stream.consumer-name=dl-it-consumer", + "redis-stream.autoclaim.min-idle-ms=0", + "redis-stream.autoclaim.interval-ms=300" + }) +class StreamAutoClaimDeadLetterTest { + + @Autowired + private RedisStreamEnhanceTemplate template; + + @Autowired + private StringRedisTemplate redis; + + @Autowired + private StreamKeyBuilder keyBuilder; + + @Test + void permanentlyFailingMessageIsDeadLetteredAndPelDrains() throws Exception { + AlwaysFailHandler.DEAD_LETTER_LATCH = new CountDownLatch(1); + AlwaysFailHandler.DEAD_LETTER_EX.set(null); + + // 先清掉历史 stream（避免上一轮残留 PEL 干扰），再重建消费组： + // delete 会把 listener 启动时建的组一并删除，而 XADD 不会重建组，故须手动用 + // XGROUP CREATE ... MKSTREAM 重新建组，否则 listener/autoclaim 全部 NOGROUP。 + String stream = keyBuilder.buildStream(AlwaysFailHandler.TOPIC); + redis.delete(stream); + redis.execute((RedisCallback) conn -> { + try { + return conn.execute("XGROUP", + "CREATE".getBytes(StandardCharsets.UTF_8), + stream.getBytes(StandardCharsets.UTF_8), + AlwaysFailHandler.GROUP.getBytes(StandardCharsets.UTF_8), + "$".getBytes(StandardCharsets.UTF_8), + "MKSTREAM".getBytes(StandardCharsets.UTF_8)); + } catch (RuntimeException e) { + return null; // BUSYGROUP：组已存在，忽略 + } + }); + + AlwaysFail msg = new AlwaysFail(); + msg.setText("boom"); + template.send(AlwaysFailHandler.TOPIC, msg, false); + + // 永久失败 + isRetry=true：经首投 RETRY → autoclaim 反复重投，maxRetry=2 时必须在有限轮内落死信。 + boolean deadLettered = AlwaysFailHandler.DEAD_LETTER_LATCH.await(15, TimeUnit.SECONDS); + assertTrue(deadLettered, "15s 内应触发 max-retry 死信落异常（当前代码因计数冻结永不触发）"); + assertTrue(AlwaysFailHandler.DEAD_LETTER_EX.get() != null + && AlwaysFailHandler.DEAD_LETTER_EX.get().contains("max retry exceeded"), + "死信落异常应携带 max-retry 信息，实际=" + AlwaysFailHandler.DEAD_LETTER_EX.get()); + + // 死信后 PEL 必须清空。强制 ACK 在 saveExceptionMsgLog 回调之后执行（latch 先于 ACK trip）， + // 故轮询等待 ACK 完成，避免与 ACK 抢跑。 + long pending = -1; + for (int i = 0; i < 50; i++) { + PendingMessages pel = redis.opsForStream() + .pending(stream, AlwaysFailHandler.GROUP, Range.unbounded(), 100); + pending = pel == null ? 0 : pel.size(); + if (pending == 0) { + break; + } + Thread.sleep(100); + } + assertEquals(0L, pending, "死信后该 group 的 PEL 应清零，实际仍有 " + pending + " 条"); + } + + @SpringBootConfiguration + @EnableAutoConfiguration + static class TestApp { + @Bean + AlwaysFailHandler alwaysFailHandler() { + return new AlwaysFailHandler(); + } + } + + @Data + @EqualsAndHashCode(callSuper = true) + static class AlwaysFail extends StreamBaseMessage { + private String text; + } + + static class AlwaysFailHandler extends EnhanceStreamConsumerHandler { + static final String TOPIC = "IT_DeadLetter_Topic"; + static final String GROUP = "IT_DeadLetter_Group"; + static volatile CountDownLatch DEAD_LETTER_LATCH = new CountDownLatch(1); + static final AtomicReference DEAD_LETTER_EX = new AtomicReference<>(); + + @Override + public String topic() { + return TOPIC; + } + + @Override + public String group() { + return GROUP; + } + + @Override + public Class messageType() { + return AlwaysFail.class; + } + + @Override + public void handleMessage(AlwaysFail message) { + throw new IllegalStateException("permanent failure: " + message.getText()); + } + + @Override + public boolean isRetry() { + return true; + } + + @Override + public boolean throwException() { + return false; + } + + /** autoclaim 超 maxRetry 落异常时回调（dispatchMessage 的 RETRY 分支不会调用本方法）。 */ + @Override + public void saveExceptionMsgLog(AlwaysFail message, String body, Exception e) { + String em = e == null ? null : e.getMessage(); + if (em != null && em.contains("max retry exceeded")) { + DEAD_LETTER_EX.set(em); + DEAD_LETTER_LATCH.countDown(); + } + } + } + + /** Redis 不可达时禁用整个测试类（上下文加载即连 Redis，故须在加载前判断）。 */ + static class RedisReachableCondition implements ExecutionCondition { + @Override + public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) { + String host = System.getProperty("spring.redis.host", "127.0.0.1"); + int port = Integer.parseInt(System.getProperty("spring.redis.port", "6379")); + try (Socket s = new Socket()) { + s.connect(new InetSocketAddress(host, port), 500); + return ConditionEvaluationResult.enabled("redis reachable at " + host + ":" + port); + } catch (Exception e) { + return ConditionEvaluationResult.disabled( + "no redis at " + host + ":" + port + ", skip integration test"); + } + } + } +} diff --git a/src/test/java/com/njcn/middle/stream/integration/StreamRoundTripTest.java b/src/test/java/com/njcn/middle/stream/integration/StreamRoundTripTest.java new file mode 100644 index 0000000..30a5029 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/integration/StreamRoundTripTest.java @@ -0,0 +1,130 @@ +package com.njcn.middle.stream.integration; + +import com.njcn.middle.stream.domain.StreamBaseMessage; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate; +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ConditionEvaluationResult; +import org.junit.jupiter.api.extension.ExecutionCondition; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; + +import java.net.InetSocketAddress; +import java.net.Socket; +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; + +/** + * 集成测试（Gate）：真实 Redis 下跑通 {@code XADD → XREADGROUP → XACK} 闭环。 + * + *

Redis 不可达时整个类被 {@link RedisReachableCondition} 禁用（skipped，非 failed）， + * 保证无 Redis 环境 {@code mvn test} 仍全绿。 + * + *

连接默认 {@code 127.0.0.1:6379} 无密码；指向其它 Redis 用系统属性覆盖（凭据不入代码）： + * {@code mvn test -Dspring.redis.host=x.x.x.x -Dspring.redis.port=12379 -Dspring.redis.password=***}。 + * 本地起 Redis：{@code docker run -d -p 6379:6379 redis:6.2}。 + */ +@ExtendWith(StreamRoundTripTest.RedisReachableCondition.class) +@SpringBootTest( + classes = StreamRoundTripTest.TestApp.class, + properties = { + "redis-stream.env-isolation=false", + "redis-stream.block-ms=500", + "redis-stream.consumer-name=it-consumer" + }) +class StreamRoundTripTest { + + @Autowired + private RedisStreamEnhanceTemplate template; + + @Test + void sendAndReceiveRoundTrip() throws Exception { + PingHandler.RECEIVED.clear(); + Ping ping = new Ping(); + ping.setText("hello-it"); + + template.send(PingHandler.TOPIC, ping, false); + + Ping got = PingHandler.RECEIVED.poll(5, TimeUnit.SECONDS); + assertNotNull(got, "5s 内应通过 XREADGROUP 收到消息"); + assertEquals("hello-it", got.getText(), "内容往返一致"); + assertEquals(ping.getKey(), got.getKey(), "key 往返一致"); + } + + @SpringBootConfiguration + @EnableAutoConfiguration + static class TestApp { + @Bean + PingHandler pingHandler() { + return new PingHandler(); + } + } + + @Data + @EqualsAndHashCode(callSuper = true) + static class Ping extends StreamBaseMessage { + private String text; + } + + static class PingHandler extends EnhanceStreamConsumerHandler { + static final String TOPIC = "IT_Ping_Topic"; + static final BlockingQueue RECEIVED = new LinkedBlockingQueue<>(); + + @Override + public String topic() { + return TOPIC; + } + + @Override + public String group() { + return "IT_Ping_Group"; + } + + @Override + public Class messageType() { + return Ping.class; + } + + @Override + public void handleMessage(Ping message) { + RECEIVED.add(message); + } + + @Override + public boolean isRetry() { + return false; + } + + @Override + public boolean throwException() { + return false; + } + } + + /** Redis 不可达时禁用整个测试类（上下文加载即连 Redis，故须在加载前判断）。 */ + static class RedisReachableCondition implements ExecutionCondition { + @Override + public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) { + String host = System.getProperty("spring.redis.host", "127.0.0.1"); + int port = Integer.parseInt(System.getProperty("spring.redis.port", "6379")); + try (Socket s = new Socket()) { + s.connect(new InetSocketAddress(host, port), 500); + return ConditionEvaluationResult.enabled("redis reachable at " + host + ":" + port); + } catch (Exception e) { + return ConditionEvaluationResult.disabled( + "no redis at " + host + ":" + port + ", skip integration test"); + } + } + } +} diff --git a/src/test/java/com/njcn/middle/stream/support/StreamKeyBuilderTest.java b/src/test/java/com/njcn/middle/stream/support/StreamKeyBuilderTest.java new file mode 100644 index 0000000..f3d883a --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/support/StreamKeyBuilderTest.java @@ -0,0 +1,29 @@ +package com.njcn.middle.stream.support; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class StreamKeyBuilderTest { + + @Test + void isolatedWithEnvAppendsSuffix() { + StreamKeyBuilder b = new StreamKeyBuilder(true, "prod"); + assertEquals("FrontData_Topic_prod", b.buildStream("FrontData_Topic"), + "开隔离且 env 非空 → topic_env"); + } + + @Test + void notIsolatedReturnsTopicAsIs() { + StreamKeyBuilder b = new StreamKeyBuilder(false, "prod"); + assertEquals("FrontData_Topic", b.buildStream("FrontData_Topic"), + "关隔离 → 原 topic（忽略 env）"); + } + + @Test + void isolatedWithBlankEnvReturnsTopicAsIs() { + assertEquals("T", new StreamKeyBuilder(true, "").buildStream("T"), "env 空串 → 原 topic"); + assertEquals("T", new StreamKeyBuilder(true, " ").buildStream("T"), "env 全空白 → 原 topic"); + assertEquals("T", new StreamKeyBuilder(true, null).buildStream("T"), "env null → 原 topic"); + } +} diff --git a/src/test/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplateTest.java b/src/test/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplateTest.java new file mode 100644 index 0000000..8361b78 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplateTest.java @@ -0,0 +1,93 @@ +package com.njcn.middle.stream.template; + +import cn.hutool.core.util.StrUtil; +import com.alibaba.fastjson.JSONObject; +import com.njcn.middle.stream.domain.StreamBaseMessage; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.core.StreamOperations; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.util.Map; + +import static com.njcn.middle.stream.constant.StreamMessageConstant.ENC_GZIP; +import static com.njcn.middle.stream.constant.StreamMessageConstant.ENC_PLAIN; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_BODY; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_ENC; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_KEY; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_TAG; +import static com.njcn.middle.stream.constant.StreamMessageConstant.F_TS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; + +class RedisStreamEnhanceTemplateTest { + + @Data + @EqualsAndHashCode(callSuper = true) + static class Foo extends StreamBaseMessage { + private String name; + } + + @SuppressWarnings({"unchecked", "rawtypes"}) + private MapRecord captureSend(boolean compress, Foo foo, String topic, StreamKeyBuilder kb) { + StringRedisTemplate redis = mock(StringRedisTemplate.class); + StreamOperations ops = mock(StreamOperations.class); + doReturn(ops).when(redis).opsForStream(); + // send() 内 ofMap 返回 MapRecord，编译期绑定 add(MapRecord) 重载（非 add(Record)） + doReturn(RecordId.of("0-1")).when(ops).add(any(MapRecord.class)); + + RedisStreamEnhanceTemplate tpl = new RedisStreamEnhanceTemplate(redis, kb); + RecordId id = tpl.send(topic, foo, compress); + assertEquals("0-1", id.getValue(), "返回 XADD 的 RecordId"); + + ArgumentCaptor cap = ArgumentCaptor.forClass(MapRecord.class); + verify(ops).add(cap.capture()); + return (MapRecord) cap.getValue(); + } + + @Test + void sendPlainFillsFiveFields() { + Foo foo = new Foo(); + foo.setTag("t1"); + foo.setName("hello"); + MapRecord rec = captureSend(false, foo, "MyTopic", new StreamKeyBuilder(false, "")); + + assertEquals("MyTopic", rec.getStream(), "关隔离 → 原 topic 作 stream 名"); + Map v = rec.getValue(); + assertEquals(5, v.size(), "固定五字段 key/tag/enc/body/ts"); + assertEquals(foo.getKey(), v.get(F_KEY), "key 取自 message"); + assertEquals("t1", v.get(F_TAG), "tag 取自 message"); + assertEquals(ENC_PLAIN, v.get(F_ENC), "不压缩 enc=plain"); + assertTrue(v.get(F_BODY).contains("hello"), "明文 body 含原文"); + assertNotNull(v.get(F_TS), "ts 非空"); + } + + @Test + void sendCompressUsesGzipAndShorterBody() { + Foo foo = new Foo(); + foo.setName(StrUtil.repeat("abcdefgh", 200)); + String json = JSONObject.toJSONString(foo); + + MapRecord rec = captureSend(true, foo, "MyTopic", new StreamKeyBuilder(false, "")); + Map v = rec.getValue(); + assertEquals(ENC_GZIP, v.get(F_ENC), "压缩 enc=gzip"); + assertTrue(v.get(F_BODY).length() < json.length(), "压缩 body 短于原 json"); + } + + @Test + void sendUsesEnvIsolatedStreamName() { + MapRecord rec = + captureSend(false, new Foo(), "MyTopic", new StreamKeyBuilder(true, "prod")); + assertEquals("MyTopic_prod", rec.getStream(), "开隔离 → stream 名带 env 后缀"); + } +}