redis-stream-springboot-starter/README.md

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++ 等异构端互通)

---

一、引入依赖

<dependency>
    <groupId>com.njcn</groupId>
    <artifactId>redis-stream-springboot-starter</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

制品发布在内网 Nexus,若拉不到依赖请确认 settings.xml / pom.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:

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):

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):

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 起一条消费线程:

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,均为可选——不配即用注释中的默认值。Spring 宽松绑定,read-count / readCount 等写法等价。

redis-stream:
  # ===== 基础 =====
  env-isolation: false   # 是否按环境隔离 stream 名:开启后名字追加 _env 后缀,多环境共用一套 Redis 时互不串扰
  env: ""                # 环境标识,env-isolation=true 且非空时作为 stream 名后缀(topic_env);收发双方须配一致
  consumer-name: ""      # 消费者名,留空自动生成唯一名。多实例务必留空,写死会被当作同一消费者互相争抢、破坏负载均衡(见「八、注意事项」)
  read-count: 10         # 单次 XREADGROUP 拉取条数(一次最多取多少条来处理)
  block-ms: 2000         # XREADGROUP 阻塞等待毫秒:无新消息时阻塞这么久再返回,影响拉取实时性与空转开销
  max-retry: 3           # 全局最大投递次数(含首投),超限落死信;可被 handler 的 getMaxRetryTimes() 单独覆写

  # ===== 崩溃恢复 · 自动认领:消费者宕机/超时未 ACK 的消息滞留 PEL,由后台定时认领重投(机制见「六、消费语义」) =====
  autoclaim:
    min-idle-ms: 60000   # 消息 idle(距上次投递)超过此值(ms)才可被其它消费者认领重投。太小会与正常重试争抢,太大则故障恢复慢
    interval-ms: 30000   # autoclaim 扫描 PEL 的周期(ms)

  # ===== Stream 长度裁剪:防无界增长,后台定时 XTRIM <stream> MAXLEN <maxlen>(精确裁剪),保留最新 maxlen 条 =====
  trim:
    maxlen: 100000       # 保留的最大条数(精确裁剪)。默认 10万;要 5万就设 50000,超出部分删最老条目
    interval-ms: 60000   # 裁剪执行周期(ms)。调小可让实际长度更贴近 maxlen(代价是 XTRIM 执行更频繁)

Stream 裁剪有两点必须知道:

1. 软上限,非实时硬卡。 两次裁剪之间 stream 会短暂超过 maxlen,峰值 ≈ maxlen + 生产速率 × interval-ms。例:maxlen=50000、每秒进 1000 条、周期 60s ⇒ 下次裁剪前最高可达 11 万,随后拉回 5 万。要更紧就调小 interval-ms。
2. 按条数裁剪,不区分是否已消费。 XTRIM MAXLEN 只删最老条目,不管是否已 ACK。maxlen 设太小且消费积压超过它时,未消费的老消息会被直接裁掉 = 丢消息(被裁的 PEL 消息会被静默强制 ACK)。⚠️ maxlen 要留足:至少覆盖「峰值生产速率 × 消费者最长停机/滞后时间」。5 万 / 10 万对一般业务够用,单条消费慢或实例可能长停时需上调。

