refactor(mq-starter-redis-stream): 去重 store 与 idempotent 配置迁至 mq-starter-idempotent-redis，模块改为依赖之

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

mq-starter-redis-stream/pom.xml

@@ -5,6 +5,7 @@
     <artifactId>mq-starter-redis-stream</artifactId>
     <dependencies>
         <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-idempotent-redis</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
         <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-autoconfigure</artifactId><version>${autoconfigure.version}</version></dependency>
         <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java

@@ -1,69 +0,0 @@
-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。
- *
- * <p>对齐 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));
-    }
-}

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java

@@ -3,7 +3,6 @@ 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;
@@ -24,15 +23,7 @@ public class RedisStreamMqDriverAutoConfiguration {
     public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder,
                              RedisStreamProperties props, RedisStreamMqProperties mqProps) {
         // reclaim idle 从 mq.redis-stream.reclaim-min-idle 读取（默认 30s）。
+        // 去重 store 不在这里装配:已中立化到 mq-starter-idempotent-redis(不按 mq.type 门控)。
         return new RedisStreamMqDriver(redis, streamKeyBuilder, props, mqProps.getReclaimMinIdle().toMillis());
     }
-
-    @Bean
-    @ConditionalOnMissingBean
-    public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis, RedisStreamMqProperties mqProps) {
-        // TTL / 前缀从 mq.redis-stream.idempotent.* 读取（默认 processing 60s / success 300s / fail 300s，前缀 mq:idem:）。
-        RedisStreamMqProperties.Idempotent idem = mqProps.getIdempotent();
-        return new RedisMqIdempotentStore(redis, idem.getKeyPrefix(),
-                idem.getProcessingTtl().getSeconds(), idem.getSuccessTtl().getSeconds(), idem.getFailTtl().getSeconds());
-    }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java

@@ -8,18 +8,13 @@ import java.time.Duration;
 /**
  * Redis Stream 驱动运行参数(前缀 {@code mq.redis-stream})。
  *
- * <p>把原先硬编码在 autoconfig 里的 reclaim idle 与去重 TTL 暴露为可配置项,便于按业务的
+ * <p>去重 TTL 配置已中立化到 {@code mq.idempotent.*}(mq-starter-idempotent-redis 模块),
- * 单条处理时延调整,而无需改源码。<b>默认值与历史硬编码保持一致</b>,不改变现有行为。
+ * 旧键 {@code mq.redis-stream.idempotent.*} 仍向后兼容但会打弃用告警。
  *
  * <pre>
  * mq:
  *   redis-stream:
  *     reclaim-min-idle: 30s        # 未 ACK 超过此时长才 reclaim 重投(须 &gt; 单条处理时延)
- *     idempotent:
- *       key-prefix: "mq:idem:"
- *       processing-ttl: 60s        # 建议 &gt; 单条处理时延,且 &lt; reclaim-min-idle 更稳
- *       success-ttl: 300s
- *       fail-ttl: 300s
  * </pre>
  */
 @Data
@@ -31,18 +26,4 @@ public class RedisStreamMqProperties {
      * 须大于单条业务的正常处理时延,避免仍在处理的消息被误 reclaim 造成重复消费。
      */
     private Duration reclaimMinIdle = Duration.ofSeconds(30);
-
-    private final Idempotent idempotent = new Idempotent();
-
-    @Data
-    public static class Idempotent {
-        /** 去重键前缀。 */
-        private String keyPrefix = "mq:idem:";
-        /** processing 标记 TTL:应大于单条业务处理时延。 */
-        private Duration processingTtl = Duration.ofSeconds(60);
-        /** success 标记 TTL:防重复消费的窗口。 */
-        private Duration successTtl = Duration.ofSeconds(300);
-        /** fail 标记 TTL:允许重新消费前的保留窗口。 */
-        private Duration failTtl = Duration.ofSeconds(300);
-    }
 }

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java

@@ -1,38 +0,0 @@
-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; }
-    }
-}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java

@@ -1,34 +0,0 @@
-package com.njcn.mq.driver.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）。
- *
- * <p>isCompleted 必须仅在 key 值为 {@code success} 时返回 true：processing / 无记录 一律 false，
- * 否则消费者崩溃后残留的 processing 标记会被 driver 当成"已完成"而 ACK，造成静默丢消息。
- */
-class RedisMqIdempotentStoreTest {
-
-    @Test
-    void isCompleted_trueOnlyForSuccessValue() {
-        StringRedisTemplate redis = mock(StringRedisTemplate.class);
-        @SuppressWarnings("unchecked")
-        ValueOperations<String, String> ops = mock(ValueOperations.class);
-        when(redis.opsForValue()).thenReturn(ops);
-        RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300);
-
-        when(ops.get("mq:idem:kSucc")).thenReturn("success");
-        when(ops.get("mq:idem:kProc")).thenReturn("processing");
-        when(ops.get("mq:idem:kNone")).thenReturn(null);
-
-        assertTrue(store.isCompleted("kSucc"), "success 应视为已完成，driver 可 ACK");
-        assertFalse(store.isCompleted("kProc"), "processing 不可视为完成，否则崩溃残留会被 ACK 丢消息");
-        assertFalse(store.isCompleted("kNone"), "无记录(标记已过期)不可视为完成");
-    }
-}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java

