feat(mq-starter-idempotent-redis): 新建 driver 中立去重模块，迁入 RedisMqIdempotentStore 及其测试
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
30
mq-starter-idempotent-redis/pom.xml
Normal file
30
mq-starter-idempotent-redis/pom.xml
Normal file
@@ -0,0 +1,30 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
|
||||
<artifactId>mq-starter-idempotent-redis</artifactId>
|
||||
<name>mq-starter-idempotent-redis 消费去重 Redis 存储(driver 中立)</name>
|
||||
<dependencies>
|
||||
<dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</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>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId><version>${springboot.version}</version></dependency>
|
||||
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
|
||||
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
|
||||
</dependencies>
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId></plugin>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<configuration>
|
||||
<includes>
|
||||
<include>**/*Test.java</include>
|
||||
<include>**/*Tests.java</include>
|
||||
<include>**/*IT.java</include>
|
||||
</includes>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
</project>
|
||||
@@ -0,0 +1,69 @@
|
||||
package com.njcn.mq.idempotent.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));
|
||||
}
|
||||
}
|
||||
@@ -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; }
|
||||
}
|
||||
}
|
||||
@@ -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）。
|
||||
*
|
||||
* <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"), "无记录(标记已过期)不可视为完成");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user