diff --git a/mq-starter-redis-stream/pom.xml b/mq-starter-redis-stream/pom.xml index 60ad23c..0347403 100644 --- a/mq-starter-redis-stream/pom.xml +++ b/mq-starter-redis-stream/pom.xml @@ -5,6 +5,7 @@ mq-starter-redis-stream com.njcnmq-starter-core1.0.0-SNAPSHOT + com.njcnmq-starter-idempotent-redis1.0.0-SNAPSHOT com.njcnredis-stream-springboot-starter1.0.0-SNAPSHOT org.springframework.bootspring-boot-autoconfigure${autoconfigure.version} com.alibabafastjson${fastjson.version} 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 deleted file mode 100644 index fd10474..0000000 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java +++ /dev/null @@ -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。 - * - *

对齐 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-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 63ee1b6..9e5ada5 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,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()); - } } diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java index cd780ef..0be29f6 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java +++ b/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})。 * - *

把原先硬编码在 autoconfig 里的 reclaim idle 与去重 TTL 暴露为可配置项,便于按业务的 - * 单条处理时延调整,而无需改源码。默认值与历史硬编码保持一致,不改变现有行为。 + *

去重 TTL 配置已中立化到 {@code mq.idempotent.*}(mq-starter-idempotent-redis 模块), + * 旧键 {@code mq.redis-stream.idempotent.*} 仍向后兼容但会打弃用告警。 * * 

  * mq:
  *   redis-stream:
  *     reclaim-min-idle: 30s        # 未 ACK 超过此时长才 reclaim 重投(须 > 单条处理时延)
- *     idempotent:
- *       key-prefix: "mq:idem:"
- *       processing-ttl: 60s        # 建议 > 单条处理时延,且 < reclaim-min-idle 更稳
- *       success-ttl: 300s
- *       fail-ttl: 300s
  *
*/ @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); - } } 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 deleted file mode 100644 index 5dd81a2..0000000 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java +++ /dev/null @@ -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; } - } -} diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java deleted file mode 100644 index f11bc97..0000000 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java +++ /dev/null @@ -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）。 - * - *

isCompleted 必须仅在 key 值为 {@code success} 时返回 true：processing / 无记录 一律 false， - * 否则消费者崩溃后残留的 processing 标记会被 driver 当成"已完成"而 ACK，造成静默丢消息。 - */ -class RedisMqIdempotentStoreTest { - - @Test - void isCompleted_trueOnlyForSuccessValue() { - StringRedisTemplate redis = mock(StringRedisTemplate.class); - @SuppressWarnings("unchecked") - ValueOperations 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"), "无记录(标记已过期)不可视为完成"); - } -} 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 98f5256..e78c632 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,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 java.lang.reflect.Method; -import java.util.Arrays; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.data.redis.core.StringRedisTemplate; 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() { - Method m = Arrays.stream(RedisStreamMqDriverAutoConfiguration.class.getDeclaredMethods()) - .filter(x -> x.getName().equals("mqIdempotentStore")) - .findFirst().orElse(null); - assertThat(m) - .as("redis-stream 模式应提供默认去重实现 @Bean(方法名 mqIdempotentStore)") - .isNotNull(); - assertThat(m.getAnnotation(Bean.class)) - .as("redis-stream 模式应提供默认去重实现") - .isNotNull(); - assertThat(MqIdempotentStore.class) - .as("默认去重实现须为 MqIdempotentStore 类型") - .isAssignableFrom(m.getReturnType()); + void idempotentStoreStillOnClasspathViaIdempotentRedisModule() { + // store 已迁到 mq-starter-idempotent-redis(pom 传递依赖):只引 redis-stream 单模块仍开箱带去重。 + // 若有人删掉该 pom 依赖,本测试编译即失败——这正是护栏目的。 + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(MqIdempotentRedisAutoConfiguration.class)) + .withBean(StringRedisTemplate.class, () -> mock(StringRedisTemplate.class)) + .run(ctx -> assertThat(ctx).hasSingleBean(MqIdempotentStore.class)); } } diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java index b3e3f08..97e8ca4 100644 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java +++ b/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）。 * *

守护两点:① 默认值与历史硬编码一致(不改变现有行为);② 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 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(), "未覆盖项保持默认"); } }