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 4dac641..63ee1b6 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 @@ -8,6 +8,7 @@ import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +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; @@ -16,17 +17,22 @@ import org.springframework.data.redis.core.StringRedisTemplate; @AutoConfigureBefore(MqCoreAutoConfiguration.class) @ConditionalOnClass(StringRedisTemplate.class) @ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream") +@EnableConfigurationProperties(RedisStreamMqProperties.class) public class RedisStreamMqDriverAutoConfiguration { @Bean @ConditionalOnMissingBean - public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) { - return new RedisStreamMqDriver(redis, streamKeyBuilder, props); + public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, + RedisStreamProperties props, RedisStreamMqProperties mqProps) { + // reclaim idle 从 mq.redis-stream.reclaim-min-idle 读取(默认 30s)。 + return new RedisStreamMqDriver(redis, streamKeyBuilder, props, mqProps.getReclaimMinIdle().toMillis()); } @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); + 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 new file mode 100644 index 0000000..cd780ef --- /dev/null +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java @@ -0,0 +1,48 @@ +package com.njcn.mq.driver.redis; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.time.Duration; + +/** + * Redis Stream 驱动运行参数(前缀 {@code mq.redis-stream})。 + * + *

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

+ * 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 +@ConfigurationProperties(prefix = "mq.redis-stream") +public class RedisStreamMqProperties { + + /** + * PEL reclaim 触发的最小 idle:消息投递后未 ACK 超过此时长才会被 reclaim 重投。 + * 须大于单条业务的正常处理时延,避免仍在处理的消息被误 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/RedisStreamMqDriverWiringTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java index 953e0de..98f5256 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 @@ -5,9 +5,9 @@ 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 java.util.Arrays; import static org.assertj.core.api.Assertions.assertThat; @@ -37,9 +37,13 @@ class RedisStreamMqDriverWiringTest { } @Test - void declaresIdempotentStoreBean() throws Exception { - Method m = RedisStreamMqDriverAutoConfiguration.class - .getDeclaredMethod("mqIdempotentStore", StringRedisTemplate.class); + 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(); 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 new file mode 100644 index 0000000..b3e3f08 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java @@ -0,0 +1,49 @@ +package com.njcn.mq.driver.redis; + +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.core.env.MapPropertySource; +import org.springframework.core.env.MutablePropertySources; + +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * {@link RedisStreamMqProperties} 默认值与绑定行为单测(纯 Binder,不 refresh context / 不连 Redis)。 + * + *

守护两点:① 默认值与历史硬编码一致(不改变现有行为);② mq.redis-stream.* 前缀能覆盖默认。 + */ +class RedisStreamMqPropertiesTest { + + @Test + 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)); + + RedisStreamMqProperties bound = new Binder(ConfigurationPropertySources.from(ps)) + .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(), "未覆盖项保持默认"); + } +}