对齐 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 一律挡回。 + // 注意：fail→reopen 的 get+set 非原子，跨实例对「共享同一 key 的不同消息」可能都放行； + // 同一条消息由 stream PEL/claim 序列化，故仅影响不同消息共享 key 的边缘场景。 + 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)); + } + + @Override + public boolean isCompleted(String key) { + // 仅 success 视为"已完成"(driver 可 ACK)。processing(并发处理中/崩溃残留)与 + // 无记录(标记已过期)一律 false，使 driver 保留消息、待重投，避免静默丢消息。 + return SUCCESS.equals(redis.opsForValue().get(prefix + key)); + } +} diff --git a/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreIT.java b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreIT.java new file mode 100644 index 0000000..078527c --- /dev/null +++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreIT.java @@ -0,0 +1,38 @@ +package com.njcn.mq.idempotent.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-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreTest.java b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreTest.java new file mode 100644 index 0000000..1b049a9 --- /dev/null +++ b/mq-starter-idempotent-redis/src/test/java/com/njcn/mq/idempotent/redis/RedisMqIdempotentStoreTest.java @@ -0,0 +1,34 @@ +package com.njcn.mq.idempotent.redis; + +import org.junit.jupiter.api.Test; +import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.data.redis.core.ValueOperations; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * {@link RedisMqIdempotentStore#isCompleted(String)} 纯单测(mock StringRedisTemplate，不依赖真实 Redis）。 + * + *
isCompleted 必须仅在 key 值为 {@code success} 时返回 true：processing / 无记录 一律 false，
+ * 否则消费者崩溃后残留的 processing 标记会被 driver 当成"已完成"而 ACK，造成静默丢消息。
+ */
+class RedisMqIdempotentStoreTest {
+
+ @Test
+ void isCompleted_trueOnlyForSuccessValue() {
+ StringRedisTemplate redis = mock(StringRedisTemplate.class);
+ @SuppressWarnings("unchecked")
+ ValueOperations