Compare commits

...

31 Commits

Author SHA1 Message Date
root
3e72653d1f test(stream): 补 autoclaim 多实例并发幂等真 Redis 集成测试
两个 consumerName 不同的 StreamAutoClaimer 用 CyclicBarrier 同时对同一条 idle 超时的
PEL 消息发 XCLAIM,断言业务只被处理一次、PEL 清零。端到端覆盖 mock 单测与单实例死信
集成测试都证明不了的真实并发竞态(XCLAIM 归属转移 + idle 重置由 Redis 串行裁判),为
B1「先抢占再处理」补上端到端铁证。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 14:33:29 +08:00
root
a2965dd80e fix(stream): autoclaim 先 XCLAIM 抢占再处理,杜绝多实例重复消费/重复死信
claimOne 此前死信分支不抢占、正常分支抢占但不校验结果,多实例并发认领同一条
PEL 消息时会被重复 dispatch、重复落死信。改 claim 返回是否真抢到(XCLAIM 不带
JUSTID 返回认领到的消息列表,空即未抢到),claimOne 先抢占成功才 dispatch / 落死信。
补 claimOne mock 单测(抢到/抢不到 × 正常/死信 共 4 例)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 14:21:38 +08:00
root
e30faf4cff fix(stream): 缺 StringRedisTemplate Bean 时自动装配优雅退场,对齐 cache
stream 装配此前仅 @ConditionalOnClass,缺 StringRedisTemplate Bean 时
(如 exclude RedisAutoConfiguration)会因构造参数缺依赖而启动失败。加
@ConditionalOnBean(StringRedisTemplate) 使整类 back off,与 cache 一致;
顺带 @Configuration(proxyBeanMethods=false) 对齐风格。补 3 个装配测试。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 11:41:56 +08:00
root
7101a1fa65 docs(cache): README 缓存章节重写为「快速上手 + 详细参考」
第十节拆为 10.1 快速上手 + 10.2 API 语义 + 10.3 三防原理 + 10.4 配置 + 10.5
约定与坑,并纳入本轮行为变更:ttl 必须 > 0、exists 排除空值标记、set(null) 用
null-ttl、key-prefix 需自带分隔符、缺 StringRedisTemplate 自动退场、配置装配期
fail-fast、抖动半开区间、降级不写缓存等。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 10:16:56 +08:00
root
0657126f60 test(cache): 补关键路径覆盖(loader 异常/double-check/泛型重载/sliding 回源/降级)
补齐 review 指出的覆盖缺口,均为既有正确行为的回归保护:loader 抛异常时锁 finally
释放 + 异常透传 + 不写缓存、抢锁后 double-check 命中跳过回源、getOrLoad 的
TypeReference 泛型重载往返、sliding 开启时回源新写入不重复续期、等待超时降级不写缓存。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 10:11:48 +08:00
root
24422d0ce4 refactor(cache): 纵深防御与风格清理(autoType/锁key/抖动上界/key校验/proxyBeanMethods)
- CacheValueCodec.decode 关闭 fastjson @type 自动类型(Feature.IgnoreAutoType):
  纵深防御;经验证常规链路本已安全(指定目标类型时 @type 被忽略),此为显式锁定契约。
- 模板锁 key 改用 keyBuilder.buildLock(key),与 CacheKeyBuilder 单一来源,消除重复派生。
- applyJitter 上界改 nextLong(bound) 对齐 spec 半开区间 [ttl, ttl*(1+ratio));测试收紧为严格 <。
- CacheKeyBuilder.build 对空白 bizKey fail-fast,避免 "cache:null" 静默键碰撞。
- RedisCacheAutoConfiguration 用 @Configuration(proxyBeanMethods=false),省 CGLIB 代理开销。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 10:08:56 +08:00
root
6a4f5e460e fix(cache): 公共 set(key,null) 改用短 null-ttl
set(key,null) 原按数据 TTL(可达 default-ttl-seconds)写入空值标记,使空标记长期
占位;改为与回源空值一致用 null-ttl-seconds(不抖动),空标记宜短。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 10:03:39 +08:00
root
c40f153642 fix(cache): exists 排除空值标记,与 get 口径一致
exists 原用 hasKey,会把持有空值标记的 key 判为"存在"(true),而 get 对同一 key
返回 null,二者口径打架。改为读值判断:空标记/缺失均返回 false,真实值返回 true。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 10:00:06 +08:00
root
8b90fe099e fix(cache): 缺 StringRedisTemplate Bean 时自动装配优雅退场
原仅 @ConditionalOnClass(类在 classpath 恒真),StringRedisTemplate Bean 缺失
(多 ConnectionFactory 致 @ConditionalOnSingleCandidate 不满足,或 exclude
RedisAutoConfiguration)时会因构造参数缺依赖而启动失败。加
@ConditionalOnBean(StringRedisTemplate),配合既有
@AutoConfigureAfter(RedisAutoConfiguration) 在 Redis 装配后求值,缺 Bean 时整类
back off 而非报错。仅改 cache,不动 stream(接受两者风格暂不一致)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 09:58:00 +08:00
root
1e673cdf83 fix(cache): 构造期校验配置项,误配在装配阶段即 fail-fast
default/null-ttl-seconds、jitter-ratio、breakdown 的 lock-ttl-ms/wait-timeout-ms/
wait-interval-ms 原无校验,负值/越界要到运行期(写入非法 TTL、抖动放大数倍)才暴露。
在 RedisCacheEnhanceTemplate 构造器手工校验,非法即抛 IllegalArgumentException;
不引 hibernate-validator,守住零新依赖。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 09:54:55 +08:00
root
0facafc745 fix(cache): 击穿等待响应中断 + interval 兜底,消除空转与无限循环
未抢到锁的自旋等待原以 waited += waitIntervalMs 推进:waitIntervalMs<=0 时永不
前进 → 无限循环 hang;且 sleep 吞中断后不退出,被中断时以 Redis 往返为节奏空转
到超时、不响应协作式取消。改为溢出安全的 deadline 截止判断 + Math.max(1,interval)
兜底,sleep 返回中断信号、被中断即 break 降级并保留中断标志。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 09:51:42 +08:00
root
ba2606ea46 fix(cache): set/expire/getOrLoad 对 ttl<=0 fail-fast,杜绝 expire 误删 key
expire(key, ttl<=0) 经 applyJitter 透传 0/负值,Redis EXPIRE 非正值会立即删 key
并返回 1,使"续期"静默删数据且报 true;set(ttl<=0) 则下发非法过期抛
RedisSystemException。统一在 set/expire/doGetOrLoad 入口 requirePositiveTtl 校验,
非正 TTL 直接抛 IllegalArgumentException。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 09:46:55 +08:00
root
8dc714b0ad docs: 修正缓存 plan 中 Mockito API 笔误(openMocks→initMocks,适配项目 3.3.3) 2026-06-26 07:18:32 +08:00
root
da3b77e19c docs(cache): README 增补缓存章节(用法与配置) 2026-06-26 07:17:02 +08:00
root
2d6a5be690 docs(stream): 七、配置项改写为带注释 YAML 并补充 Stream 裁剪说明 2026-06-26 07:15:22 +08:00
root
effa7bf38a test(cache): 集成测试击穿并发(互斥锁下 loader 仅回源一次)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-26 05:10:35 +08:00
root
6c512ba3b2 test(cache): 集成测试显式 expire 续期与滑动过期(真实值续、空标记不续) 2026-06-26 05:08:25 +08:00
root
b5a649110e test(cache): 集成测试穿透防护(空值缓存,loader 仅调用一次) 2026-06-26 05:06:05 +08:00
root
9abff913e5 test(cache): 集成测试 set/get/exists/delete 闭环 + Redis 探测脚手架 2026-06-26 05:03:19 +08:00
root
c4ad3670fc feat(cache): 自动装配 + spring.factories 注册 2026-06-26 04:59:41 +08:00
root
037fdc4f9c feat(cache): getOrLoad 三防(空值穿透 + TTL 抖动雪崩 + 互斥锁击穿 + 超时降级) 2026-06-26 04:55:42 +08:00
root
d164b89067 feat(cache): 模板读写基础(set/get/delete/exists/expire + TTL 抖动 + 滑动续期) 2026-06-26 04:51:07 +08:00
root
9605dca962 feat(cache): 配置属性 redis-cache.*(TTL/抖动/滑动/击穿锁) 2026-06-26 04:43:28 +08:00
root
ecbdfd8907 feat(cache): 值编解码器(fastjson 序列化 + 空值标记) 2026-06-26 04:41:02 +08:00
root
ce370afa83 feat(cache): 缓存常量与 key 构造器(前缀 + 环境隔离 + 锁 key 派生) 2026-06-26 04:38:02 +08:00
root
40882ecfaa docs: 新增 Redis 缓存增强模板实现计划
11 个 TDD 任务:常量/KeyBuilder → Codec → Properties → 模板读写 →
getOrLoad 三防 → 自动装配 → 4 个集成测试(RoundTrip/穿透/expire-滑动/击穿并发)
→ README + 全量回归。含完整代码、精确命令与对照 spec 的自检。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 21:01:19 +08:00
root
ec2a8daf4b docs: 新增 Redis 缓存增强模板设计文档
在现有 starter 内补齐编程式缓存能力的设计 spec:RedisCacheEnhanceTemplate
提供 get/set/delete/exists/expire/getOrLoad;getOrLoad 内置穿透/雪崩/击穿三防,
支持显式 expire 与可选滑动过期;独立 com.njcn.middle.cache 包、配置前缀 redis-cache,
不改动 stream、不新增依赖。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 20:49:08 +08:00
root
b649d6151d chore: 清理 StreamTrimScheduler 过时 javadoc 与死代码
裁剪逻辑已改用 opsForStream().trim() 精确裁剪,但 javadoc 仍写着
XTRIM ... MAXLEN ~(近似裁剪),且 trimAll() 里残留注释掉的旧
connection.execute("XTRIM", ...) 死代码,二者均与现状矛盾。

- javadoc 改为精确裁剪,并注明 2.3.x 无近似重载故不用 ~
- 删除注释掉的旧 XTRIM 裸命令死代码
- 删除随之失效的 import(RedisCallback、StandardCharsets)

注:旧裸命令报错的真因是 ByteArrayOutput 无法解析 XTRIM 的整数
回复(CommandOutput.set(long) 抛 IllegalStateException),与 ~ 无关。
纯注释/import 清理,逻辑零改动,compile 通过。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 16:21:21 +08:00
root
373b488249 Merge remote-tracking branch 'origin/main' 2026-06-25 14:47:01 +08:00
root
e3dc7aedf6 docs: add redis stream demo implementation plan 2026-06-25 11:05:07 +08:00
root
eb70ffabf2 docs: add redis stream demo design 2026-06-25 11:02:01 +08:00
28 changed files with 4967 additions and 33 deletions

140
README.md
View File

