From 407df3045746450bad7780b875a86b9f39a41ae8 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Mon, 29 Jun 2026 19:59:07 +0800 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-redis-stream):=20=E5=8E=BB?= =?UTF-8?q?=E9=87=8D=20Redis=20=E9=BB=98=E8=AE=A4=E5=AE=9E=E7=8E=B0=20Redi?= =?UTF-8?q?sMqIdempotentStore?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 实现 MqIdempotentStore：setIfAbsent 原子占用 + processing/success/fail 状态机 + TTL - 修正 legacy 读写 key 不一致缺陷（读写统一前缀键） - autoconfig 在 redis-stream 模式装配为默认 store（@ConditionalOnMissingBean） - 集成测试连真实 redis 验证状态机；@Bean 声明契约护栏 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../driver/redis/RedisMqIdempotentStore.java | 60 +++++++++++++++++++ .../RedisStreamMqDriverAutoConfiguration.java | 8 +++ .../redis/RedisMqIdempotentStoreIT.java | 38 ++++++++++++ .../redis/RedisStreamMqDriverWiringTest.java | 17 ++++++ 4 files changed, 123 insertions(+) create mode 100644 mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java create mode 100644 mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java new file mode 100644 index 0000000..3abd48d --- /dev/null +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java @@ -0,0 +1,60 @@ +package com.njcn.mq.driver.redis; + +import com.njcn.mq.spi.MqIdempotentStore; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.time.Duration; + +/** + * 基于 Redis 的去重默认实现。状态机 processing / success / fail，各带 TTL。 + * + *

对齐 legacy filter+consumeSuccess+saveException 的语义，但修正了 legacy「读裸 key、写带前缀 key」 + * 导致去重失效的缺陷：本实现读写统一使用 {@code prefix + key}。 + */ +public class RedisMqIdempotentStore implements MqIdempotentStore { + + private static final String PROCESSING = "processing"; + private static final String SUCCESS = "success"; + private static final String FAIL = "fail"; + + private final StringRedisTemplate redis; + private final String prefix; + private final long processingTtl; + private final long successTtl; + private final long failTtl; + + public RedisMqIdempotentStore(StringRedisTemplate redis, String prefix, + long processingTtlSeconds, long successTtlSeconds, long failTtlSeconds) { + this.redis = redis; + this.prefix = prefix; + this.processingTtl = processingTtlSeconds; + this.successTtl = successTtlSeconds; + this.failTtl = failTtlSeconds; + } + + @Override + public boolean tryAcquire(String key) { + String k = prefix + key; + // setIfAbsent 原子占用：key 不存在时写 processing 并放行。 + Boolean acquired = redis.opsForValue().setIfAbsent(k, PROCESSING, Duration.ofSeconds(processingTtl)); + if (Boolean.TRUE.equals(acquired)) { + return true; + } + // key 已存在：fail（上次失败）允许重新消费；processing / success 一律挡回。 + if (FAIL.equals(redis.opsForValue().get(k))) { + redis.opsForValue().set(k, PROCESSING, Duration.ofSeconds(processingTtl)); + return true; + } + return false; + } + + @Override + public void markSuccess(String key) { + redis.opsForValue().set(prefix + key, SUCCESS, Duration.ofSeconds(successTtl)); + } + + @Override + public void markFail(String key) { + redis.opsForValue().set(prefix + key, FAIL, Duration.ofSeconds(failTtl)); + } +} diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java index e9736a5..4dac641 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java @@ -3,6 +3,7 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties; import com.njcn.middle.stream.support.StreamKeyBuilder; import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqIdempotentStore; import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; @@ -21,4 +22,11 @@ public class RedisStreamMqDriverAutoConfiguration { public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) { return new RedisStreamMqDriver(redis, streamKeyBuilder, props); } + + @Bean + @ConditionalOnMissingBean + public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis) { + // 默认 TTL：processing 60s / success 300s / fail 300s，对齐 legacy 量级；统一前缀（修正 key 一致性）。 + return new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300); + } } diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java new file mode 100644 index 0000000..5dd81a2 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java @@ -0,0 +1,38 @@ +package com.njcn.mq.driver.redis; + +import org.junit.jupiter.api.Test; +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory; +import org.springframework.data.redis.core.StringRedisTemplate; + +import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +class RedisMqIdempotentStoreIT { + + @Test + void stateMachine_firstAcquire_thenBlocked_failReopens_successBlocks() { + StringRedisTemplate redis = newRedisOrSkip(); + RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:it:", 60, 300, 300); + String key = "k_" + System.nanoTime(); + + assertTrue(store.tryAcquire(key), "首次应放行并占用 processing"); + assertFalse(store.tryAcquire(key), "processing 中应挡回"); + + store.markFail(key); + assertTrue(store.tryAcquire(key), "上次 fail 后应允许重新消费"); + + store.markSuccess(key); + assertFalse(store.tryAcquire(key), "success 后应挡回，避免重复消费"); + } + + private static StringRedisTemplate newRedisOrSkip() { + try { + LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379); + cf.afterPropertiesSet(); + StringRedisTemplate t = new StringRedisTemplate(cf); + t.afterPropertiesSet(); + t.hasKey("ping"); + return t; + } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; } + } +} diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java index 3e2663a..953e0de 100644 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java @@ -1,8 +1,13 @@ package com.njcn.mq.driver.redis; import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; +import com.njcn.mq.spi.MqIdempotentStore; import org.junit.jupiter.api.Test; import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.context.annotation.Bean; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.lang.reflect.Method; import static org.assertj.core.api.Assertions.assertThat; @@ -30,4 +35,16 @@ class RedisStreamMqDriverWiringTest { .as("@AutoConfigureBefore 必须指向 MqCoreAutoConfiguration") .contains(MqCoreAutoConfiguration.class); } + + @Test + void declaresIdempotentStoreBean() throws Exception { + Method m = RedisStreamMqDriverAutoConfiguration.class + .getDeclaredMethod("mqIdempotentStore", StringRedisTemplate.class); + assertThat(m.getAnnotation(Bean.class)) + .as("redis-stream 模式应提供默认去重实现") + .isNotNull(); + assertThat(MqIdempotentStore.class) + .as("默认去重实现须为 MqIdempotentStore 类型") + .isAssignableFrom(m.getReturnType()); + } }