Middleware/mq-starter
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mq-starter-redis-stream): 去重 TTL / reclaim idle 外置为 mq.redis-stream.* 可配置

Browse Source 
原先 reclaim idle 与去重 TTL/前缀硬编码在 autoconfig 里，按业务处理时延调优只能改源码。

- 新增 RedisStreamMqProperties(前缀 mq.redis-stream)：reclaim-min-idle + idempotent.{key-prefix,processing/success/fail-ttl}
- 默认值与历史硬编码完全一致（reclaim 30s / processing 60s / success 300s / fail 300s / mq:idem:），不改变现有行为
- autoconfig 改由 Properties 取值装配 driver 与 RedisMqIdempotentStore
- 补默认值一致性 + mq.redis-stream.* 绑定覆盖单测；wiring 测试改按方法名反射以适配新签名

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
hongawen
2026-07-09 20:41:21 +08:00
parent e660afe2a2
commit 84034fb7d0
4 changed files with 116 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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());
}
}

48
mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqProperties.java Normal file
View File

@@ -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})。
*
* <p>把原先硬编码在 autoconfig 里的 reclaim idle 与去重 TTL 暴露为可配置项,便于按业务的
* 单条处理时延调整,而无需改源码。<b>默认值与历史硬编码保持一致</b>,不改变现有行为。
*
* <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
@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);
}
}

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

@@ -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();

49
mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqPropertiesTest.java Normal file
View File

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