@@ -222,18 +222,35 @@ public class OrderConsumer extends EnhanceStreamConsumerHandler<OrderMessage> {
## 七、配置项
| 配置项 | 默认值 | 说明 |
|---|---|---|
| `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 ~`) |
> 全部可配项如下,前缀 `redis-stream`,**均为可选**——不配即用注释中的默认值。Spring 宽松绑定,`read-count` / `readCount` 等写法等价。
```yaml
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 ~`)重载,故走精确裁剪。
---
@@ -242,7 +259,7 @@ public class OrderConsumer extends EnhanceStreamConsumerHandler<OrderMessage> {
- **多实例部署不要写死 `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)。
- **Stream 裁剪**:框架按 `trim.maxlen` 精确裁剪 stream 防止无界增长(详见「七、配置项 · Stream 长度裁剪」)。请确保 `maxlen` 足够大,避免重试中的消息在达到 `max-retry` 前被裁掉(被裁掉的 PEL 消息会被静默强制 ACK)。
- **无 handler 时容器空转**:未注册任何 `EnhanceStreamConsumerHandler` Bean 时,监听/认领/裁剪线程均不启动,仅 `RedisStreamEnhanceTemplate` 可用于纯发送场景。
---
@@ -255,3 +272,100 @@ public class OrderConsumer extends EnhanceStreamConsumerHandler<OrderMessage> {
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` 一行搞定「读缓存 → 未命中回源 → 写回」,三防自动生效:
```java
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>)`<br>`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)`<br>`getOrLoad(key, type, ttlSeconds, loader)`<br>`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` 等写法等价。
```yaml
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 也能工作;反之亦然。

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,595 @@
# Redis Stream Demo Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Build an independent Spring Boot demo project that imports `redis-stream-springboot-starter` and verifies real Redis Stream publish/subscribe behavior.
**Architecture:** The demo is a downstream Maven Spring Boot application under `C:\code\hatch\redis-stream-starter-demo`. It depends on the starter from the local Maven repository, sends a `DemoMessage` with `RedisStreamEnhanceTemplate`, and consumes it with an `EnhanceStreamConsumerHandler` implementation. Integration testing uses the provided Redis service when reachable and skips when unreachable.
**Tech Stack:** Java 8, Maven, Spring Boot `2.3.12.RELEASE`, JUnit 5, `redis-stream-springboot-starter:1.0.0-SNAPSHOT`, Redis Stream.
---
## File Structure
- Create: `C:\code\hatch\redis-stream-starter-demo\pom.xml`
- Maven project definition, Spring Boot 2.3.12 dependency management, starter dependency, test setup.
- Create: `C:\code\hatch\redis-stream-starter-demo\.gitignore`
- Keeps build output and IDE files out of the demo project.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\resources\application.yml`
- Redis host/port defaults and overridable password environment variable.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoApplication.java`
- Spring Boot entry point.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoMessage.java`
- Demo message payload extending `StreamBaseMessage`.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoRuntimeInbox.java`
- In-memory queue shared by runtime logging and integration tests.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoProducer.java`
- Sends demo messages through the starter template.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoConsumer.java`
- Starter consumer handler for `demo-topic`.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoStartupRunner.java`
- Sends one demo message when running the app manually.
- Create: `C:\code\hatch\redis-stream-starter-demo\src\test\java\com\njcn\demo\redisstream\DemoRoundTripIntegrationTest.java`
- Real Redis publish/subscribe integration test.
## Task 1: Prepare Build Scaffolding
**Files:**
- Create: `C:\code\hatch\redis-stream-starter-demo\pom.xml`
- Create: `C:\code\hatch\redis-stream-starter-demo\.gitignore`
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\resources\application.yml`
- [ ] **Step 1: Install the starter into the local Maven repository**
Run from the starter repository:
```bash
mvn install
```
Expected: `BUILD SUCCESS`, with `redis-stream-springboot-starter-1.0.0-SNAPSHOT.jar` installed into the local Maven repository.
- [ ] **Step 2: Create demo project directories**
Run:
```powershell
New-Item -ItemType Directory -Force `
'C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream', `
'C:\code\hatch\redis-stream-starter-demo\src\main\resources', `
'C:\code\hatch\redis-stream-starter-demo\src\test\java\com\njcn\demo\redisstream'
```
Expected: all three directories exist.
- [ ] **Step 3: Create `pom.xml`**
Write this exact content:
```xml
<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.njcn.demo</groupId>
<artifactId>redis-stream-starter-demo</artifactId>
<version>1.0.0-SNAPSHOT</version>
<packaging>jar</packaging>
<name>redis-stream-starter-demo</name>
<description>Demo application for redis-stream-springboot-starter</description>
<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>
<spring-boot.version>2.3.12.RELEASE</spring-boot.version>
</properties>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-dependencies</artifactId>
<version>${spring-boot.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<dependencies>
<dependency>
<groupId>com.njcn</groupId>
<artifactId>redis-stream-springboot-starter</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<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>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<version>2.22.2</version>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
<version>${spring-boot.version}</version>
</plugin>
</plugins>
</build>
</project>
```
- [ ] **Step 4: Create `.gitignore`**
Write this exact content:
```gitignore
target/
.idea/
*.iml
.classpath
.project
.settings/
```
- [ ] **Step 5: Create `application.yml`**
Write this exact content:
```yaml
spring:
redis:
host: ${DEMO_REDIS_HOST:192.168.1.22}
port: ${DEMO_REDIS_PORT:12379}
password: ${DEMO_REDIS_PASSWORD:}
redis-stream:
env-isolation: true
env: demo
consumer-name: demo-consumer
read-count: 10
block-ms: 500
max-retry: 1
```
- [ ] **Step 6: Verify build scaffolding resolves dependencies**
Run from the demo project:
```bash
mvn test -DskipTests
```
Expected: `BUILD SUCCESS`.
## Task 2: Write the Round-Trip Integration Test First
**Files:**
- Create: `C:\code\hatch\redis-stream-starter-demo\src\test\java\com\njcn\demo\redisstream\DemoRoundTripIntegrationTest.java`
- [ ] **Step 1: Write the failing test**
Write this exact content:
```java
package com.njcn.demo.redisstream;
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.test.context.SpringBootTest;
import org.springframework.data.redis.connection.stream.RecordId;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
@ExtendWith(DemoRoundTripIntegrationTest.RedisReachableCondition.class)
@SpringBootTest
class DemoRoundTripIntegrationTest {
@Autowired
private DemoProducer producer;
@Autowired
private DemoRuntimeInbox inbox;
@Test
void publishesAndConsumesDemoMessage() throws Exception {
inbox.clear();
String payload = "integration-" + System.currentTimeMillis();
DemoMessage sent = new DemoMessage();
sent.setPayload(payload);
sent.setSource("integration-test");
sent.setTag("demo");
RecordId recordId = producer.send(sent);
DemoMessage received = inbox.poll(10, TimeUnit.SECONDS);
assertNotNull(recordId, "XADD should return a Redis stream record id");
assertNotNull(received, "consumer should receive the sent message within 10 seconds");
assertEquals(sent.getKey(), received.getKey(), "message key should round-trip");
assertEquals(payload, received.getPayload(), "payload should round-trip");
assertEquals("integration-test", received.getSource(), "source should round-trip");
}
static class RedisReachableCondition implements ExecutionCondition {
@Override
public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {
String host = System.getProperty("spring.redis.host",
System.getenv().getOrDefault("DEMO_REDIS_HOST", "192.168.1.22"));
int port = Integer.parseInt(System.getProperty("spring.redis.port",
System.getenv().getOrDefault("DEMO_REDIS_PORT", "12379")));
try (Socket socket = new Socket()) {
socket.connect(new InetSocketAddress(host, port), 500);
return ConditionEvaluationResult.enabled("redis reachable at " + host + ":" + port);
} catch (Exception e) {
return ConditionEvaluationResult.disabled("redis unreachable at " + host + ":" + port);
}
}
}
}
```
- [ ] **Step 2: Run the test and verify it fails because application classes do not exist**
Run from the demo project with the Redis password supplied through the process environment:
```bash
mvn test -Dtest=DemoRoundTripIntegrationTest
```
Expected: compilation failure mentioning missing symbols such as `DemoProducer`, `DemoRuntimeInbox`, and `DemoMessage`.
## Task 3: Implement Minimal Demo Application Classes
**Files:**
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoApplication.java`
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoMessage.java`
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoRuntimeInbox.java`
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoProducer.java`
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoConsumer.java`
- [ ] **Step 1: Create `DemoApplication.java`**
Write this exact content:
```java
package com.njcn.demo.redisstream;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class DemoApplication {
public static void main(String[] args) {
SpringApplication.run(DemoApplication.class, args);
}
}
```
- [ ] **Step 2: Create `DemoMessage.java`**
Write this exact content:
```java
package com.njcn.demo.redisstream;
import com.njcn.middle.stream.domain.StreamBaseMessage;
public class DemoMessage extends StreamBaseMessage {
private String payload;
public String getPayload() {
return payload;
}
public void setPayload(String payload) {
this.payload = payload;
}
}
```
- [ ] **Step 3: Create `DemoRuntimeInbox.java`**
Write this exact content:
```java
package com.njcn.demo.redisstream;
import org.springframework.stereotype.Component;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@Component
public class DemoRuntimeInbox {
private final BlockingQueue<DemoMessage> messages = new LinkedBlockingQueue<>();
public void add(DemoMessage message) {
messages.add(message);
}
public DemoMessage poll(long timeout, TimeUnit unit) throws InterruptedException {
return messages.poll(timeout, unit);
}
public void clear() {
messages.clear();
}
}
```
- [ ] **Step 4: Create `DemoProducer.java`**
Write this exact content:
```java
package com.njcn.demo.redisstream;
import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.stereotype.Service;
@Service
public class DemoProducer {
public static final String TOPIC = "demo-topic";
private final RedisStreamEnhanceTemplate streamTemplate;
public DemoProducer(RedisStreamEnhanceTemplate streamTemplate) {
this.streamTemplate = streamTemplate;
}
public RecordId send(DemoMessage message) {
return streamTemplate.send(TOPIC, message, false);
}
}
```
- [ ] **Step 5: Create `DemoConsumer.java`**
Write this exact content:
```java
package com.njcn.demo.redisstream;
import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
@Component
public class DemoConsumer extends EnhanceStreamConsumerHandler<DemoMessage> {
public static final String GROUP = "demo-group";
private static final Logger log = LoggerFactory.getLogger(DemoConsumer.class);
private final DemoRuntimeInbox inbox;
public DemoConsumer(DemoRuntimeInbox inbox) {
this.inbox = inbox;
}
@Override
public String topic() {
return DemoProducer.TOPIC;
}
@Override
public String group() {
return GROUP;
}
@Override
public Class<DemoMessage> messageType() {
return DemoMessage.class;
}
@Override
public void handleMessage(DemoMessage message) {
log.info("Consumed Redis Stream demo message: key={}, payload={}, source={}, tag={}",
message.getKey(), message.getPayload(), message.getSource(), message.getTag());
inbox.add(message);
}
@Override
public boolean isRetry() {
return false;
}
@Override
public boolean throwException() {
return false;
}
}
```
- [ ] **Step 6: Run the integration test and verify it passes**
Run from the demo project with the Redis password supplied through the process environment:
```bash
mvn test -Dtest=DemoRoundTripIntegrationTest
```
Expected: `BUILD SUCCESS`, and the test reports one successful test when Redis is reachable.
## Task 4: Add Manual Startup Publisher
**Files:**
- Create: `C:\code\hatch\redis-stream-starter-demo\src\main\java\com\njcn\demo\redisstream\DemoStartupRunner.java`
- [ ] **Step 1: Write the failing startup-runner test by extending the integration test**
Modify `DemoRoundTripIntegrationTest.java` to include this field and test method:
```java
@Autowired
private DemoStartupRunner startupRunner;
@Test
void startupRunnerPublishesDemoMessage() throws Exception {
inbox.clear();
startupRunner.run();
DemoMessage received = inbox.poll(10, TimeUnit.SECONDS);
assertNotNull(received, "startup runner should publish a message");
assertEquals("startup-runner", received.getSource(), "startup runner should set source");
assertEquals("hello redis stream starter", received.getPayload(), "startup runner payload should match");
}
```
The full test class should still keep the first `publishesAndConsumesDemoMessage` test unchanged.
- [ ] **Step 2: Run the test and verify it fails because `DemoStartupRunner` does not exist**
Run:
```bash
mvn test -Dtest=DemoRoundTripIntegrationTest
```
Expected: compilation failure mentioning missing symbol `DemoStartupRunner`.
- [ ] **Step 3: Create `DemoStartupRunner.java`**
Write this exact content:
```java
package com.njcn.demo.redisstream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.stereotype.Component;
@Component
public class DemoStartupRunner implements ApplicationRunner {
private static final Logger log = LoggerFactory.getLogger(DemoStartupRunner.class);
private final DemoProducer producer;
public DemoStartupRunner(DemoProducer producer) {
this.producer = producer;
}
@Override
public void run(ApplicationArguments args) {
run();
}
public RecordId run() {
DemoMessage message = new DemoMessage();
message.setPayload("hello redis stream starter");
message.setSource("startup-runner");
message.setTag("demo");
RecordId recordId = producer.send(message);
log.info("Published Redis Stream demo message: recordId={}, key={}, payload={}",
recordId, message.getKey(), message.getPayload());
return recordId;
}
}
```
- [ ] **Step 4: Run the integration test and verify both tests pass**
Run:
```bash
mvn test -Dtest=DemoRoundTripIntegrationTest
```
Expected: `BUILD SUCCESS`, with two successful tests when Redis is reachable.
## Task 5: Final Verification
**Files:**
- No new files.
- [ ] **Step 1: Run the starter build**
Run from the starter repository:
```bash
mvn install
```
Expected: `BUILD SUCCESS`.
- [ ] **Step 2: Run all demo tests**
Run from the demo project with the Redis password supplied through the process environment:
```bash
mvn test
```
Expected: `BUILD SUCCESS`.
- [ ] **Step 3: Run the demo application**
Run from the demo project with the Redis password supplied through the process environment:
```bash
mvn spring-boot:run
```
Expected: startup logs show a published Redis Stream message and a consumed Redis Stream message. Stop the process after the message is consumed.
- [ ] **Step 4: Capture final file status**
Run from the starter repository:
```bash
git status --short
```
Expected: only the implementation plan file is uncommitted unless execution commits it separately.
## Self-Review
- Spec coverage: Tasks cover local starter installation, independent demo project creation, Redis configuration, producer, consumer, startup publishing, integration testing, and manual runtime verification.
- Placeholder scan: No TBD/TODO markers are used. Redis password is intentionally not written into this committed plan.
- Type consistency: `DemoMessage`, `DemoProducer`, `DemoConsumer`, `DemoRuntimeInbox`, `DemoStartupRunner`, and `DemoRoundTripIntegrationTest` use matching package names and method signatures.

View File

@@ -0,0 +1,360 @@
# Redis 缓存增强模板 设计文档
- 日期:2026-06-25
- 状态:已评审待实现
- 所属:`redis-stream-springboot-starter`(在现有 starter 内**新增**缓存能力,不改动 stream)
---
## 一、背景与目标
当前 starter 只封装了 Redis Streams 收发,缺了最常用的**缓存**能力。本次在同一个 starter 内补齐一个**编程式缓存增强模板**,把缓存里最容易踩坑的三件事——**TTL 管理、序列化、三大经典问题(穿透/雪崩/击穿)防护**——全部封装进去,业务侧一行 `getOrLoad` 即可拿到带完整防护的缓存读取。
设计原则与现有 stream 端保持一致:
- **零业务依赖、开箱即用**:引入依赖 + 配好 Redis 连接即可注入使用,无需 `@EnableXxx`
- **编程式注入模板**:对齐现有 `RedisStreamEnhanceTemplate` 的使用气质,提供 `RedisCacheEnhanceTemplate`
- **与 stream 完全解耦**:独立包、独立配置前缀、独立自动装配类,互不影响;只用缓存不用 stream 时缓存仍可单独工作(反之亦然)。
- **复用既有依赖**:`StringRedisTemplate` + `fastjson` + `hutool`,不新增任何依赖。
### 目标(In Scope)
1. 编程式缓存模板 `RedisCacheEnhanceTemplate`:`get / set / delete / exists / expire / getOrLoad`;其中 `expire` 为显式续期,并支持**可选的读时自动续期(滑动过期,默认关闭)**。
2. `getOrLoad` 内置三防:
- **穿透**:回源结果为 `null` 时缓存空值标记(短 TTL),后续直接命中空标记返回 `null`
- **雪崩**:写入 TTL 叠加随机抖动,避免大量 key 同时失效。
- **击穿**:热点 key 失效时用 Redis 分布式互斥锁,保证只有一个线程回源重建,其余等待后重读。
3. 统一 Key 前缀 + 环境隔离(复用 stream 的环境隔离思路,缓存独立配置)。
4. fastjson 序列化,支持泛型读回(`Class<T>` / `TypeReference<T>`)。
5. 独立配置项 `redis-cache.*`,全部可选、有合理默认。
6. 单元测试 + 集成测试(沿用现有"Redis 不可达自动 skip"约定)。
### 非目标(Out of Scope,YAGNI)
- 注解式缓存(`@Cacheable` 等)——Spring 自带,本次不做。
- 批量操作(`mget`/`mset`)、`increment`、分布式锁通用工具——本次不做,留待真实需求。
- 多级缓存(本地 + Redis)、缓存预热、监控指标——不做。
- 击穿的"逻辑过期"方案——已选定互斥锁方案,不实现逻辑过期。
-`artifactId`/坐标——见「十二、备注」,本次不动。
---
## 二、整体设计与包结构
新增 `com.njcn.middle.cache` 包,与现有 `com.njcn.middle.stream` 平行:
```
com.njcn.middle.cache
├── autoconfig
│ ├── RedisCacheProperties.java # 配置属性,前缀 redis-cache
│ └── RedisCacheAutoConfiguration.java # 自动装配(独立于 stream)
├── support
│ ├── CacheKeyBuilder.java # key 拼装:前缀 + 环境隔离
│ └── CacheValueCodec.java # 值 <-> 字符串(fastjson + 空值标记)
├── constant
│ └── CacheConstant.java # 空值标记 sentinel、锁后缀等常量
└── template
└── RedisCacheEnhanceTemplate.java # 核心:get/set/delete/exists/getOrLoad
```
每个单元职责单一、可独立测试:
| 单元 | 做什么 | 依赖 |
|---|---|---|
| `CacheKeyBuilder` | 业务 key → 实际 Redis key(加前缀,开隔离时加 env);派生锁 key | 配置(prefix/env) |
| `CacheValueCodec` | 对象 ↔ 字符串:`encode(obj)``decode(str, type)`;空值标记常量与判定 | fastjson |
| `RedisCacheEnhanceTemplate` | 对外 API;编排三防逻辑(锁、抖动、空值) | `StringRedisTemplate`、上面两者、配置 |
| `RedisCacheProperties` | 绑定 `redis-cache.*` | — |
| `RedisCacheAutoConfiguration` | 装配上述 Bean(均 `@ConditionalOnMissingBean`) | `StringRedisTemplate` |
`spring.factories` 追加 `RedisCacheAutoConfiguration`(与现有 `RedisStreamAutoConfiguration` 并列两行):
```
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration,\
com.njcn.middle.cache.autoconfig.RedisCacheAutoConfiguration
```
> 缓存模块**无后台线程**(纯模板),不像 stream 有监听/认领/裁剪线程,因此即使不调用也只是多一个空闲 Bean,零开销。
---
## 三、配置项(前缀 `redis-cache`,全部可选)
```yaml
redis-cache:
# ===== Key =====
key-prefix: "cache:" # 统一 key 前缀,便于在 Redis 中归类/批量清理
env-isolation: false # 是否按环境隔离 key(与 stream 同思路,但独立配置)
env: "" # 环境标识,env-isolation=true 且非空时拼入 key
# ===== TTL =====
default-ttl-seconds: 3600 # set/getOrLoad 不传 TTL 时的默认过期(秒)
null-ttl-seconds: 60 # 空值标记的 TTL(秒)——穿透防护,宜短
jitter-ratio: 0.1 # TTL 随机抖动比例——雪崩防护;0 表示关闭抖动
# ===== 滑动过期(读时自动续期)=====
sliding-expire-enabled: false # 开启后:读命中真实值即自动续 default-ttl-seconds(走抖动);命中空值标记不续
# ===== 击穿:互斥锁 =====
breakdown:
enabled: true # 关闭后 getOrLoad 退化为"无锁直接回源"
lock-ttl-ms: 10000 # 锁持有 TTL(ms),防持锁线程崩溃导致死锁
wait-timeout-ms: 3000 # 未抢到锁的线程最长等待(ms),超时降级直接回源
wait-interval-ms: 50 # 未抢到锁时自旋重读缓存的间隔(ms)
```
对应 `RedisCacheProperties`:顶层字段 + 内嵌 `Breakdown` 静态类(参照现有 `RedisStreamProperties``Autoclaim`/`Trim` 写法)。
> 说明:**数据 TTL 用秒**(缓存语义直觉),**锁相关用毫秒**(锁是亚秒级,毫秒更合适);命名后缀 `-seconds`/`-ms` 明确区分,避免单位混淆。
---
## 四、Key 设计(`CacheKeyBuilder`)
```
基础: keyPrefix + bizKey
开启环境隔离: keyPrefix + env + ":" + bizKey
击穿锁 key: <上面得到的 dataKey> + ":__lock__"
```
示例(`key-prefix=cache:`):
| 配置 | `build("user:1")` | 锁 key |
|---|---|---|
| 默认 | `cache:user:1` | `cache:user:1:__lock__` |
| `env-isolation=true, env=prod` | `cache:prod:user:1` | `cache:prod:user:1:__lock__` |
锁 key 由数据 key **派生**(同前缀、同环境),保证数据与其锁始终落在同一隔离域。
> 与 stream 的差异:stream 环境隔离用**后缀** `topic_env`;缓存用**前缀段** `prefix:env:key` 更符合 Redis key 的层级命名习惯,且便于 `SCAN cache:prod:*` 按环境清理。两者各自独立,不要求一致。
---
## 五、序列化与空值标记(`CacheValueCodec`)
- 底层用 `StringRedisTemplate`(与 stream 端一致),value 一律存**字符串**。
- 非空值:`JSON.toJSONString(value)`(fastjson)。读回:`JSON.parseObject(str, Class<T>)``JSON.parseObject(str, TypeReference<T>)`,支持泛型(如 `List<User>`)。
- **空值标记(sentinel)**:选用一个正常 JSON 绝不可能产生的字符串常量,内含 NUL 控制字符:
```java
// CacheConstant.java —— 用 Java unicode 转义书写,\u0000 为 NUL 控制字符
public static final String NULL_SENTINEL = "\u0000NULL\u0000";
```
fastjson 序列化任何对象都不会产出含 `\u0000` 的裸文本,故与真实值无冲突风险。判定空值即 `NULL_SENTINEL.equals(raw)`。
### 三态读取
`getOrLoad` 必须区分"**未命中**(要回源)"与"**命中空值标记**(直接返回 null,不回源)"。为此 codec/模板内部按三态处理,而**对外** `get` 接口仍只返回 `T`(两种 null 情形对调用方都表现为 `null`):
| Redis 原始返回 | 含义 | `get` 返回 | `getOrLoad` 行为 |
|---|---|---|---|
| `null` | 未命中 | `null` | 进入回源流程(抢锁/回源) |
| `NULL_SENTINEL` | 命中空值标记 | `null` | 直接返回 `null`,**不回源** |
| 其它字符串 | 命中真实值 | 反序列化后的 `T` | 直接返回 |
实现上,模板内部用一个轻量三态(如枚举 `MISS / HIT_NULL / HIT` + 值)承载读取结果,对外 API 不暴露。
### 反序列化失败的处理
- **`get(...)` 公共方法**:反序列化失败抛 `IllegalStateException`(脏数据应显式暴露,而非静默吞掉)。
- **`getOrLoad(...)` 内部读取**:捕获反序列化异常 → 记 `warn` 日志 → **当作未命中处理**(降级回源重建),避免一条脏缓存让该 key 永久读失败。
---
## 六、核心 API
```java
public class RedisCacheEnhanceTemplate {
// ---- 写 ----
void set(String key, Object value); // 用 default-ttl-seconds
void set(String key, Object value, long ttlSeconds); // 指定 TTL;内部叠加雪崩抖动
// ---- 读 ----
<T> T get(String key, Class<T> type);
<T> T get(String key, TypeReference<T> type);
// ---- 删 / 判存 ----
boolean delete(String key);
boolean exists(String key);
// ---- 续期 ----
boolean expire(String key, long ttlSeconds); // 仅对已存在 key 生效;TTL 走雪崩抖动
// ---- 读取或回源(三防核心)----
<T> T getOrLoad(String key, Class<T> type, Supplier<T> loader); // 用 default-ttl
<T> T getOrLoad(String key, Class<T> type, long ttlSeconds, Supplier<T> loader);
<T> T getOrLoad(String key, TypeReference<T> type, long ttlSeconds, Supplier<T> loader);
}
```
- `set(key, value, ttlSeconds)`:写入时实际 TTL = `applyJitter(ttlSeconds)`(见七)。
- `loader` 用 `java.util.function.Supplier<T>`,业务自定义回源逻辑(查 DB 等),返回 `null` 合法(表示数据不存在 → 触发空值缓存)。
- `expire(key, ttlSeconds)`:底层 `EXPIRE`,实际 TTL 同样走 `applyJitter`;key 不存在(已过期/从未写入)时返回 `false`。
### 滑动过期(读时自动续期,默认关闭)
`sliding-expire-enabled=true` 时,`get` 与 `getOrLoad` **命中真实值**会自动触发一次 `expire(key, default-ttl-seconds)`(走抖动),实现"最近被访问就持续保活"的滑动窗口语义。约定:
- **只续真实值**;**命中空值标记不续期**——否则"不存在"的 key 被反复查询会长期占位,削弱穿透防护的短 TTL 设计。
- **回源新写入不额外续期**(刚 `set` 过,TTL 已是最新)。
- 续期时长统一为 `default-ttl-seconds`(滑动是全局行为,不感知各 key 当初的自定义 TTL);需要按 key 精确续期请用显式 `expire`。
- 续期失败(并发下 key 恰好过期)忽略,不影响本次读取返回值。
---
## 七、`getOrLoad` 算法(三防细节)
```
getOrLoad(key, type, ttlSeconds, loader):
dataKey = keyBuilder.build(key)
// 1) 先读(三态);滑动过期开启且命中真实值时,顺带续期(空值标记不续)
r = tryGet(dataKey, type)
if r == HIT or r == HIT_NULL: # 命中真实值或空值标记
if slidingEnabled and r == HIT:
expire(dataKey, default-ttl-seconds)
return r.value # HIT_NULL 时为 null
# 2) 未命中 → 击穿保护
if not breakdown.enabled:
return loadAndCache(dataKey, ttlSeconds, loader) # 无锁,直接回源
lockKey = dataKey + ":__lock__"
token = UUID # hutool IdUtil.fastSimpleUUID()
locked = SET lockKey token NX PX lock-ttl-ms
if locked:
try:
r2 = tryGet(dataKey, type) # double-check:抢锁期间可能已被他人写入
if r2 == HIT or r2 == HIT_NULL:
return r2.value
return loadAndCache(dataKey, ttlSeconds, loader)
finally:
releaseLock(lockKey, token) # Lua:token 匹配才 del,防误删
else:
# 没抢到锁:自旋等待重读
waited = 0
while waited < wait-timeout-ms:
sleep(wait-interval-ms); waited += wait-interval-ms
r3 = tryGet(dataKey, type)
if r3 == HIT or r3 == HIT_NULL:
return r3.value
# 超时降级:直接回源返回(不写缓存,避免与持锁者竞争写),记 warn
log.warn("cache breakdown wait timeout, fallback to loader, key={}", key)
return loader.get()
loadAndCache(dataKey, ttlSeconds, loader):
value = loader.get()
if value == null:
SET dataKey NULL_SENTINEL EX null-ttl-seconds # 穿透防护
else:
SET dataKey JSON(value) EX applyJitter(ttlSeconds) # 雪崩防护
return value
applyJitter(ttlSeconds):
if jitter-ratio <= 0: return ttlSeconds
extra = floor(random() * ttlSeconds * jitter-ratio) # ThreadLocalRandom,[0, ttl*ratio)
return ttlSeconds + extra # 只增不减,不会提前失效
```
要点:
- **抢锁原子性**:`SET lockKey token NX PX <ttl>` 单命令完成"不存在才设并带过期",无 `SETNX` + `EXPIRE` 两步竞态。
- **释放锁安全**:用 Lua 脚本 `if GET(lockKey)==token then DEL(lockKey)`,避免业务回源耗时超过 `lock-ttl-ms`、锁已过期被他人重设时误删别人的锁。
- **double-check**:抢到锁后再读一次,吃掉"等待抢锁的瞬间持锁者刚写完"的窗口,避免重复回源。
- **超时降级**:等待超过 `wait-timeout-ms` 仍无值 → 直接 `loader.get()` 返回(不写缓存),牺牲一次回源换取**不被锁永久阻塞**的可用性。
- **抖动只增不减**:`applyJitter` 在原 TTL 上加 `[0, ttl*ratio)`,保证不会因抖动让缓存比预期更早失效。
---
## 八、自动装配(`RedisCacheAutoConfiguration`)
参照 `RedisStreamAutoConfiguration` 写法:
```java
@Configuration
@ConditionalOnClass(StringRedisTemplate.class)
@AutoConfigureAfter(RedisAutoConfiguration.class) // 确保 StringRedisTemplate 就绪
@EnableConfigurationProperties(RedisCacheProperties.class)
public class RedisCacheAutoConfiguration {
@Bean @ConditionalOnMissingBean
CacheKeyBuilder cacheKeyBuilder(RedisCacheProperties p) { ... }
@Bean @ConditionalOnMissingBean
CacheValueCodec cacheValueCodec() { ... }
@Bean @ConditionalOnMissingBean
RedisCacheEnhanceTemplate redisCacheEnhanceTemplate(
StringRedisTemplate redis, CacheKeyBuilder kb, CacheValueCodec codec, RedisCacheProperties p) { ... }
}
```
- 全部 `@ConditionalOnMissingBean`:业务可自定义覆盖。
- 与 stream 装配类**互不引用**;`spring.factories` 追加注册。
- 不依赖任何 `EnhanceStreamConsumerHandler`,无 `ObjectProvider` 收集逻辑。
---
## 九、错误处理与降级小结
| 场景 | 行为 |
|---|---|
| `set` 序列化失败 | 抛 `IllegalStateException`(包装原因) |
| `get` 反序列化失败 | 抛 `IllegalStateException`(脏数据显式暴露) |
| `getOrLoad` 内部读缓存反序列化失败 | 记 `warn`,当未命中 → 降级回源重建 |
| `loader` 返回 `null` | 缓存空值标记(`null-ttl-seconds`),返回 `null` |
| `loader` 抛异常 | 异常上抛(不写缓存,不吞);锁在 `finally` 释放 |
| 抢锁等待超时 | 记 `warn`,降级直接 `loader.get()` 返回(不写缓存) |
| `expire`/滑动续期目标 key 不存在 | 不报错;`expire` 返回 `false`,滑动续期静默跳过 |
| Redis 连接异常 | 上抛由调用方处理(本期不做 Redis 故障本地降级,YAGNI) |
---
## 十、测试策略
延续现有约定:单元测试 mock,集成测试连真实 Redis 且**不可达自动 skip**(复用 `StreamRoundTripTest.RedisReachableCondition` 同款 `ExecutionCondition`,`-Dspring.redis.host/port/password` 覆盖连接)。
### 单元测试(无需 Redis)
- `CacheKeyBuilderTest`:默认前缀、环境隔离前缀段、锁 key 派生。
- `CacheValueCodecTest`:对象/基本类型/泛型 `TypeReference` 往返;空值标记 encode/判定;脏字符串 decode 抛异常。
- `RedisCachePropertiesTest`:默认值、宽松绑定(`default-ttl-seconds`/`defaultTtlSeconds`)、内嵌 `breakdown.*`。
- `RedisCacheEnhanceTemplateTest`(mock `StringRedisTemplate`/`ValueOperations`):
- `getOrLoad` 命中真实值 → 不回源;
- 命中空值标记 → 返回 null 且不回源;
- 未命中 → 回源一次并写缓存;
- `loader` 返回 null → 写空值标记(TTL=null-ttl);
- `applyJitter` 结果落在 `[ttl, ttl*(1+ratio))`;`ratio=0` 时等于原值;
- 内部读取反序列化失败 → 降级回源;
- `expire`:已存在 key 返回 `true` 并刷新 TTL(走抖动),不存在 key 返回 `false`;
- 滑动开启时:命中真实值触发一次续期、命中空值标记**不**续期、回源新写入**不**重复续期。
### 集成测试(需 Redis,skip-if-unreachable)
- `CacheRoundTripTest`:`set` → `get` 往返一致;TTL 生效(`getExpire` 在区间内);`delete` 后不存在。
- `CachePenetrationTest`:`getOrLoad` 对不存在的数据回源返回 null → 二次调用命中空标记、**loader 不再被调用**(`AtomicInteger` 计数 == 1)。
- `CacheBreakdownConcurrencyTest`:N 个线程并发 `getOrLoad` 同一未命中热点 key,`loader` 内 `sleep` 模拟慢回源 → 断言 `loader` **仅被调用 1 次**,N 个线程都拿到同一结果。
- `CacheExpireSlidingTest`:显式 `expire` 后 `getExpire` 变长、对不存在 key 返回 `false`;开启 `sliding-expire-enabled` 后重复 `get` 命中真实值持续刷新 TTL,而空值标记的 TTL 不被刷新。
---
## 十一、对现有代码的影响
- **不修改** stream 任何现有类。
- 仅**新增** `com.njcn.middle.cache.*` 与对应测试。
- `spring.factories` 追加一行注册(并列,不影响 stream 装配)。
- 实现阶段在 `README.md` 增补「缓存」章节(用法与配置),与现有 stream 文档并列。
- 不新增第三方依赖。
---
## 十二、备注
- **`artifactId` 维持 `redis-stream-springboot-starter` 不变**:已发布坐标,贸然改名破坏下游引用。缓存功能以独立包形式并入;若未来 starter 能力进一步泛化,再单独评估是否重命名为更通用的坐标(本期不做)。
- 击穿锁是一把**够用的轻量分布式锁**(SET NX PX + Lua 释放),不抽象成对外通用锁工具——保持 YAGNI,仅服务于缓存重建。

View File

@@ -0,0 +1,116 @@
# Redis Stream Starter Demo Design
## Goal
Create an independent Spring Boot demo project at `C:\code\hatch\redis-stream-starter-demo` that imports the local `redis-stream-springboot-starter` and verifies Redis Stream message publish and subscribe behavior against the provided Redis service.
## Context
The current repository is a Maven single-module Spring Boot starter:
- Group/artifact/version: `com.njcn:redis-stream-springboot-starter:1.0.0-SNAPSHOT`
- Spring Boot baseline: `2.3.12.RELEASE`
- Java baseline: 8
- Auto-configuration entry: `META-INF/spring.factories`
- Producer API: `RedisStreamEnhanceTemplate#send(topic, message, compress)`
- Consumer API: subclass `EnhanceStreamConsumerHandler<T>`
## Demo Project Location
The demo project will be created outside this repository at:
```text
C:\code\hatch\redis-stream-starter-demo
```
This keeps the starter repository unchanged except for this design document and allows the demo to behave like a real downstream application.
## Dependency Flow
Before building the demo, install the starter into the local Maven repository:
```bash
mvn install
```
The demo `pom.xml` will then depend on:
```xml
<dependency>
<groupId>com.njcn</groupId>
<artifactId>redis-stream-springboot-starter</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
```
The demo will use Spring Boot `2.3.12.RELEASE` to match the starter's dependency baseline.
## Redis Configuration
The demo will use the provided Redis Stream service:
- Host: configured as `192.168.1.22`
- Port: configured as `12379`
- Password: configured in the local demo application config, not copied into this design document
The demo should allow overriding these values with environment variables or JVM system properties so the sample remains reusable.
## Runtime Components
The demo will include:
- `DemoApplication`: Spring Boot entry point.
- `DemoMessage`: message payload extending `StreamBaseMessage`.
- `DemoProducer`: injects `RedisStreamEnhanceTemplate` and publishes a message.
- `DemoConsumer`: extends `EnhanceStreamConsumerHandler<DemoMessage>` and subscribes to the same topic and group.
- `DemoStartupRunner`: sends one message after the application starts and logs the returned Redis `RecordId`.
The consumer will log received message content and store consumed messages in an in-memory queue for integration test assertions.
## Message Flow
1. Application starts and auto-configuration creates the starter beans.
2. `DemoStartupRunner` creates a `DemoMessage`.
3. `DemoProducer` calls `RedisStreamEnhanceTemplate#send("demo-topic", message, false)`.
4. The starter writes the message to Redis Stream with `XADD`.
5. The starter listener reads the message through `XREADGROUP`.
6. `DemoConsumer#handleMessage` receives the deserialized `DemoMessage`.
7. The starter ACKs the message after successful handling.
## Testing
Add an integration test that:
- Starts the demo Spring context.
- Uses the real Redis service when reachable.
- Sends a unique test message.
- Waits for the consumer to receive it.
- Asserts that the received message payload and key match the sent message.
If Redis is unreachable, the integration test should be skipped instead of failed. This keeps local builds deterministic while still validating the real publish/subscribe flow when the service is available.
## Error Handling
The demo consumer will return `false` for `isRetry()` so test failures are surfaced through assertions instead of retry loops. Unexpected handler exceptions will be logged by the starter path and ACKed according to the existing handler behavior.
## Verification Commands
From the starter repository:
```bash
mvn install
```
From the demo project:
```bash
mvn test
mvn spring-boot:run
```
## Out of Scope
- Publishing the starter to Nexus.
- Changing starter production code.
- Adding Docker or Testcontainers.
- Building a UI or HTTP API for the demo.

View File

@@ -0,0 +1,47 @@
package com.njcn.middle.cache.autoconfig;
import com.njcn.middle.cache.support.CacheKeyBuilder;
import com.njcn.middle.cache.support.CacheValueCodec;
import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
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;
/**
* Redis 缓存自动装配。装配在 {@link RedisAutoConfiguration} 之后,确保 {@link StringRedisTemplate} 就绪。
* 与 stream 装配互不引用;纯模板、无后台线程。
*/
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(StringRedisTemplate.class)
@ConditionalOnBean(StringRedisTemplate.class)
@AutoConfigureAfter(RedisAutoConfiguration.class)
@EnableConfigurationProperties(RedisCacheProperties.class)
public class RedisCacheAutoConfiguration {
@Bean
@ConditionalOnMissingBean
public CacheKeyBuilder cacheKeyBuilder(RedisCacheProperties props) {
return new CacheKeyBuilder(props.getKeyPrefix(), props.isEnvIsolation(), props.getEnv());
}
@Bean
@ConditionalOnMissingBean
public CacheValueCodec cacheValueCodec() {
return new CacheValueCodec();
}
@Bean
@ConditionalOnMissingBean
public RedisCacheEnhanceTemplate redisCacheEnhanceTemplate(StringRedisTemplate redis,
CacheKeyBuilder cacheKeyBuilder,
CacheValueCodec cacheValueCodec,
RedisCacheProperties props) {
return new RedisCacheEnhanceTemplate(redis, cacheKeyBuilder, cacheValueCodec, props);
}
}

View File

@@ -0,0 +1,48 @@
package com.njcn.middle.cache.autoconfig;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
* Redis 缓存配置属性,前缀 {@code redis-cache}。全部可选,有合理默认。
*/
@Data
@ConfigurationProperties("redis-cache")
public class RedisCacheProperties {
/** 统一 key 前缀。 */
private String keyPrefix = "cache:";
/** 是否按环境隔离 key。 */
private boolean envIsolation = false;
/** 环境标识(隔离开启且非空时拼入 key)。 */
private String env = "";
/** set/getOrLoad 不传 TTL 时的默认过期(秒)。 */
private long defaultTtlSeconds = 3600;
/** 空值标记 TTL(秒)——穿透防护,宜短。 */
private long nullTtlSeconds = 60;
/** TTL 随机抖动比例——雪崩防护;0 关闭抖动。 */
private double jitterRatio = 0.1;
/** 读命中真实值时自动续期(滑动过期),默认关闭。 */
private boolean slidingExpireEnabled = false;
private Breakdown breakdown = new Breakdown();
/** 击穿:互斥锁。 */
@Data
public static class Breakdown {
/** 关闭后 getOrLoad 退化为无锁直接回源。 */
private boolean enabled = true;
/** 锁持有 TTL(ms),防持锁线程崩溃死锁。 */
private long lockTtlMs = 10000;
/** 未抢到锁的线程最长等待(ms),超时降级直接回源。 */
private long waitTimeoutMs = 3000;
/** 未抢到锁时自旋重读间隔(ms)。 */
private long waitIntervalMs = 50;
}
}

View File

@@ -0,0 +1,19 @@
package com.njcn.middle.cache.constant;
/**
* Redis 缓存相关常量。
*/
public final class CacheConstant {
private CacheConstant() {
}
/**
* 空值标记:回源结果为 null 时写入此值(穿透防护)。
* 内含 NUL 控制字符(Java unicode 转义书写),fastjson 序列化任何对象都不会产出含 NUL 的裸文本,故无冲突。
*/
public static final String NULL_SENTINEL = "\u0000NULL\u0000";
/** 击穿互斥锁 key 后缀(由数据 key 派生)。 */
public static final String LOCK_SUFFIX = ":__lock__";
}

View File

@@ -0,0 +1,39 @@
package com.njcn.middle.cache.support;
import cn.hutool.core.util.StrUtil;
import com.njcn.middle.cache.constant.CacheConstant;
/**
* 缓存 key 构造器:统一前缀 +(可选)环境隔离段。
*
* <p>基础:{@code keyPrefix + bizKey};开启隔离且 env 非空:{@code keyPrefix + env + ":" + bizKey}。
* 锁 key 由数据 key 派生,保证数据与其锁落在同一隔离域。
*/
public class CacheKeyBuilder {
private final String keyPrefix;
private final boolean envIsolation;
private final String env;
public CacheKeyBuilder(String keyPrefix, boolean envIsolation, String env) {
this.keyPrefix = keyPrefix == null ? "" : keyPrefix;
this.envIsolation = envIsolation;
this.env = env;
}
/** 业务 key → 实际数据 key。 */
public String build(String bizKey) {
if (StrUtil.isBlank(bizKey)) {
throw new IllegalArgumentException("bizKey must not be blank");
}
if (envIsolation && StrUtil.isNotBlank(env)) {
return keyPrefix + env + ":" + bizKey;
}
return keyPrefix + bizKey;
}
/** 数据 key 派生锁 key。 */
public String buildLock(String bizKey) {
return build(bizKey) + CacheConstant.LOCK_SUFFIX;
}
}

View File

@@ -0,0 +1,38 @@
package com.njcn.middle.cache.support;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.alibaba.fastjson.parser.Feature;
import com.njcn.middle.cache.constant.CacheConstant;
/**
* 缓存值编解码:对象 ↔ 字符串(fastjson),并承载空值标记。
*
* <p>解码失败直接抛 fastjson 异常({@link RuntimeException}),由调用方决定显式暴露(public get)
* 还是当未命中降级(getOrLoad 内部)。
*/
public class CacheValueCodec {
/** 对象 → 字符串:null → 空值标记;否则 fastjson 序列化。 */
public String encode(Object value) {
if (value == null) {
return CacheConstant.NULL_SENTINEL;
}
return JSON.toJSONString(value);
}
/** 是否空值标记。 */
public boolean isNullSentinel(String raw) {
return CacheConstant.NULL_SENTINEL.equals(raw);
}
/** 字符串 → 对象(Class)。关闭 @type 自动类型(纵深防御:禁止脏值据 @type 实例化任意类)。 */
public <T> T decode(String raw, Class<T> type) {
return JSON.parseObject(raw, type, Feature.IgnoreAutoType);
}
/** 字符串 → 对象(TypeReference,支持泛型)。同样关闭 @type 自动类型。 */
public <T> T decode(String raw, TypeReference<T> type) {
return JSON.parseObject(raw, type, Feature.IgnoreAutoType);
}
}

View File

@@ -0,0 +1,317 @@
package com.njcn.middle.cache.template;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.TypeReference;
import com.njcn.middle.cache.autoconfig.RedisCacheProperties;
import com.njcn.middle.cache.constant.CacheConstant;
import com.njcn.middle.cache.support.CacheKeyBuilder;
import com.njcn.middle.cache.support.CacheValueCodec;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
/**
* Redis 缓存增强模板。
*
* <p>底层 {@link StringRedisTemplate} 存字符串,fastjson 序列化。本类(Task 4)实现读写基础;
* {@code getOrLoad} 三防(穿透/雪崩/击穿)由 Task 5 追加。
*/
@Slf4j
public class RedisCacheEnhanceTemplate {
/** 释放锁 Lua:token 匹配才删,防误删他人锁。 */
private static final DefaultRedisScript<Long> UNLOCK_SCRIPT = new DefaultRedisScript<>(
"if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
Long.class);
private final StringRedisTemplate redis;
private final CacheKeyBuilder keyBuilder;
private final CacheValueCodec codec;
private final RedisCacheProperties props;
public RedisCacheEnhanceTemplate(StringRedisTemplate redis, CacheKeyBuilder keyBuilder,
CacheValueCodec codec, RedisCacheProperties props) {
validateProps(props);
this.redis = redis;
this.keyBuilder = keyBuilder;
this.codec = codec;
this.props = props;
}
// ---- 写 ----
public void set(String key, Object value) {
set(key, value, props.getDefaultTtlSeconds());
}
public void set(String key, Object value, long ttlSeconds) {
requirePositiveTtl(ttlSeconds);
// 空值(写入空标记)按短 null-ttl 处理,与回源 null 一致,避免空标记长期占位
long ttl = (value == null) ? props.getNullTtlSeconds() : applyJitter(ttlSeconds);
redis.opsForValue().set(keyBuilder.build(key), codec.encode(value), ttl, TimeUnit.SECONDS);
}
// ---- 读 ----
public <T> T get(String key, Class<T> type) {
return doGet(key, raw -> codec.decode(raw, type));
}
public <T> T get(String key, TypeReference<T> type) {
return doGet(key, raw -> codec.decode(raw, type));
}
private <T> T doGet(String key, Function<String, T> decoder) {
String dataKey = keyBuilder.build(key);
String raw = redis.opsForValue().get(dataKey);
if (raw == null || codec.isNullSentinel(raw)) {
return null;
}
T value;
try {
value = decoder.apply(raw);
} catch (RuntimeException e) {
throw new IllegalStateException("cache decode failed, key=" + dataKey, e);
}
slidingTouch(dataKey);
return value;
}
// ---- 删 / 判存 ----
public boolean delete(String key) {
return Boolean.TRUE.equals(redis.delete(keyBuilder.build(key)));
}
public boolean exists(String key) {
// 与 get 口径一致:空值标记不算"存在"(物理 hasKey 会把空标记算存在,与 get→null 打架)
String raw = redis.opsForValue().get(keyBuilder.build(key));
return raw != null && !codec.isNullSentinel(raw);
}
// ---- 续期 ----
public boolean expire(String key, long ttlSeconds) {
requirePositiveTtl(ttlSeconds);
return Boolean.TRUE.equals(
redis.expire(keyBuilder.build(key), applyJitter(ttlSeconds), TimeUnit.SECONDS));
}
// ---- 内部 ----
/** 滑动过期:开启时对已存在的真实值 key 续期(直接对 dataKey 操作,失败忽略)。 */
private void slidingTouch(String dataKey) {
if (!props.isSlidingExpireEnabled()) {
return;
}
try {
redis.expire(dataKey, applyJitter(props.getDefaultTtlSeconds()), TimeUnit.SECONDS);
} catch (RuntimeException e) {
log.warn("[redis-cache] sliding expire failed key={}", dataKey, e);
}
}
/** 在原 TTL 上叠加 [0, ttl*ratio) 的随机抖动(只增不减,半开区间);ratio<=0 或 ttl<=0 时原样返回。 */
long applyJitter(long ttlSeconds) {
double ratio = props.getJitterRatio();
if (ratio <= 0 || ttlSeconds <= 0) {
return ttlSeconds;
}
long bound = (long) (ttlSeconds * ratio);
if (bound <= 0) {
return ttlSeconds;
}
return ttlSeconds + ThreadLocalRandom.current().nextLong(bound);
}
/** TTL 必须为正:0/负会被 Redis 当作"立即删除"(EXPIRE)或非法过期时间(SET),显式 fail-fast。 */
private static void requirePositiveTtl(long ttlSeconds) {
if (ttlSeconds <= 0) {
throw new IllegalArgumentException("ttlSeconds must be positive, but was " + ttlSeconds);
}
}
/** 构造期校验配置项,误配在装配阶段即 fail-fast,而非运行期才暴露。 */
private static void validateProps(RedisCacheProperties props) {
if (props.getDefaultTtlSeconds() <= 0) {
throw new IllegalArgumentException(
"redis-cache.default-ttl-seconds must be positive, but was " + props.getDefaultTtlSeconds());
}
if (props.getNullTtlSeconds() <= 0) {
throw new IllegalArgumentException(
"redis-cache.null-ttl-seconds must be positive, but was " + props.getNullTtlSeconds());
}
double ratio = props.getJitterRatio();
if (ratio < 0 || ratio >= 1) {
throw new IllegalArgumentException(
"redis-cache.jitter-ratio must be in [0,1), but was " + ratio);
}
RedisCacheProperties.Breakdown bd = props.getBreakdown();
if (bd == null) {
throw new IllegalArgumentException("redis-cache.breakdown must not be null");
}
if (bd.getLockTtlMs() <= 0) {
throw new IllegalArgumentException(
"redis-cache.breakdown.lock-ttl-ms must be positive, but was " + bd.getLockTtlMs());
}
if (bd.getWaitTimeoutMs() < 0) {
throw new IllegalArgumentException(
"redis-cache.breakdown.wait-timeout-ms must be >= 0, but was " + bd.getWaitTimeoutMs());
}
if (bd.getWaitIntervalMs() <= 0) {
throw new IllegalArgumentException(
"redis-cache.breakdown.wait-interval-ms must be positive, but was " + bd.getWaitIntervalMs());
}
}
// ---- 读取或回源(三防核心)----
public <T> T getOrLoad(String key, Class<T> type, Supplier<T> loader) {
return getOrLoad(key, type, props.getDefaultTtlSeconds(), loader);
}
public <T> T getOrLoad(String key, Class<T> type, long ttlSeconds, Supplier<T> loader) {
return doGetOrLoad(key, raw -> codec.decode(raw, type), ttlSeconds, loader);
}
public <T> T getOrLoad(String key, TypeReference<T> type, long ttlSeconds, Supplier<T> loader) {
return doGetOrLoad(key, raw -> codec.decode(raw, type), ttlSeconds, loader);
}
private <T> T doGetOrLoad(String key, Function<String, T> decoder, long ttlSeconds, Supplier<T> loader) {
requirePositiveTtl(ttlSeconds);
String dataKey = keyBuilder.build(key);
Lookup<T> r = tryGet(dataKey, decoder);
if (r.hit) {
if (r.realValue) {
slidingTouch(dataKey);
}
return r.value;
}
RedisCacheProperties.Breakdown bd = props.getBreakdown();
if (!bd.isEnabled()) {
return loadAndCache(dataKey, ttlSeconds, loader);
}
String lockKey = keyBuilder.buildLock(key); // 复用 keyBuilder,锁 key 规则单一来源
String token = IdUtil.fastSimpleUUID();
boolean locked = Boolean.TRUE.equals(redis.opsForValue()
.setIfAbsent(lockKey, token, Duration.ofMillis(bd.getLockTtlMs())));
if (locked) {
try {
Lookup<T> r2 = tryGet(dataKey, decoder); // double-check
if (r2.hit) {
if (r2.realValue) {
slidingTouch(dataKey);
}
return r2.value;
}
return loadAndCache(dataKey, ttlSeconds, loader);
} finally {
releaseLock(lockKey, token);
}
}
// 未抢到锁:自旋等待重读,超时或被中断则降级直接回源
long intervalMs = Math.max(1, bd.getWaitIntervalMs()); // 兜底:interval<=0 时不前进会空转
long deadlineNanos = System.nanoTime() + bd.getWaitTimeoutMs() * 1_000_000L;
while (System.nanoTime() - deadlineNanos < 0) { // 溢出安全的截止判断
if (!sleep(intervalMs)) {
break; // 被中断:停止等待、降级回源(中断标志已保留,响应协作式取消)
}
Lookup<T> r3 = tryGet(dataKey, decoder);
if (r3.hit) {
if (r3.realValue) {
slidingTouch(dataKey);
}
return r3.value;
}
}
log.warn("[redis-cache] breakdown wait timeout/interrupted, fallback to loader, key={}", key);
return loader.get();
}
/** 读取三态:未命中 / 命中空值标记 / 命中真实值;反序列化失败当未命中降级。 */
private <T> Lookup<T> tryGet(String dataKey, Function<String, T> decoder) {
String raw = redis.opsForValue().get(dataKey);
if (raw == null) {
return Lookup.miss();
}
if (codec.isNullSentinel(raw)) {
return Lookup.nullHit();
}
try {
return Lookup.hit(decoder.apply(raw));
} catch (RuntimeException e) {
log.warn("[redis-cache] decode failed, treat as miss key={}", dataKey, e);
return Lookup.miss();
}
}
/** 回源并写缓存:null → 空值标记(穿透);非 null → 抖动 TTL(雪崩)。 */
private <T> T loadAndCache(String dataKey, long ttlSeconds, Supplier<T> loader) {
T value = loader.get();
if (value == null) {
redis.opsForValue().set(dataKey, CacheConstant.NULL_SENTINEL,
props.getNullTtlSeconds(), TimeUnit.SECONDS);
} else {
redis.opsForValue().set(dataKey, codec.encode(value),
applyJitter(ttlSeconds), TimeUnit.SECONDS);
}
return value;
}
private void releaseLock(String lockKey, String token) {
try {
redis.execute(UNLOCK_SCRIPT, Collections.singletonList(lockKey), token);
} catch (RuntimeException e) {
log.warn("[redis-cache] release lock failed lockKey={}", lockKey, e);
}
}
/** 休眠;返回 false 表示被中断(已重置中断标志,调用方应停止等待)。 */
private static boolean sleep(long ms) {
try {
Thread.sleep(ms);
return true;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/** 读取三态载体(对外不暴露)。 */
private static final class Lookup<T> {
final boolean hit; // 命中(真实值或空值标记)
final boolean realValue; // 命中且为真实值
final T value;
private Lookup(boolean hit, boolean realValue, T value) {
this.hit = hit;
this.realValue = realValue;
this.value = value;
}
static <T> Lookup<T> miss() {
return new Lookup<>(false, false, null);
}
static <T> Lookup<T> nullHit() {
return new Lookup<>(true, false, null);
}
static <T> Lookup<T> hit(T v) {
return new Lookup<>(true, true, v);
}
}
}

View File

@@ -8,6 +8,7 @@ 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.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
@@ -26,8 +27,9 @@ import java.util.stream.Collectors;
* Phase 2 由 message-boot 注入 Processor 即生效。装配在 {@link RedisAutoConfiguration} 之后,
* 确保 {@link StringRedisTemplate} 已就绪。
*/
@Configuration
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(StringRedisTemplate.class)
@ConditionalOnBean(StringRedisTemplate.class)
@AutoConfigureAfter(RedisAutoConfiguration.class)
@EnableConfigurationProperties(RedisStreamProperties.class)
public class RedisStreamAutoConfiguration {

View File

@@ -17,6 +17,7 @@ import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.nio.charset.StandardCharsets;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
@@ -166,14 +167,17 @@ public class StreamAutoClaimer implements SmartLifecycle {
}
/**
* 处理单条 PEL 消息:取内容 → 死信判定前置 → 认领重投。
* 处理单条 PEL 消息:取内容 → XCLAIM 抢占 → 死信判定 → 认领重投。
*
* <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数;因 {@link #claim} 用<b>不带 JUSTID</b>
* 的 XCLAIM每次认领会把投递计数 +1故 deliveries 随 autoclaim 轮次单调增长
* <p><b>先抢占再处理</b>{@link #claim} 抢不到idle 不足 minIdle或并发实例已认领并重置 idle
* 即直接退出,保证多实例下同一条消息只被一个实例 dispatch / 落死信(幂等),不重复消费、不重复落死信
*
* <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数XPENDING 报告值);因 {@link #claim}
* 用<b>不带 JUSTID</b> 的 XCLAIM每次认领会把投递计数 +1故 deliveries 随 autoclaim 轮次单调增长。
* {@code maxRetry} 语义 = 最多投递 maxRetry 次(含首投),第 maxRetry 次仍 RETRY 即落死信。
* 死信判定<b>前置于 dispatch</b>:达上限直接落异常 + 强制 ACK不再多跑一次业务副作用。
* 死信分支抢占后<b>只落异常 + 强制 ACK不再 dispatch</b>,不跑业务副作用。
*/
private void claimOne(EnhanceStreamConsumerHandler<?> handler, String stream, String group,
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));
@@ -185,6 +189,12 @@ public class StreamAutoClaimer implements SmartLifecycle {
MapRecord<String, Object, Object> rec = records.get(0);
Map<String, String> fields = toStringMap(rec.getValue());
// 先 XCLAIM 抢占仅当真正认领到idle 足够且未被并发实例抢走)才由本实例独占处理;
// 抢不到说明已有其它实例认领,直接退出——杜绝多实例对同一条消息重复 dispatch 或重复落死信。
if (!claim(stream, group, minIdle, id)) {
return;
}
if (deliveries >= maxRetry) {
// 投递计数已达上限 → 落异常 + 强制 ACK死信不再无限滞留 PEL
// 由 handler 内部反序列化后回调,使死信回调能拿到业务消息对象(解析失败兜底 null
@@ -194,8 +204,7 @@ public class StreamAutoClaimer implements SmartLifecycle {
return;
}
// 认领到本 claimer同时把投递计数 +1、重置 idle),再重投
claim(stream, group, minIdle, id);
// 认领投递计数 +1、idle 重置)→ 重投
ConsumeResult result = handler.dispatchMessage(fields);
if (result == ConsumeResult.ACK) {
redis.opsForStream().acknowledge(stream, group, rec.getId());
@@ -209,14 +218,24 @@ public class StreamAutoClaimer implements SmartLifecycle {
* <p>不带 JUSTID 时 XCLAIM 会转移归属、重置 idle 并把投递计数 +1——后者是 maxRetry 死信判定的
* 依据listener 用 {@code XREADGROUP lastConsumed()} 永不重读 PEL若此处用 JUSTID 则投递计数
* 永久冻结在首投值 1{@code deliveries >= maxRetry} 恒不成立、死信分支永不触发。
*
* @return 是否真正认领到本条XCLAIM 抢到返回非空消息列表,没抢到返回空。返回类型未知时保守当作
* 抢到(宁可退化回"可能重复"也不让消息因误判没抢到而永久卡 PEL
*/
private void claim(String stream, String group, long minIdle, String id) {
redis.execute((RedisCallback<Object>) connection -> connection.execute("XCLAIM",
private boolean claim(String stream, String group, long minIdle, String id) {
Object claimed = 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)));
if (claimed instanceof Collection) {
return !((Collection<?>) claimed).isEmpty();
}
if (claimed instanceof Object[]) {
return ((Object[]) claimed).length > 0;
}
return claimed != null;
}
private static Map<String, String> toStringMap(Map<Object, Object> raw) {

View File

@@ -4,10 +4,8 @@ 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;
@@ -20,7 +18,8 @@ import java.util.concurrent.TimeUnit;
* 定时裁剪 stream 长度,防止无界增长。
*
* <p>按 {@code trim.intervalMs} 周期对每个 handler 的 stream 执行
* {@code XTRIM <stream> MAXLEN ~ <maxlen>}{@code ~} 近似裁剪,性能更好)。
* {@code XTRIM <stream> MAXLEN <maxlen>}精确裁剪,经 spring-data 高级 API
* 2.3.x 无近似裁剪重载,故不用 {@code ~})。
*
* <p>无 handler 时不启动。正确性由 C11 集成测试覆盖。
*/
@@ -82,10 +81,6 @@ public class StreamTrimScheduler implements SmartLifecycle {
for (String stream : distinctStreams()) {
try {
redis.opsForStream().trim(stream, maxlen);
// redis.execute((RedisCallback<Object>) connection -> connection.execute("XTRIM",
// stream.getBytes(StandardCharsets.UTF_8),
// "MAXLEN".getBytes(StandardCharsets.UTF_8),
// String.valueOf(maxlen).getBytes(StandardCharsets.UTF_8)));
} catch (Exception e) {
log.warn("[redis-stream] trim failed stream={}", stream, e);
}

View File

@@ -1,2 +1,3 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration
com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration,\
com.njcn.middle.cache.autoconfig.RedisCacheAutoConfiguration

View File

@@ -0,0 +1,65 @@
package com.njcn.middle.cache.autoconfig;
import com.njcn.middle.cache.support.CacheKeyBuilder;
import com.njcn.middle.cache.support.CacheValueCodec;
import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
class RedisCacheAutoConfigurationTest {
private final ApplicationContextRunner runner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
RedisAutoConfiguration.class, RedisCacheAutoConfiguration.class))
.withPropertyValues("spring.redis.host=127.0.0.1", "spring.redis.port=6379");
@Test
void registersCacheBeans() {
runner.run(ctx -> {
assertThat(ctx).hasSingleBean(RedisCacheEnhanceTemplate.class);
assertThat(ctx).hasSingleBean(CacheKeyBuilder.class);
assertThat(ctx).hasSingleBean(CacheValueCodec.class);
});
}
@Test
void backsOffWhenUserDefinesOwnTemplate() {
runner.withUserConfiguration(CustomConfig.class).run(ctx -> {
assertThat(ctx).hasSingleBean(RedisCacheEnhanceTemplate.class);
assertThat(ctx.getBean(RedisCacheEnhanceTemplate.class))
.isSameAs(ctx.getBean("myTemplate"));
});
}
@Test
void backsOffWhenNoStringRedisTemplateBean() {
// 不装配 RedisAutoConfiguration,上下文内无 StringRedisTemplate Bean:
// 应整类 back off 优雅退场,而非因缺依赖启动失败。
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(RedisCacheAutoConfiguration.class))
.run(ctx -> assertThat(ctx)
.hasNotFailed()
.doesNotHaveBean(RedisCacheEnhanceTemplate.class));
}
@Configuration
static class CustomConfig {
@Bean
RedisCacheEnhanceTemplate myTemplate() {
return new RedisCacheEnhanceTemplate(
new StringRedisTemplate(mock(RedisConnectionFactory.class)),
new CacheKeyBuilder("cache:", false, ""),
new CacheValueCodec(),
new RedisCacheProperties());
}
}
}

View File

@@ -0,0 +1,44 @@
package com.njcn.middle.cache.autoconfig;
import org.junit.jupiter.api.Test;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.boot.context.properties.source.ConfigurationPropertySources;
import org.springframework.mock.env.MockEnvironment;
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 RedisCachePropertiesTest {
@Test
void defaults() {
RedisCacheProperties p = new RedisCacheProperties();
assertEquals("cache:", p.getKeyPrefix());
assertFalse(p.isEnvIsolation());
assertEquals("", p.getEnv());
assertEquals(3600, p.getDefaultTtlSeconds());
assertEquals(60, p.getNullTtlSeconds());
assertEquals(0.1, p.getJitterRatio());
assertFalse(p.isSlidingExpireEnabled());
assertTrue(p.getBreakdown().isEnabled());
assertEquals(10000, p.getBreakdown().getLockTtlMs());
assertEquals(3000, p.getBreakdown().getWaitTimeoutMs());
assertEquals(50, p.getBreakdown().getWaitIntervalMs());
}
@Test
void relaxedBinding() {
MockEnvironment env = new MockEnvironment()
.withProperty("redis-cache.default-ttl-seconds", "1800")
.withProperty("redis-cache.sliding-expire-enabled", "true")
.withProperty("redis-cache.breakdown.lock-ttl-ms", "5000")
.withProperty("redis-cache.breakdown.enabled", "false");
RedisCacheProperties p = Binder.get(env)
.bind("redis-cache", RedisCacheProperties.class).get();
assertEquals(1800, p.getDefaultTtlSeconds());
assertTrue(p.isSlidingExpireEnabled());
assertEquals(5000, p.getBreakdown().getLockTtlMs());
assertFalse(p.getBreakdown().isEnabled());
}
}

View File

@@ -0,0 +1,89 @@
package com.njcn.middle.cache.integration;
import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
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 java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* 集成测试:N 个线程并发 getOrLoad 同一未命中热点 key,慢回源放大竞争窗口,
* 互斥锁应保证 loader 仅被调用 1 次,且所有线程拿到同一结果。
*/
@ExtendWith(RedisAvailableCondition.class)
@SpringBootTest(classes = CacheBreakdownConcurrencyTest.TestApp.class,
properties = {"redis-cache.key-prefix=it-cache:"})
class CacheBreakdownConcurrencyTest {
@Autowired
private RedisCacheEnhanceTemplate cache;
@Test
void onlyOneLoaderUnderConcurrency() throws Exception {
String key = "bd:hot:1";
cache.delete(key);
AtomicInteger calls = new AtomicInteger();
java.util.function.Supplier<User> loader = () -> {
calls.incrementAndGet();
try {
Thread.sleep(300); // 慢回源
} catch (InterruptedException ignored) {
Thread.currentThread().interrupt();
}
return new User(1L, "hot");
};
int n = 10;
ExecutorService pool = Executors.newFixedThreadPool(n);
CountDownLatch ready = new CountDownLatch(n);
CountDownLatch go = new CountDownLatch(1);
List<Future<User>> futures = new ArrayList<>();
for (int i = 0; i < n; i++) {
futures.add(pool.submit(() -> {
ready.countDown();
go.await();
return cache.getOrLoad(key, User.class, 60, loader);
}));
}
ready.await();
go.countDown();
for (Future<User> f : futures) {
assertEquals("hot", f.get(10, TimeUnit.SECONDS).getName());
}
assertEquals(1, calls.get(), "并发下 loader 应只回源一次");
pool.shutdownNow();
cache.delete(key);
}
@SpringBootConfiguration
@EnableAutoConfiguration
static class TestApp {
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class User {
private Long id;
private String name;
}
}

View File

@@ -0,0 +1,102 @@
package com.njcn.middle.cache.integration;
import com.njcn.middle.cache.support.CacheKeyBuilder;
import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
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.data.redis.core.StringRedisTemplate;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* 集成测试:显式 expire 续期、滑动过期续真实值、空值标记不被续。
* default-ttl=3600,故滑动续期会把 TTL 拉到远大于初始 60s。
*/
@ExtendWith(RedisAvailableCondition.class)
@SpringBootTest(classes = CacheExpireSlidingTest.TestApp.class,
properties = {
"redis-cache.key-prefix=it-cache:",
"redis-cache.default-ttl-seconds=3600",
"redis-cache.null-ttl-seconds=60",
"redis-cache.jitter-ratio=0",
"redis-cache.sliding-expire-enabled=true"
})
class CacheExpireSlidingTest {
@Autowired
private RedisCacheEnhanceTemplate cache;
@Autowired
private StringRedisTemplate redis;
@Autowired
private CacheKeyBuilder keyBuilder;
private long ttl(String bizKey) {
Long e = redis.getExpire(keyBuilder.build(bizKey), TimeUnit.SECONDS);
return e == null ? -1 : e;
}
@Test
void explicitExpireExtendsTtl() {
String key = "exp:1";
cache.set(key, new User(1L, "a"), 60);
assertTrue(ttl(key) <= 60 && ttl(key) > 0);
assertTrue(cache.expire(key, 600));
assertTrue(ttl(key) > 100, "expire 应把 TTL 续到 600 量级");
cache.delete(key);
}
@Test
void expireOnMissingKeyReturnsFalse() {
String key = "exp:missing";
cache.delete(key);
assertFalse(cache.expire(key, 600));
}
@Test
void slidingRenewsRealValueOnGet() {
String key = "sld:1";
cache.set(key, new User(1L, "a"), 60); // 初始 ~60s
assertTrue(ttl(key) <= 60 && ttl(key) > 0);
cache.get(key, User.class); // 命中真实值 → 滑动续到 default 3600
assertTrue(ttl(key) > 1000, "滑动过期应把 TTL 续到 default 量级");
cache.delete(key);
}
@Test
void slidingDoesNotRenewNullSentinel() {
String key = "sld:missing";
cache.delete(key);
Supplier<User> loader = () -> null;
assertNull(cache.getOrLoad(key, User.class, 60, loader)); // 写空标记 ttl=60
long before = ttl(key);
assertTrue(before <= 60 && before > 0);
assertNull(cache.get(key, User.class)); // 命中空标记,不续
assertTrue(ttl(key) <= 60, "空值标记不应被滑动续期");
cache.delete(key);
}
@SpringBootConfiguration
@EnableAutoConfiguration
static class TestApp {
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class User {
private Long id;
private String name;
}
}

View File

@@ -0,0 +1,59 @@
package com.njcn.middle.cache.integration;
import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
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 java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
/** 集成测试:回源返回 null 时缓存空值标记,二次查询命中空标记、loader 不再被调用。 */
@ExtendWith(RedisAvailableCondition.class)
@SpringBootTest(classes = CachePenetrationTest.TestApp.class,
properties = {"redis-cache.key-prefix=it-cache:", "redis-cache.null-ttl-seconds=30"})
class CachePenetrationTest {
@Autowired
private RedisCacheEnhanceTemplate cache;
@Test
void nullResultCachedAndLoaderCalledOnce() {
String key = "pen:missing:1";
cache.delete(key);
AtomicInteger calls = new AtomicInteger();
Supplier<User> loader = () -> {
calls.incrementAndGet();
return null; // 数据不存在
};
assertNull(cache.getOrLoad(key, User.class, 60, loader));
assertNull(cache.getOrLoad(key, User.class, 60, loader)); // 命中空值标记
assertEquals(1, calls.get());
cache.delete(key);
}
@SpringBootConfiguration
@EnableAutoConfiguration
static class TestApp {
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class User {
private Long id;
private String name;
}
}

View File

@@ -0,0 +1,59 @@
package com.njcn.middle.cache.integration;
import com.njcn.middle.cache.template.RedisCacheEnhanceTemplate;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
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 static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* 集成测试:真实 Redis 下 set/get/exists/delete 闭环。Redis 不可达自动 skip。
*
* <p>连接默认 127.0.0.1:6379;覆盖用 {@code -Dspring.redis.host=… -Dspring.redis.port=… -Dspring.redis.password=…}。
*/
@ExtendWith(RedisAvailableCondition.class)
@SpringBootTest(classes = CacheRoundTripTest.TestApp.class,
properties = {"redis-cache.key-prefix=it-cache:"})
class CacheRoundTripTest {
@Autowired
private RedisCacheEnhanceTemplate cache;
@Test
void setGetExistsDelete() {
String key = "rt:user:1";
cache.delete(key);
cache.set(key, new User(1L, "alice"), 60);
User got = cache.get(key, User.class);
assertEquals("alice", got.getName());
assertTrue(cache.exists(key));
assertTrue(cache.delete(key));
assertNull(cache.get(key, User.class));
assertFalse(cache.exists(key));
}
@SpringBootConfiguration
@EnableAutoConfiguration
static class TestApp {
}
@Data
@AllArgsConstructor
@NoArgsConstructor
static class User {
private Long id;
private String name;
}
}

View File

@@ -0,0 +1,24 @@
package com.njcn.middle.cache.integration;
import org.junit.jupiter.api.extension.ConditionEvaluationResult;
import org.junit.jupiter.api.extension.ExecutionCondition;
import org.junit.jupiter.api.extension.ExtensionContext;
import java.net.InetSocketAddress;
import java.net.Socket;
/** Redis 不可达时禁用整个测试类(skipped,非 failed),保证无 Redis 环境 mvn test 仍全绿。 */
public class RedisAvailableCondition 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");
}
}
}

View File

@@ -0,0 +1,43 @@
package com.njcn.middle.cache.support;
import org.junit.jupiter.api.Test;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
class CacheKeyBuilderTest {
@Test
void buildWithPrefixOnly() {
CacheKeyBuilder kb = new CacheKeyBuilder("cache:", false, "");
assertEquals("cache:user:1", kb.build("user:1"));
}
@Test
void buildWithEnvIsolation() {
CacheKeyBuilder kb = new CacheKeyBuilder("cache:", true, "prod");
assertEquals("cache:prod:user:1", kb.build("user:1"));
}
@Test
void envIsolationButBlankEnvFallsBack() {
CacheKeyBuilder kb = new CacheKeyBuilder("cache:", true, "");
assertEquals("cache:user:1", kb.build("user:1"));
}
@Test
void buildLockDerivesFromDataKey() {
assertEquals("cache:user:1:__lock__",
new CacheKeyBuilder("cache:", false, "").buildLock("user:1"));
assertEquals("cache:prod:user:1:__lock__",
new CacheKeyBuilder("cache:", true, "prod").buildLock("user:1"));
}
@Test
void buildRejectsBlankBizKey() {
CacheKeyBuilder kb = new CacheKeyBuilder("cache:", false, "");
assertThrows(IllegalArgumentException.class, () -> kb.build(null));
assertThrows(IllegalArgumentException.class, () -> kb.build(""));
assertThrows(IllegalArgumentException.class, () -> kb.build(" "));
}
}

View File

@@ -0,0 +1,73 @@
package com.njcn.middle.cache.support;
import com.alibaba.fastjson.TypeReference;
import com.njcn.middle.cache.constant.CacheConstant;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
class CacheValueCodecTest {
private final CacheValueCodec codec = new CacheValueCodec();
@Data
@AllArgsConstructor
@NoArgsConstructor
static class Foo {
private Long id;
private String name;
}
@Test
void encodeNullReturnsSentinel() {
assertEquals(CacheConstant.NULL_SENTINEL, codec.encode(null));
assertTrue(codec.isNullSentinel(codec.encode(null)));
assertFalse(codec.isNullSentinel("{\"id\":1}"));
}
@Test
void encodeDecodePojoRoundTrip() {
Foo foo = new Foo(1L, "alice");
String raw = codec.encode(foo);
Foo back = codec.decode(raw, Foo.class);
assertEquals(foo, back);
}
@Test
void encodeDecodeStringRoundTrip() {
String raw = codec.encode("abc");
assertEquals("abc", codec.decode(raw, String.class));
}
@Test
void decodeGenericWithTypeReference() {
List<Integer> list = Arrays.asList(1, 2, 3);
String raw = codec.encode(list);
List<Integer> back = codec.decode(raw, new TypeReference<List<Integer>>() {
});
assertEquals(list, back);
}
@Test
void decodeDirtyStringThrows() {
assertThrows(RuntimeException.class, () -> codec.decode("{not-json", Foo.class));
}
@Test
void decodeIgnoresAtTypeHint() {
// 含 @type 的脏值(如 Redis 值被外部篡改注入)应被忽略,安全解析为目标类型
String raw = "{\"@type\":\"java.util.HashMap\",\"id\":1,\"name\":\"a\"}";
Foo foo = codec.decode(raw, Foo.class);
assertEquals(1L, foo.getId());
assertEquals("a", foo.getName());
}
}

View File

@@ -0,0 +1,427 @@
package com.njcn.middle.cache.template;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.njcn.middle.cache.autoconfig.RedisCacheProperties;
import com.njcn.middle.cache.constant.CacheConstant;
import com.njcn.middle.cache.support.CacheKeyBuilder;
import com.njcn.middle.cache.support.CacheValueCodec;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.script.RedisScript;
import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyList;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.atMost;
import static org.mockito.Mockito.lenient;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class RedisCacheEnhanceTemplateTest {
@Mock
StringRedisTemplate redis;
@Mock
ValueOperations<String, String> valueOps;
CacheValueCodec codec = new CacheValueCodec();
CacheKeyBuilder keyBuilder = new CacheKeyBuilder("cache:", false, "");
RedisCacheProperties props = new RedisCacheProperties();
RedisCacheEnhanceTemplate template;
@Data
@AllArgsConstructor
@NoArgsConstructor
static class User {
private Long id;
private String name;
}
@BeforeEach
void init() {
MockitoAnnotations.initMocks(this);
lenient().when(redis.opsForValue()).thenReturn(valueOps);
template = new RedisCacheEnhanceTemplate(redis, keyBuilder, codec, props);
}
@Test
void setSerializesWithExactTtlWhenNoJitter() {
props.setJitterRatio(0);
User u = new User(1L, "a");
template.set("user:1", u, 100);
verify(valueOps).set("cache:user:1", JSON.toJSONString(u), 100L, TimeUnit.SECONDS);
}
@Test
void getHitDeserializes() {
when(valueOps.get("cache:user:1")).thenReturn(JSON.toJSONString(new User(1L, "a")));
User got = template.get("user:1", User.class);
assertEquals("a", got.getName());
}
@Test
void getMissReturnsNull() {
when(valueOps.get("cache:user:1")).thenReturn(null);
assertNull(template.get("user:1", User.class));
}
@Test
void getNullSentinelReturnsNull() {
when(valueOps.get("cache:user:1")).thenReturn(CacheConstant.NULL_SENTINEL);
assertNull(template.get("user:1", User.class));
}
@Test
void getDirtyDataThrowsIllegalState() {
when(valueOps.get("cache:user:1")).thenReturn("{not-json");
assertThrows(IllegalStateException.class, () -> template.get("user:1", User.class));
}
@Test
void expireDelegatesWithJitterTtl() {
props.setJitterRatio(0);
when(redis.expire("cache:user:1", 200L, TimeUnit.SECONDS)).thenReturn(true);
assertTrue(template.expire("user:1", 200));
verify(redis).expire("cache:user:1", 200L, TimeUnit.SECONDS);
}
@Test
void applyJitterWithinBounds() {
props.setJitterRatio(0.1);
long bound = (long) (100 * 0.1);
for (int i = 0; i < 500; i++) {
long ttl = template.applyJitter(100);
assertTrue(ttl >= 100 && ttl < 100 + bound, "ttl=" + ttl); // 半开区间 [100,110)
}
props.setJitterRatio(0);
assertEquals(100, template.applyJitter(100));
}
@Test
void slidingOffDoesNotRenewOnGet() {
when(valueOps.get("cache:user:1")).thenReturn(JSON.toJSONString(new User(1L, "a")));
template.get("user:1", User.class);
verify(redis, never()).expire(any(), anyLong(), any());
}
@Test
void slidingOnRenewsRealValueOnGet() {
props.setSlidingExpireEnabled(true);
props.setJitterRatio(0);
when(valueOps.get("cache:user:1")).thenReturn(JSON.toJSONString(new User(1L, "a")));
when(redis.expire(eq("cache:user:1"), anyLong(), eq(TimeUnit.SECONDS))).thenReturn(true);
template.get("user:1", User.class);
verify(redis).expire("cache:user:1", props.getDefaultTtlSeconds(), TimeUnit.SECONDS);
}
@Test
void slidingOnDoesNotRenewNullSentinel() {
props.setSlidingExpireEnabled(true);
when(valueOps.get("cache:user:1")).thenReturn(CacheConstant.NULL_SENTINEL);
assertNull(template.get("user:1", User.class));
verify(redis, never()).expire(any(), anyLong(), any());
}
@Test
void getOrLoadHitDoesNotCallLoader() {
when(valueOps.get("cache:user:1")).thenReturn(JSON.toJSONString(new User(1L, "a")));
AtomicInteger calls = new AtomicInteger();
Supplier<User> loader = () -> {
calls.incrementAndGet();
return new User(9L, "x");
};
User got = template.getOrLoad("user:1", User.class, 60, loader);
assertEquals("a", got.getName());
assertEquals(0, calls.get());
}
@Test
void getOrLoadNullSentinelHitReturnsNullNoLoader() {
when(valueOps.get("cache:user:1")).thenReturn(CacheConstant.NULL_SENTINEL);
AtomicInteger calls = new AtomicInteger();
assertNull(template.getOrLoad("user:1", User.class, 60, () -> {
calls.incrementAndGet();
return new User(1L, "a");
}));
assertEquals(0, calls.get());
}
@Test
void getOrLoadMissAcquiresLockLoadsAndCaches() {
props.setJitterRatio(0);
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(eq("cache:user:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
User loaded = new User(1L, "a");
User got = template.getOrLoad("user:1", User.class, 60, () -> loaded);
assertEquals("a", got.getName());
verify(valueOps).set("cache:user:1", JSON.toJSONString(loaded), 60L, TimeUnit.SECONDS);
// 释放锁:执行了 Lua 脚本
verify(redis).execute(any(RedisScript.class), anyList(), any());
}
@Test
void getOrLoadLoaderReturnsNullWritesSentinel() {
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(eq("cache:user:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
assertNull(template.getOrLoad("user:1", User.class, 60, () -> null));
verify(valueOps).set("cache:user:1", CacheConstant.NULL_SENTINEL,
props.getNullTtlSeconds(), TimeUnit.SECONDS);
}
@Test
void getOrLoadNotLockedWaitsThenReadsValue() {
props.getBreakdown().setWaitIntervalMs(1);
props.getBreakdown().setWaitTimeoutMs(1000);
// 初始 tryGet 返回 null,未抢到锁,随后等待中读到值
when(valueOps.get("cache:user:1"))
.thenReturn(null)
.thenReturn(null)
.thenReturn(JSON.toJSONString(new User(1L, "a")));
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
AtomicInteger calls = new AtomicInteger();
User got = template.getOrLoad("user:1", User.class, 60, () -> {
calls.incrementAndGet();
return new User(9L, "x");
});
assertEquals("a", got.getName());
assertEquals(0, calls.get());
}
@Test
void getOrLoadWaitTimeoutFallsBackToLoader() {
props.getBreakdown().setWaitIntervalMs(1);
props.getBreakdown().setWaitTimeoutMs(5);
when(valueOps.get("cache:user:1")).thenReturn(null); // 始终未命中
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
AtomicInteger calls = new AtomicInteger();
User got = template.getOrLoad("user:1", User.class, 60, () -> {
calls.incrementAndGet();
return new User(7L, "fallback");
});
assertEquals("fallback", got.getName());
assertEquals(1, calls.get());
}
@Test
void getOrLoadBreakdownDisabledLoadsWithoutLock() {
props.getBreakdown().setEnabled(false);
when(valueOps.get("cache:user:1")).thenReturn(null);
User got = template.getOrLoad("user:1", User.class, 60, () -> new User(1L, "a"));
assertEquals("a", got.getName());
verify(valueOps, never()).setIfAbsent(anyString(), anyString(), any(Duration.class));
}
@Test
void getOrLoadInternalDecodeFailureFallsBackToReload() {
// 初始读到脏数据(当未命中降级),抢到锁后回源重建
when(valueOps.get("cache:user:1")).thenReturn("{dirty").thenReturn("{dirty");
when(valueOps.setIfAbsent(eq("cache:user:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
User got = template.getOrLoad("user:1", User.class, 60, () -> new User(1L, "fresh"));
assertEquals("fresh", got.getName());
}
// ---- #1+#2: ttl<=0 入口校验(fail-fast,不下发非法过期) ----
@Test
void setRejectsNonPositiveTtl() {
User u = new User(1L, "a");
assertThrows(IllegalArgumentException.class, () -> template.set("user:1", u, 0));
assertThrows(IllegalArgumentException.class, () -> template.set("user:1", u, -1));
verify(valueOps, never()).set(anyString(), anyString(), anyLong(), any());
}
@Test
void expireRejectsNonPositiveTtl() {
assertThrows(IllegalArgumentException.class, () -> template.expire("user:1", 0));
assertThrows(IllegalArgumentException.class, () -> template.expire("user:1", -5));
verify(redis, never()).expire(anyString(), anyLong(), any());
}
@Test
void getOrLoadRejectsNonPositiveTtl() {
props.getBreakdown().setEnabled(false); // 避免跑红时走入等待自旋
assertThrows(IllegalArgumentException.class,
() -> template.getOrLoad("user:1", User.class, 0, () -> new User(1L, "a")));
}
// ---- #3+#4: 击穿等待的中断响应与 interval 兜底 ----
@Test
void getOrLoadWaitIntervalZeroDoesNotSpinForever() {
props.getBreakdown().setWaitIntervalMs(0); // 误配 0:旧实现 waited 永不前进 → 无限循环
props.getBreakdown().setWaitTimeoutMs(50);
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
User got = assertTimeoutPreemptively(Duration.ofSeconds(2), () ->
template.getOrLoad("user:1", User.class, 60, () -> new User(7L, "fb")));
assertEquals("fb", got.getName());
}
@Test
void getOrLoadInterruptedDuringWaitStopsSpinning() {
props.getBreakdown().setWaitIntervalMs(20);
props.getBreakdown().setWaitTimeoutMs(10000); // 大超时:旧实现中断后空转数百次到超时
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
Thread.currentThread().interrupt(); // 当前线程预置中断,getOrLoad 同步执行可感知
User got = template.getOrLoad("user:1", User.class, 60, () -> new User(7L, "fb"));
boolean wasInterrupted = Thread.interrupted(); // 读取并清除,避免污染后续测试
assertEquals("fb", got.getName());
assertTrue(wasInterrupted, "中断标志应被保留(协作式取消)");
// 中断后立即停止等待:仅首次 tryGet 读一次,而非空转数百次
verify(valueOps, atMost(2)).get("cache:user:1");
}
// ---- #5: 构造期配置 fail-fast 校验 ----
private RedisCacheEnhanceTemplate newTemplateWith(Consumer<RedisCacheProperties> mut) {
RedisCacheProperties p = new RedisCacheProperties();
mut.accept(p);
return new RedisCacheEnhanceTemplate(redis, keyBuilder, codec, p);
}
@Test
void constructorRejectsInvalidProps() {
assertThrows(IllegalArgumentException.class, () -> newTemplateWith(p -> p.setDefaultTtlSeconds(0)));
assertThrows(IllegalArgumentException.class, () -> newTemplateWith(p -> p.setNullTtlSeconds(-1)));
assertThrows(IllegalArgumentException.class, () -> newTemplateWith(p -> p.setJitterRatio(1.0)));
assertThrows(IllegalArgumentException.class, () -> newTemplateWith(p -> p.setJitterRatio(-0.1)));
assertThrows(IllegalArgumentException.class, () -> newTemplateWith(p -> p.getBreakdown().setLockTtlMs(0)));
assertThrows(IllegalArgumentException.class, () -> newTemplateWith(p -> p.getBreakdown().setWaitIntervalMs(0)));
}
@Test
void constructorAcceptsDefaults() {
new RedisCacheEnhanceTemplate(redis, keyBuilder, codec, new RedisCacheProperties());
}
// ---- #7: exists 排除空值标记(与 get 口径一致) ----
@Test
void existsTrueForRealValue() {
when(valueOps.get("cache:user:1")).thenReturn(JSON.toJSONString(new User(1L, "a")));
assertTrue(template.exists("user:1"));
}
@Test
void existsFalseForNullSentinel() {
when(valueOps.get("cache:user:1")).thenReturn(CacheConstant.NULL_SENTINEL);
assertFalse(template.exists("user:1"));
}
@Test
void existsFalseForMissing() {
when(valueOps.get("cache:user:1")).thenReturn(null);
assertFalse(template.exists("user:1"));
}
// ---- #9: 公共 set(key,null) 用短 null-ttl(与回源空值一致) ----
@Test
void setNullUsesNullTtlNotDataTtl() {
props.setJitterRatio(0);
template.set("user:1", null, 3600); // 传入数据 TTL,但写空标记应改用短 null-ttl
verify(valueOps).set("cache:user:1", CacheConstant.NULL_SENTINEL,
props.getNullTtlSeconds(), TimeUnit.SECONDS);
}
// ---- #11: 补关键路径覆盖(loader 异常 / double-check / TypeReference / sliding 回源 / 降级不写缓存) ----
@Test
void getOrLoadLoaderThrowsReleasesLockAndPropagates() {
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(eq("cache:user:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
RuntimeException boom = new RuntimeException("boom");
RuntimeException thrown = assertThrows(RuntimeException.class,
() -> template.getOrLoad("user:1", User.class, 60, () -> {
throw boom;
}));
assertSame(boom, thrown); // 异常原样透传
verify(redis).execute(any(RedisScript.class), anyList(), any()); // 锁在 finally 释放
verify(valueOps, never()).set(anyString(), anyString(), anyLong(), any()); // 未写缓存
}
@Test
void getOrLoadDoubleCheckHitSkipsLoader() {
props.setJitterRatio(0);
when(valueOps.get("cache:user:1"))
.thenReturn(null) // 首次 tryGet miss
.thenReturn(JSON.toJSONString(new User(1L, "a"))); // 抢锁后 double-check 命中
when(valueOps.setIfAbsent(eq("cache:user:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
AtomicInteger calls = new AtomicInteger();
User got = template.getOrLoad("user:1", User.class, 60, () -> {
calls.incrementAndGet();
return new User(9L, "x");
});
assertEquals("a", got.getName());
assertEquals(0, calls.get()); // double-check 命中,未回源
verify(redis).execute(any(RedisScript.class), anyList(), any()); // 锁仍释放
}
@Test
void getOrLoadTypeReferenceRoundTrip() {
props.setJitterRatio(0);
when(valueOps.get("cache:list:1")).thenReturn(null);
when(valueOps.setIfAbsent(eq("cache:list:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
List<Integer> loaded = Arrays.asList(1, 2, 3);
List<Integer> got = template.getOrLoad("list:1",
new TypeReference<List<Integer>>() {
}, 60, () -> loaded);
assertEquals(loaded, got);
verify(valueOps).set("cache:list:1", JSON.toJSONString(loaded), 60L, TimeUnit.SECONDS);
}
@Test
void getOrLoadSlidingDoesNotRenewOnFreshLoad() {
props.setSlidingExpireEnabled(true);
props.setJitterRatio(0);
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(eq("cache:user:1:__lock__"), anyString(), any(Duration.class)))
.thenReturn(true);
template.getOrLoad("user:1", User.class, 60, () -> new User(1L, "a"));
// 回源新写入已自带 TTL,不应再触发滑动续期 expire
verify(redis, never()).expire(eq("cache:user:1"), anyLong(), any());
}
@Test
void getOrLoadWaitTimeoutFallbackDoesNotWriteCache() {
props.getBreakdown().setWaitIntervalMs(1);
props.getBreakdown().setWaitTimeoutMs(5);
when(valueOps.get("cache:user:1")).thenReturn(null);
when(valueOps.setIfAbsent(anyString(), anyString(), any(Duration.class))).thenReturn(false);
template.getOrLoad("user:1", User.class, 60, () -> new User(7L, "fb"));
verify(valueOps, never()).set(anyString(), anyString(), anyLong(), any()); // 降级不写缓存
}
}

View File

@@ -0,0 +1,71 @@
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.support.StreamKeyBuilder;
import com.njcn.middle.stream.support.StreamTrimScheduler;
import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate;
import org.junit.jupiter.api.Test;
import org.springframework.boot.autoconfigure.AutoConfigurations;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
/**
* 自动装配测试:验证 stream starter 的 Bean 装配、用户自定义回退、以及缺 {@link StringRedisTemplate}
* 时整类优雅退场(与 cache 装配口径对齐,不因缺依赖启动失败)。
*/
class RedisStreamAutoConfigurationTest {
private final ApplicationContextRunner runner = new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(
RedisAutoConfiguration.class, RedisStreamAutoConfiguration.class))
.withPropertyValues("spring.redis.host=127.0.0.1", "spring.redis.port=6379");
@Test
void registersStreamBeans() {
runner.run(ctx -> {
assertThat(ctx).hasSingleBean(StreamKeyBuilder.class);
assertThat(ctx).hasSingleBean(RedisStreamEnhanceTemplate.class);
assertThat(ctx).hasSingleBean(StreamListenerContainer.class);
assertThat(ctx).hasSingleBean(StreamAutoClaimer.class);
assertThat(ctx).hasSingleBean(StreamTrimScheduler.class);
});
}
@Test
void backsOffWhenUserDefinesOwnTemplate() {
runner.withUserConfiguration(CustomConfig.class).run(ctx -> {
assertThat(ctx).hasSingleBean(RedisStreamEnhanceTemplate.class);
assertThat(ctx.getBean(RedisStreamEnhanceTemplate.class))
.isSameAs(ctx.getBean("myTemplate"));
});
}
@Test
void backsOffWhenNoStringRedisTemplateBean() {
// 不装配 RedisAutoConfiguration,上下文内无 StringRedisTemplate Bean:
// 应整类 back off 优雅退场,而非因缺依赖启动失败(与 cache 装配口径一致)。
new ApplicationContextRunner()
.withConfiguration(AutoConfigurations.of(RedisStreamAutoConfiguration.class))
.run(ctx -> assertThat(ctx)
.hasNotFailed()
.doesNotHaveBean(RedisStreamEnhanceTemplate.class));
}
@Configuration
static class CustomConfig {
@Bean
RedisStreamEnhanceTemplate myTemplate() {
return new RedisStreamEnhanceTemplate(
new StringRedisTemplate(mock(RedisConnectionFactory.class)),
new StreamKeyBuilder(false, ""));
}
}
}

View File

@@ -0,0 +1,231 @@
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 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.data.domain.Range;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.PendingMessages;
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.net.InetSocketAddress;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
* 集成测试Gateautoclaim 多实例并发认领同一条 idle 超时的 PEL 消息时,业务必须只被处理一次。
*
* <p>端到端验证 B1 幂等:两个 consumerName 不同的 {@link StreamAutoClaimer} 用 {@link CyclicBarrier}
* 同时对同一条消息发 {@code XCLAIM}(不带 JUSTID。Redis 单线程串行处理命令——先到者认领成功、把
* idle 重置为 0后到者看到 {@code idle≈0 < minIdle} 必然抢不到({@link StreamAutoClaimer#claim}
* 返回 false→ 直接退出,不 dispatch。故无论哪个线程先到{@code handleMessage} 恒只触发一次。
*
* <p>这条用例覆盖纯 mock 单测({@code StreamAutoClaimerClaimTest})和单实例死信集成测试都证明不了的
* <b>真实并发竞态</b>mock 短路了 XCLAIM 的真实归属转移 + idle 重置语义,唯有真 Redis 能裁判。
*
* <p>测试放在 {@code container} 包以调用 package-private 的 {@link StreamAutoClaimer#claimOne}
* 直接复现 autoclaim 单条认领,不依赖定时器时序,结果确定(非 flaky
*
* <p>Redis 不可达时整类禁用skipped。连接默认 {@code 127.0.0.1:6379},用
* {@code -Dspring.redis.host=… -Dspring.redis.port=…} 覆盖。本地起:{@code docker run -d -p 6379:6379 redis:6.2}。
*/
@ExtendWith(StreamAutoClaimConcurrentClaimTest.RedisReachableCondition.class)
@SpringBootTest(
classes = StreamAutoClaimConcurrentClaimTest.TestApp.class,
properties = {
"redis-stream.env-isolation=false",
"redis-stream.read-count=10"
})
class StreamAutoClaimConcurrentClaimTest {
private static final String TOPIC = "IT_Concurrent_Topic";
private static final String GROUP = "IT_Concurrent_Group";
private static final long MIN_IDLE_MS = 500L;
@Autowired
private RedisStreamEnhanceTemplate template;
@Autowired
private StringRedisTemplate redis;
@Autowired
private StreamKeyBuilder keyBuilder;
@Test
void onlyOneInstanceProcessesUnderConcurrentClaim() throws Exception {
CountingHandler handler = new CountingHandler();
String stream = keyBuilder.buildStream(TOPIC);
// 清历史 + 重建组delete 会连带删组XADD 不重建,故须手动 MKSTREAM 建组)。
redis.delete(stream);
createGroup(stream, GROUP);
// 发一条真实消息(含 enc/body可被 dispatchMessage 正常 decode→反序列化
ConcurrentMsg msg = new ConcurrentMsg();
msg.setText("process-once");
template.send(TOPIC, msg, false);
// 用 origin consumer XREADGROUP 读进 PEL不 ACK投递计数=1之后 idle 从 0 起涨。
List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
Consumer.from(GROUP, "origin-consumer"),
StreamReadOptions.empty().count(10),
StreamOffset.create(stream, ReadOffset.lastConsumed()));
assertEquals(1, recs == null ? 0 : recs.size(), "应有 1 条进入 PEL");
String id = recs.get(0).getId().getValue();
// 等 idle 超过 minIdle使首个认领者满足 XCLAIM 的 min-idle 门槛。
Thread.sleep(MIN_IDLE_MS + 300L);
StreamAutoClaimer claimerA = startedClaimer("concurrent-claimer-A", handler);
StreamAutoClaimer claimerB = startedClaimer("concurrent-claimer-B", handler);
try {
CyclicBarrier gate = new CyclicBarrier(2);
ExecutorService pool = Executors.newFixedThreadPool(2);
List<Future<Void>> futures = pool.invokeAll(Arrays.asList(
claimTask(gate, claimerA, handler, stream, id),
claimTask(gate, claimerB, handler, stream, id)));
for (Future<Void> f : futures) {
f.get(10, TimeUnit.SECONDS); // 解包任务内异常,使其转为测试失败
}
pool.shutdownNow();
} finally {
claimerA.stop();
claimerB.stop();
}
assertEquals(1, handler.handled.get(),
"两个实例并发认领同一条,业务只应被处理一次(多实例幂等)");
PendingMessages pel = redis.opsForStream().pending(stream, GROUP, Range.unbounded(), 100);
assertEquals(0L, pel == null ? 0 : pel.size(), "认领方 ACK 后 PEL 应清零");
}
/** 两个 claimer 用 deliveries=1< maxRetry走正常 dispatch 分支barrier 让二者尽量同时发 XCLAIM。 */
private Callable<Void> claimTask(CyclicBarrier gate, StreamAutoClaimer claimer,
CountingHandler handler, String stream, String id) {
return () -> {
gate.await();
claimer.claimOne(handler, stream, GROUP, MIN_IDLE_MS, 100, id, 1);
return null;
};
}
/** new + start()start() 初始化 claimConsumerinterval 设大,定时器在测试窗内不触发,避免干扰。 */
private StreamAutoClaimer startedClaimer(String consumerName, CountingHandler handler) {
RedisStreamProperties props = new RedisStreamProperties();
props.setConsumerName(consumerName);
props.getAutoclaim().setIntervalMs(600000);
StreamAutoClaimer claimer = new StreamAutoClaimer(
redis, keyBuilder, props, Collections.singletonList(handler));
claimer.start();
return claimer;
}
private void createGroup(String stream, String group) {
redis.execute((RedisCallback<Object>) conn -> {
try {
return conn.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));
} catch (RuntimeException e) {
return null; // BUSYGROUP组已存在忽略
}
});
}
/** 不注册任何 handler beanautoconfig 的 listener/autoclaimer 因 handlers 为空而空转,不干扰本测试。 */
@SpringBootConfiguration
@EnableAutoConfiguration
static class TestApp {
}
@Data
@EqualsAndHashCode(callSuper = true)
static class ConcurrentMsg extends StreamBaseMessage {
private String text;
}
/** 真实 dispatch 路径计数decode→反序列化→handleMessage统计业务被处理的次数。 */
static class CountingHandler extends EnhanceStreamConsumerHandler<ConcurrentMsg> {
final AtomicInteger handled = new AtomicInteger();
@Override
public String topic() {
return TOPIC;
}
@Override
public String group() {
return GROUP;
}
@Override
public Class<ConcurrentMsg> messageType() {
return ConcurrentMsg.class;
}
@Override
public void handleMessage(ConcurrentMsg message) {
handled.incrementAndGet();
}
@Override
public boolean isRetry() {
return true;
}
@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");
}
}
}
}

View File

@@ -0,0 +1,141 @@
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.ConsumeResult;
import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler;
import com.njcn.middle.stream.support.StreamKeyBuilder;
import org.junit.jupiter.api.Test;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* 单元测试:autoclaim 认领单条 PEL 消息({@link StreamAutoClaimer#claimOne})的多实例幂等。
*
* <p>核心:XCLAIM 真正抢到(返回非空)才由本实例 dispatch / 落死信;抢不到(并发实例已认领、
* idle 被重置)直接退出,杜绝重复消费与重复死信回调。
*/
class StreamAutoClaimerClaimTest {
private static final String STREAM = "T";
private static final String GROUP = "G";
private static final String ID = "1-0";
private final StringRedisTemplate redis = mock(StringRedisTemplate.class);
@SuppressWarnings("unchecked")
private final StreamOperations<String, Object, Object> ops = mock(StreamOperations.class);
@SuppressWarnings("unchecked")
private StreamAutoClaimer newClaimer(RecordingHandler handler) {
when(redis.opsForStream()).thenReturn(ops);
Map<Object, Object> fields = new HashMap<>();
fields.put("enc", "plain");
fields.put("body", "{}");
MapRecord<String, Object, Object> rec = mock(MapRecord.class);
when(rec.getValue()).thenReturn(fields);
when(rec.getId()).thenReturn(RecordId.of(ID));
when(ops.range(eq(STREAM), any(Range.class))).thenReturn(Collections.singletonList(rec));
return new StreamAutoClaimer(redis, new StreamKeyBuilder(false, ""),
new RedisStreamProperties(), Collections.singletonList(handler));
}
@SuppressWarnings("unchecked")
private void stubClaimAcquired(boolean acquired) {
when(redis.execute(any(RedisCallback.class)))
.thenReturn(acquired ? Collections.singletonList(new Object()) : Collections.emptyList());
}
@Test
void skipsDispatchWhenClaimLost() {
RecordingHandler handler = new RecordingHandler();
StreamAutoClaimer claimer = newClaimer(handler);
stubClaimAcquired(false); // 并发实例已抢走,本实例 XCLAIM 抢不到
claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 0);
assertFalse(handler.dispatched, "未抢到 XCLAIM 不应 dispatch");
verify(ops, never()).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
}
@Test
void skipsDeadLetterWhenClaimLost() {
RecordingHandler handler = new RecordingHandler();
StreamAutoClaimer claimer = newClaimer(handler);
stubClaimAcquired(false);
claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 9); // deliveries 已超限
assertFalse(handler.deadLettered, "未抢到 XCLAIM 不应重复落死信");
verify(ops, never()).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
}
@Test
void dispatchesWhenClaimAcquired() {
RecordingHandler handler = new RecordingHandler();
StreamAutoClaimer claimer = newClaimer(handler);
stubClaimAcquired(true);
claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 0);
assertTrue(handler.dispatched, "抢到 XCLAIM 应 dispatch");
verify(ops, times(1)).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
}
@Test
void deadLettersWhenClaimAcquiredAndOverLimit() {
RecordingHandler handler = new RecordingHandler();
StreamAutoClaimer claimer = newClaimer(handler);
stubClaimAcquired(true);
claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 9);
assertTrue(handler.deadLettered, "抢到且超限应落死信");
assertFalse(handler.dispatched, "死信不应再 dispatch 业务");
verify(ops, times(1)).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class));
}
static class TestMsg extends StreamBaseMessage {
}
/** 记录 dispatch / 死信回调是否被触发的测试桩。 */
static class RecordingHandler extends EnhanceStreamConsumerHandler<TestMsg> {
boolean dispatched = false;
boolean deadLettered = false;
@Override public String topic() { return STREAM; }
@Override public String group() { return GROUP; }
@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 ConsumeResult dispatchMessage(Map<String, String> fields) {
dispatched = true;
return ConsumeResult.ACK;
}
@Override
public void saveExceptionFromFields(Map<String, String> fields, Exception cause) {
deadLettered = true;
}
}
}