@@ -1,15 +1,16 @@
 package com.njcn.mq.driver.redis;
 
 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
+import com.njcn.mq.idempotent.redis.MqIdempotentRedisAutoConfiguration;
 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.boot.autoconfigure.AutoConfigurations;
-
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
-import java.lang.reflect.Method;
+import org.springframework.data.redis.core.StringRedisTemplate;
-import java.util.Arrays;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 /**
  * 装配顺序回归护栏（mq-starter 自身）。
@@ -37,18 +38,12 @@ class RedisStreamMqDriverWiringTest {
     }
 
     @Test
-    void declaresIdempotentStoreBean() {
+    void idempotentStoreStillOnClasspathViaIdempotentRedisModule() {
-        Method m = Arrays.stream(RedisStreamMqDriverAutoConfiguration.class.getDeclaredMethods())
+        // store 已迁到 mq-starter-idempotent-redis(pom 传递依赖):只引 redis-stream 单模块仍开箱带去重。
-                .filter(x -> x.getName().equals("mqIdempotentStore"))
+        // 若有人删掉该 pom 依赖,本测试编译即失败——这正是护栏目的。
-                .findFirst().orElse(null);
+        new ApplicationContextRunner()
-        assertThat(m)
+                .withConfiguration(AutoConfigurations.of(MqIdempotentRedisAutoConfiguration.class))
-                .as("redis-stream 模式应提供默认去重实现 @Bean(方法名 mqIdempotentStore)")
+                .withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class))
-                .isNotNull();
+                .run(ctx -> assertThat(ctx).hasSingleBean(MqIdempotentStore.class));
-        assertThat(m.getAnnotation(Bean.class))
-                .as("redis-stream 模式应提供默认去重实现")
-                .isNotNull();
-        assertThat(MqIdempotentStore.class)
-                .as("默认去重实现须为 MqIdempotentStore 类型")
-                .isAssignableFrom(m.getReturnType());
     }
 }

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java

@@ -16,6 +16,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
  * {@link RedisStreamMqProperties} 默认值与绑定行为单测(纯 Binder，不 refresh context / 不连 Redis）。
  *
  * <p>守护两点:① 默认值与历史硬编码一致(不改变现有行为);② mq.redis-stream.* 前缀能覆盖默认。
+ * 去重 TTL 配置已中立化到 mq.idempotent.*,由 mq-starter-idempotent-redis 模块守护。
  */
 class RedisStreamMqPropertiesTest {
 
@@ -23,18 +24,12 @@ class RedisStreamMqPropertiesTest {
     void defaults_matchLegacyHardcodedValues() {
         RedisStreamMqProperties p = new RedisStreamMqProperties();
         assertEquals(Duration.ofSeconds(30), p.getReclaimMinIdle(), "reclaim idle 默认应保持 30s");
-        assertEquals("mq:idem:", p.getIdempotent().getKeyPrefix());
-        assertEquals(Duration.ofSeconds(60), p.getIdempotent().getProcessingTtl());
-        assertEquals(Duration.ofSeconds(300), p.getIdempotent().getSuccessTtl());
-        assertEquals(Duration.ofSeconds(300), p.getIdempotent().getFailTtl());
     }
 
     @Test
     void binds_mqRedisStreamPrefix_overridesDefaults() {
         Map<String, Object> src = new HashMap<>();
         src.put("mq.redis-stream.reclaim-min-idle", "90s");
-        src.put("mq.redis-stream.idempotent.processing-ttl", "15s");
-        src.put("mq.redis-stream.idempotent.key-prefix", "biz:idem:");
         MutablePropertySources ps = new MutablePropertySources();
         ps.addFirst(new MapPropertySource("test", src));
 
@@ -42,8 +37,5 @@ class RedisStreamMqPropertiesTest {
                 .bind("mq.redis-stream", RedisStreamMqProperties.class).get();
 
         assertEquals(Duration.ofSeconds(90), bound.getReclaimMinIdle());
-        assertEquals(Duration.ofSeconds(15), bound.getIdempotent().getProcessingTtl());
-        assertEquals("biz:idem:", bound.getIdempotent().getKeyPrefix());
-        assertEquals(Duration.ofSeconds(300), bound.getIdempotent().getSuccessTtl(), "未覆盖项保持默认");
     }
 }