注:spring-data-redis 2.3.x 高级 API 仅支持精确裁剪,无近似裁剪(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 防止无界增长(详见「七、配置项 · Stream 长度裁剪」)。请确保 maxlen 足够大,避免重试中的消息在达到 max-retry 前被裁掉(被裁掉的 PEL 消息会被静默强制 ACK)。
• 无 handler 时容器空转:未注册任何 EnhanceStreamConsumerHandler Bean 时,监听/认领/裁剪线程均不启动,仅 RedisStreamEnhanceTemplate 可用于纯发送场景。

---

九、本地验证(可选)

集成测试需要本地 Redis;不可达时整类自动 skip(不影响 mvn test 全绿):

docker run -d -p 6379:6379 redis:6.2
mvn test

---

十、Redis 缓存(RedisCacheEnhanceTemplate)

除 Stream 收发外,starter 同样提供开箱即用的编程式缓存:把 TTL 管理、fastjson 序列化、穿透/雪崩/击穿三防与滑动续期,全部封装进一个 getOrLoad。注入即用,无需任何注解;缓存模块与 Stream 完全独立——只用缓存、不注册任何 handler 也能工作,反之亦然。

依赖与连接配置同 Stream:引入本 starter、配好 spring.redis.* 即自动装配出 RedisCacheEnhanceTemplate Bean(在 RedisAutoConfiguration 之后生效)。若上下文内没有 StringRedisTemplate Bean(例如自定义了多个 RedisConnectionFactory、或 exclude 了 RedisAutoConfiguration),缓存自动装配会整体退场而非启动失败;此时如需缓存请自行提供 StringRedisTemplate。

10.1 快速上手

注入 RedisCacheEnhanceTemplate,用 getOrLoad 一行搞定「读缓存 → 未命中回源 → 写回」,三防自动生效:

import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import org.springframework.stereotype.Service;

@Service
public class UserCache {

    private final RedisCacheEnhanceTemplate cache;

    public UserCache(RedisCacheEnhanceTemplate cache) {
        this.cache = cache;
    }

    public User getUser(Long id) {
        // 命中即返回;未命中则回源 loadFromDb 并写回,自动防穿透/雪崩/击穿
        // 第三参是 TTL(秒,必须 > 0);想用默认 TTL 可省略该参数
        return cache.getOrLoad("user:" + id, User.class, 3600, () -> loadFromDb(id));
    }

    public void evictUser(Long id) {
        cache.delete("user:" + id);          // 失效
    }
}

不配置任何 redis-cache.* 即用默认值(default-ttl 1 小时、抖动 10%、击穿互斥锁开启),开箱即用。需要直接读写时另有 set / get / exists / expire,见下。

10.2 API 一览与语义

方法	语义 / 边界
set(key, value[, ttlSeconds])	写缓存(fastjson 序列化)。真实值 TTL 叠加随机抖动(防雪崩);value 为 null 时写空值标记并改用短 null-ttl-seconds;不传 ttlSeconds 用 default-ttl-seconds。ttlSeconds 必须 > 0,否则抛 IllegalArgumentException。
get(key, Class<T>) get(key, TypeReference<T>)	读缓存,支持泛型。未命中或命中空值标记 → 返回 null;脏数据(无法反序列化)→ 抛 IllegalStateException(显式暴露便于排查)。
exists(key)	是否存在真实值。空值标记与未命中均返回 false(与 get 口径一致,读值判断而非物理 EXISTS)。
delete(key)	删除,返回是否确实删除了该 key。
expire(key, ttlSeconds)	显式续期(走抖动)。key 不存在 → false;ttlSeconds 必须 > 0,否则抛 IllegalArgumentException(避免传 0/负被 Redis 当成「立即删除」)。
getOrLoad(key, type, loader) getOrLoad(key, type, ttlSeconds, loader) getOrLoad(key, TypeReference, ttlSeconds, loader)	读取或回源,三防齐全。ttlSeconds 缺省用 default-ttl-seconds,显式传入时必须 > 0;loader 返回 null 会被缓存为空值标记(短 null-ttl)。

两套反序列化语义:公共 get 遇脏数据抛 IllegalStateException(让坏数据显式暴露);getOrLoad 内部读到脏数据则当未命中降级回源并覆盖重建(自愈)。

10.3 三防与滑动续期(工作原理)

• 穿透:loader 回源得到 null 时写入空值标记 + 短 null-ttl-seconds(不抖动);后续命中空标记直接返回 null,不再反复打到后端。
• 雪崩:真实值写入时 TTL 叠加 [0, ttl × jitter-ratio) 的随机抖动(半开区间,只增不减),打散集中过期。jitter-ratio = 0 关闭抖动。
• 击穿:未命中时用 SET NX PX(setIfAbsent + lock-ttl-ms)抢互斥锁:
  • 抢到锁者先 double-check(再读一次,吃掉「刚有人写好」的窗口),仍未命中才回源、写回,finally 用 Lua 脚本按 token 匹配释放锁(只删自己的锁,不误删他人)。
  • 未抢到锁者按 wait-interval-ms 自旋重读,读到值即返回;超时(wait-timeout-ms)或线程被中断则降级直接 loader.get() 返回。
  • 关闭(breakdown.enabled = false)则退化为无锁直接回源。
• 滑动续期(sliding-expire-enabled = true):读命中真实值时自动续 default-ttl-seconds;空值标记不续(保持穿透防护的短 TTL);回源新写入自带 TTL、不重复续。

⚠️ 降级取舍:持锁者回源若慢于 wait-timeout-ms(默认 3s)或回源抛异常,等待者会各自降级回源——瞬时被放大为 N 次回源,且降级路径不写缓存(有意为之,优先可用性)。务必让 loader 回源耗时显著小于 wait-timeout-ms。

10.4 配置项(前缀 redis-cache,均可选)

不配即用注释中的默认值。Spring 宽松绑定,null-ttl-seconds / nullTtlSeconds 等写法等价。

redis-cache:
  key-prefix: "cache:"           # 统一 key 前缀;须自带分隔符(如尾冒号),否则 cache + user → cacheuser
  env-isolation: false           # 按环境隔离 key:开启且 env 非空 → key 形如 cache:env:bizKey
  env: ""                        # 环境标识,隔离开启且非空时拼入 key
  default-ttl-seconds: 3600      # set/getOrLoad 不传 TTL 时的默认过期(秒);必须 > 0
  null-ttl-seconds: 60           # 空值标记 TTL(秒)——穿透防护,宜短;必须 > 0
  jitter-ratio: 0.1              # TTL 随机抖动比例(雪崩防护),取值 [0,1);0 关闭抖动
  sliding-expire-enabled: false  # 读命中真实值自动续 default-ttl(滑动过期;空值标记不续)
  breakdown:                     # 击穿:互斥锁
    enabled: true                # 关闭则退化为无锁直接回源
    lock-ttl-ms: 10000           # 锁持有 TTL(ms),防持锁线程崩溃死锁;必须 > 0
    wait-timeout-ms: 3000        # 未抢到锁的最长等待(ms),超时降级直接回源;>= 0
    wait-interval-ms: 50         # 未抢到锁的自旋重读间隔(ms);必须 > 0

上述取值在装配时即被校验:TTL 非正、jitter-ratio 越界 [0,1)、lock-ttl-ms / wait-interval-ms 非正等会让上下文启动即失败(fail-fast),而非留到运行期才暴露。

10.5 必读约定与坑

• TTL 必须为正:set / expire / getOrLoad(显式 TTL)传入 ttlSeconds ≤ 0 一律抛 IllegalArgumentException。Redis 对 EXPIRE 非正值的语义是「立即删除 key」,框架拦在入口避免「续期变删数据」。本模板面向带 TTL 的缓存,不支持「永不过期」。
• key-prefix 需自带分隔符:key 按 前缀 + (env + ':') + bizKey 直接拼接,前缀不带尾分隔符会粘连(cache + user:1 → cacheuser:1)。默认 cache: 已自带冒号。bizKey 不可为空白,否则抛 IllegalArgumentException。
• exists 不等于物理存在:只在有真实值时返回 true,空值标记返回 false(与 get 一致)。
• 缺 Redis 自动退场:无 StringRedisTemplate Bean 时缓存装配整体 back off、不报错;此时注入 RedisCacheEnhanceTemplate 会因无此 Bean 而失败。
• 脏数据两套语义:公共 get 抛 IllegalStateException;getOrLoad 内部当未命中降级并重建。
• 覆盖默认 Bean:CacheKeyBuilder / CacheValueCodec / RedisCacheEnhanceTemplate 均 @ConditionalOnMissingBean,自定义同类型 Bean 即可覆盖。

缓存模块与 Stream 完全独立:只用缓存不注册任何 handler 也能工作;反之亦然。