feat: Redis Stream Spring Boot starter 首次提交

封装 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) <noreply@anthropic.com>
2026-06-24 20:55:12 +08:00
commit 74f68c9680
25 changed files with 2365 additions and 0 deletions
--- a/.gitignore
+++ 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
--- a/README.md
+++ 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
+<dependency>
+    <groupId>com.njcn</groupId>
+    <artifactId>redis-stream-springboot-starter</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+</dependency>
+```
+
+制品发布在内网 Nexus,若拉不到依赖请确认 `settings.xml` / `pom.xml` 配置了仓库:
+
+```xml
+<repository>
+    <id>nexus-snapshots</id>
+    <url>http://192.168.1.22:8001/nexus/content/repositories/snapshots/</url>
+</repository>
+```
+
+> 无需 `@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<T>` 并**注册为 Spring Bean**,框架会为每个 handler 起一条消费线程:
+
+```java
+import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler;
+import org.springframework.stereotype.Component;
+
+@Component
+public class OrderConsumer extends EnhanceStreamConsumerHandler<OrderMessage> {
+
+    @Override
+    public String topic() {
+        return "order-topic";        // 与发送方一致
+    }
+
+    @Override
+    public String group() {
+        return "order-consumer-group"; // 消费组名,启动时自动 XGROUP CREATE
+    }
+
+    @Override
+    public Class<OrderMessage> 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
+```
--- a/pom.xml
+++ b/pom.xml
@@ -0,0 +1,129 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.njcn</groupId>
+    <artifactId>redis-stream-springboot-starter</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <packaging>jar</packaging>
+    <name>redis stream中间件二次开发sdk</name>
+    <description>封装 Redis Streams 收发，对标 rocket-mq-springboot-starter，零业务依赖</description>
+
+    <distributionManagement>
+        <repository>
+            <id>nexus-releases</id>
+            <name>Nexus Release Repository</name>
+            <url>http://192.168.1.22:8001/nexus/content/repositories/releases/</url>
+        </repository>
+        <snapshotRepository>
+            <id>nexus-snapshots</id>
+            <name>Nexus Snapshot Repository</name>
+            <url>http://192.168.1.22:8001/nexus/content/repositories/snapshots/</url>
+        </snapshotRepository>
+    </distributionManagement>
+
+    <properties>
+        <java.version>1.8</java.version>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
+        <springboot.version>2.3.12.RELEASE</springboot.version>
+        <!--
+          autoconfigure 与 SB 同版本（不沿用 rocket starter 的 2.0.0.RELEASE）：
+          2.0.0 会经 Maven 最近优先把 spring-context 降到 5.0.x，而 SmartLifecycle.stop(Runnable)/
+          isAutoStartup() 在 5.2 才是 default 方法，C7/C8/C9 容器实现 SmartLifecycle 会编译失败。
+        -->
+        <autoconfigure.version>2.3.12.RELEASE</autoconfigure.version>
+        <fastjson.version>1.2.83</fastjson.version>
+        <hutool.version>5.7.9</hutool.version>
+        <lombok.version>1.18.18</lombok.version>
+    </properties>
+
+    <dependencies>
+        <!--Redis（含 spring-data-redis 2.3.x + lettuce）-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+            <version>${springboot.version}</version>
+        </dependency>
+        <!--自动装配-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+            <version>${autoconfigure.version}</version>
+        </dependency>
+        <!--避免idea后端配置类报红-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+            <version>${springboot.version}</version>
+        </dependency>
+        <!--JSON-->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>${fastjson.version}</version>
+        </dependency>
+        <!--java工具包（雪花id等）-->
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>${hutool.version}</version>
+        </dependency>
+        <!--java简化开发工具（optional：仅编译期需要，不传递给下游消费方）-->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>${lombok.version}</version>
+            <optional>true</optional>
+        </dependency>
+        <!--测试-->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <version>${springboot.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.8.1</version>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                    <encoding>UTF-8</encoding>
+                </configuration>
+            </plugin>
+            <!--
+              默认 super-POM 绑定的 surefire 2.12.4 不识别 JUnit 5(Jupiter)，会导致 Tests run: 0（静默假绿）。
+              本项目未使用 spring-boot-starter-parent，故显式升到 2.22.2 以驱动 JUnit Platform。
+            -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>2.22.2</version>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <version>2.2.1</version>
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <goals>
+                            <goal>jar-no-fork</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
--- a/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamAutoConfiguration.java
+++ 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 自动装配。
+ *
+ * <p>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<EnhanceStreamConsumerHandler<?>> handlers) {
+        return new StreamListenerContainer(redis, streamKeyBuilder, props, collect(handlers));
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public StreamAutoClaimer streamAutoClaimer(StringRedisTemplate redis,
+                                               StreamKeyBuilder streamKeyBuilder,
+                                               RedisStreamProperties props,
+                                               ObjectProvider<EnhanceStreamConsumerHandler<?>> handlers) {
+        return new StreamAutoClaimer(redis, streamKeyBuilder, props, collect(handlers));
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public StreamTrimScheduler streamTrimScheduler(StringRedisTemplate redis,
+                                                   StreamKeyBuilder streamKeyBuilder,
+                                                   RedisStreamProperties props,
+                                                   ObjectProvider<EnhanceStreamConsumerHandler<?>> handlers) {
+        return new StreamTrimScheduler(redis, streamKeyBuilder, props, collect(handlers));
+    }
+
+    private static List<EnhanceStreamConsumerHandler<?>> collect(ObjectProvider<EnhanceStreamConsumerHandler<?>> handlers) {
+        return handlers.orderedStream().collect(Collectors.toList());
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/autoconfig/RedisStreamProperties.java
+++ 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}。
+ *
+ * <p>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;
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/codec/MessageCodec.java
+++ 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;
+
+/**
+ * 消息体编解码器。
+ *
+ * <p>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;
+        }
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/constant/StreamMessageConstant.java
+++ b/src/main/java/com/njcn/middle/stream/constant/StreamMessageConstant.java
@@ -0,0 +1,43 @@
+package com.njcn.middle.stream.constant;
+
+/**
+ * Redis Stream 消息相关常量。
+ *
+ * <p>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";
+}
--- a/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java
+++ 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 中长时间未确认（消费者宕机）的消息。
+ *
+ * <p>spring-data-redis 2.3.x 无 {@code XAUTOCLAIM}（需 2.6+），故走
+ * {@code pending} + 原生 {@code XCLAIM ... JUSTID} + {@code range} 取内容 + 重新 dispatch。
+ * 属崩溃恢复兜底、非主链路（主链路重试由消费线程 RETRY 留 PEL 触发本认领）。
+ *
+ * <p>投递次数超过 {@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<EnhanceStreamConsumerHandler<?>> handlers;
+
+    private ScheduledExecutorService scheduler;
+    private volatile boolean running = false;
+    private String claimConsumer;
+
+    public StreamAutoClaimer(StringRedisTemplate redis, StreamKeyBuilder keyBuilder,
+                             RedisStreamProperties props,
+                             List<EnhanceStreamConsumerHandler<?>> 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<String> 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 消息：取内容 → 死信判定前置 → 认领重投。
+     *
+     * <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数；因 {@link #claim} 用<b>不带 JUSTID</b>
+     * 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。
+     * {@code maxRetry} 语义 = 最多投递 maxRetry 次（含首投），第 maxRetry 次仍 RETRY 即落死信。
+     * 死信判定<b>前置于 dispatch</b>：达上限直接落异常 + 强制 ACK，不再多跑一次业务副作用。
+     */
+    private void claimOne(EnhanceStreamConsumerHandler<?> handler, String stream, String group,
+                          long minIdle, int maxRetry, String id, long deliveries) {
+        List<MapRecord<String, Object, Object>> 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<String, Object, Object> rec = records.get(0);
+        Map<String, String> 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}（<b>不带 JUSTID</b>）。
+     *
+     * <p>不带 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<Object>) 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<String, String> toStringMap(Map<Object, Object> raw) {
+        Map<String, String> 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;
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/container/StreamListenerContainer.java
+++ 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 一条消费组消费线程。
+ *
+ * <p>{@code XREADGROUP} 阻塞读 → {@link EnhanceStreamConsumerHandler#dispatchMessage} →
+ * 结果 ACK 则 {@code XACK}，RETRY 则不确认（留 PEL 由 autoclaim 兜）。
+ *
+ * <p>无 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<EnhanceStreamConsumerHandler<?>> handlers;
+
+    private final List<Thread> workers = new ArrayList<>();
+    private volatile boolean running = false;
+
+    public StreamListenerContainer(StringRedisTemplate redis, StreamKeyBuilder keyBuilder,
+                                   RedisStreamProperties props,
+                                   List<EnhanceStreamConsumerHandler<?>> 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<Object>) 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<MapRecord<String, Object, Object>> 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<String, Object, Object> rec : records) {
+                    Map<String, String> 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<String, String> toStringMap(Map<Object, Object> raw) {
+        Map<String, String> 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;
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/domain/StreamBaseMessage.java
+++ 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 消息基类，业务消息继承此类。
+ *
+ * <p>自带基类、<b>不依赖 message-api</b>，保证 starter 零业务耦合。
+ */
+@Data
+public class StreamBaseMessage {
+
+    /**
+     * 消息业务 key，默认雪花 id。
+     *
+     * <p>{@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;
+}
--- a/src/main/java/com/njcn/middle/stream/handler/ConsumeResult.java
+++ b/src/main/java/com/njcn/middle/stream/handler/ConsumeResult.java
@@ -0,0 +1,14 @@
+package com.njcn.middle.stream.handler;
+
+/**
+ * 消费结果。
+ *
+ * <ul>
+ *   <li>{@link #ACK}：确认消费（含成功、被 filter 过滤、毒消息丢弃、不重试的失败）；
+ *   <li>{@link #RETRY}：不确认，留在 PEL，由 autoclaim 后续重投。
+ * </ul>
+ */
+public enum ConsumeResult {
+    ACK,
+    RETRY
+}
--- a/src/main/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandler.java
+++ 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）。
+ *
+ * <p>对标 rocket starter 的 {@code EnhanceConsumerMessageHandler}，但接收的是
+ * 反序列化前的 {@code Map<String,String>}（Redis Stream 字段）。
+ */
+public abstract class EnhanceStreamConsumerHandler<T extends StreamBaseMessage> {
+
+    /**
+     * {@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<T> messageType();
+
+    /** 业务处理。 */
+    public abstract void handleMessage(T message) throws Exception;
+
+    /** 失败是否重试。 */
+    public abstract boolean isRetry();
+
+    /**
+     * 是否向上抛异常。
+     *
+     * <p><b>Redis 模式 no-op</b>：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}。
+     *
+     * <p>由 autoclaim 超 maxRetry 时调用。相比直接传 {@code null},本方法让死信回调能拿到解析后的业务
+     * 对象(与 {@link #dispatchMessage} 非重试失败路径一致)。仅做反序列化、<b>不调用 {@link #handleMessage}</b>,
+     * 不触发任何业务副作用;反序列化失败(理论上不会——毒消息在首投即 ACK,不入此路径)时 message 兜底为 null。
+     *
+     * @param fields stream 字段（含 enc/body）
+     * @param cause  死信原因
+     */
+    public void saveExceptionFromFields(Map<String, String> 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 维护。
+     *
+     * <p>默认返回 {@link #INHERIT_GLOBAL_MAX_RETRY},表示沿用全局配置 {@code redis-stream.max-retry};
+     * 覆写为正整数(≥1)即对本 handler 单独生效,优先级高于全局配置。
+     */
+    public int getMaxRetryTimes() {
+        return INHERIT_GLOBAL_MAX_RETRY;
+    }
+
+    /**
+     * 分发一条消息（已从 stream 读出的字段 map）。
+     *
+     * <p>流程：decode(enc,body) → 反序列化 → filter → handleMessage → consumeSuccess → ACK。
+     * 解码/反序列化失败（毒消息）落异常并 ACK（不卡队列）；handleMessage 异常时 {@link #isRetry()}
+     * 真则 RETRY（留 PEL 由 autoclaim 兜），否则落异常 ACK。
+     *
+     * <p>注：{@link StreamBaseMessage#getRetryTimes()} 主链路不读不判——重试计数与超限落异常由
+     * autoclaim 维护（区别于 rocket 在 dispatch 内做 retryTimes 早退）。
+     *
+     * @param fields stream 字段（含 enc/body）
+     * @return 消费结果
+     */
+    public ConsumeResult dispatchMessage(Map<String, String> 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;
+        }
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/support/StreamKeyBuilder.java
+++ 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）构造器。
+ *
+ * <p>开启环境隔离且 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;
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/support/StreamTrimScheduler.java
+++ 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 长度，防止无界增长。
+ *
+ * <p>按 {@code trim.intervalMs} 周期对每个 handler 的 stream 执行
+ * {@code XTRIM <stream> MAXLEN ~ <maxlen>}（{@code ~} 近似裁剪，性能更好）。
+ *
+ * <p>无 handler 时不启动。正确性由 C11 集成测试覆盖。
+ */
+@Slf4j
+public class StreamTrimScheduler implements SmartLifecycle {
+
+    private final StringRedisTemplate redis;
+    private final StreamKeyBuilder keyBuilder;
+    private final RedisStreamProperties props;
+    private final List<EnhanceStreamConsumerHandler<?>> handlers;
+
+    private ScheduledExecutorService scheduler;
+    private volatile boolean running = false;
+
+    public StreamTrimScheduler(StringRedisTemplate redis, StreamKeyBuilder keyBuilder,
+                               RedisStreamProperties props,
+                               List<EnhanceStreamConsumerHandler<?>> 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<Object>) 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<String> distinctStreams() {
+        Set<String> streams = new LinkedHashSet<>();
+        for (EnhanceStreamConsumerHandler<?> handler : handlers) {
+            streams.add(keyBuilder.buildStream(handler.topic()));
+        }
+        return streams;
+    }
+}
--- a/src/main/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplate.java
+++ 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 发送模板。
+ *
+ * <p>组装 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 <T extends StreamBaseMessage> 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<String, String> 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<String, String, String> record =
+                StreamRecords.newRecord().in(stream).ofMap(fields);
+        return redis.opsForStream().add(record);
+    }
+}
--- a/src/main/resources/META-INF/spring.factories
+++ b/src/main/resources/META-INF/spring.factories
@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration
--- a/src/test/java/com/njcn/middle/stream/autoconfig/RedisStreamPropertiesTest.java
+++ 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");
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/codec/MessageCodecTest.java
+++ 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 直接返回");
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerMaxRetryTest.java
+++ 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}）。
+ *
+ * <p>不依赖 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<TestMsg> {
+        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<TestMsg> 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;
+        }
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/domain/StreamBaseMessageTest.java
+++ 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 必须唯一");
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/handler/EnhanceStreamConsumerHandlerTest.java
+++ 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<Foo> {
+        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<Foo> 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<String, String> fieldsOf(Foo foo, boolean compress) {
+        String json = JSON.toJSONString(foo);
+        MessageCodec.Encoded e = MessageCodec.encodeBody(json, compress);
+        Map<String, String> 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<String, String> 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<String, String> 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<String, String> 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 仍透传");
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/integration/StreamAutoClaimDeadLetterTest.java
+++ 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。
+ *
+ * <p>覆盖 review 发现的 Major：{@code XCLAIM ... JUSTID} 不递增投递计数、且 listener 用
+ * {@code lastConsumed()} 永不重读 PEL，会让 {@code getTotalDeliveryCount()} 永久冻结在首投值 1，
+ * 使 {@code deliveries >= maxRetry} 恒 false、死信分支永不执行 → 消息被无限重投、PEL 不收敛。
+ *
+ * <p>本用例配 {@code maxRetry=2}（须 ≥2 才能暴露冻结，=1 时 1>=1 会被掩盖）、{@code min-idle-ms=0}、
+ * 小 {@code interval-ms}，断言：有限时间内 handler 收到 max-retry 落异常回调，且该 group 的 PEL 清零。
+ *
+ * <p>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<Object>) 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<AlwaysFail> {
+        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<String> DEAD_LETTER_EX = new AtomicReference<>();
+
+        @Override
+        public String topic() {
+            return TOPIC;
+        }
+
+        @Override
+        public String group() {
+            return GROUP;
+        }
+
+        @Override
+        public Class<AlwaysFail> 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");
+            }
+        }
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/integration/StreamRoundTripTest.java
+++ 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} 闭环。
+ *
+ * <p>Redis 不可达时整个类被 {@link RedisReachableCondition} 禁用（skipped，非 failed），
+ * 保证无 Redis 环境 {@code mvn test} 仍全绿。
+ *
+ * <p>连接默认 {@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<Ping> {
+        static final String TOPIC = "IT_Ping_Topic";
+        static final BlockingQueue<Ping> RECEIVED = new LinkedBlockingQueue<>();
+
+        @Override
+        public String topic() {
+            return TOPIC;
+        }
+
+        @Override
+        public String group() {
+            return "IT_Ping_Group";
+        }
+
+        @Override
+        public Class<Ping> 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");
+            }
+        }
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/support/StreamKeyBuilderTest.java
+++ 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");
+    }
+}
--- a/src/test/java/com/njcn/middle/stream/template/RedisStreamEnhanceTemplateTest.java
+++ 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<String, String, String> 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<MapRecord> cap = ArgumentCaptor.forClass(MapRecord.class);
+        verify(ops).add(cap.capture());
+        return (MapRecord<String, String, String>) cap.getValue();
+    }
+
+    @Test
+    void sendPlainFillsFiveFields() {
+        Foo foo = new Foo();
+        foo.setTag("t1");
+        foo.setName("hello");
+        MapRecord<String, String, String> rec = captureSend(false, foo, "MyTopic", new StreamKeyBuilder(false, ""));
+
+        assertEquals("MyTopic", rec.getStream(), "关隔离 → 原 topic 作 stream 名");
+        Map<String, String> 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<String, String, String> rec = captureSend(true, foo, "MyTopic", new StreamKeyBuilder(false, ""));
+        Map<String, String> 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<String, String, String> rec =
+                captureSend(false, new Foo(), "MyTopic", new StreamKeyBuilder(true, "prod"));
+        assertEquals("MyTopic_prod", rec.getStream(), "开隔离 → stream 名带 env 后缀");
+    }
+}