From 3e72653d1f3c408e16153255655687b84547f68e Mon Sep 17 00:00:00 2001 From: root Date: Fri, 26 Jun 2026 14:33:29 +0800 Subject: [PATCH] =?UTF-8?q?test(stream):=20=E8=A1=A5=20autoclaim=20?= =?UTF-8?q?=E5=A4=9A=E5=AE=9E=E4=BE=8B=E5=B9=B6=E5=8F=91=E5=B9=82=E7=AD=89?= =?UTF-8?q?=E7=9C=9F=20Redis=20=E9=9B=86=E6=88=90=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 两个 consumerName 不同的 StreamAutoClaimer 用 CyclicBarrier 同时对同一条 idle 超时的 PEL 消息发 XCLAIM,断言业务只被处理一次、PEL 清零。端到端覆盖 mock 单测与单实例死信 集成测试都证明不了的真实并发竞态(XCLAIM 归属转移 + idle 重置由 Redis 串行裁判),为 B1「先抢占再处理」补上端到端铁证。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../StreamAutoClaimConcurrentClaimTest.java | 231 ++++++++++++++++++ 1 file changed, 231 insertions(+) create mode 100644 src/test/java/com/njcn/middle/stream/container/StreamAutoClaimConcurrentClaimTest.java diff --git a/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimConcurrentClaimTest.java b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimConcurrentClaimTest.java new file mode 100644 index 0000000..0627f90 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimConcurrentClaimTest.java @@ -0,0 +1,231 @@ +package com.njcn.middle.stream.container; + +import com.njcn.middle.stream.autoconfig.RedisStreamProperties; +import com.njcn.middle.stream.domain.StreamBaseMessage; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import com.njcn.middle.stream.template.RedisStreamEnhanceTemplate; +import lombok.Data; +import lombok.EqualsAndHashCode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ConditionEvaluationResult; +import org.junit.jupiter.api.extension.ExecutionCondition; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.api.extension.ExtensionContext; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.stream.Consumer; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.PendingMessages; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.connection.stream.StreamReadOptions; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.net.InetSocketAddress; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * 集成测试（Gate）：autoclaim 多实例并发认领同一条 idle 超时的 PEL 消息时，业务必须只被处理一次。 + * + *

端到端验证 B1 幂等：两个 consumerName 不同的 {@link StreamAutoClaimer} 用 {@link CyclicBarrier} + * 同时对同一条消息发 {@code XCLAIM}（不带 JUSTID）。Redis 单线程串行处理命令——先到者认领成功、把 + * idle 重置为 0；后到者看到 {@code idle≈0 < minIdle} 必然抢不到（{@link StreamAutoClaimer#claim} + * 返回 false）→ 直接退出，不 dispatch。故无论哪个线程先到，{@code handleMessage} 恒只触发一次。 + * + *

这条用例覆盖纯 mock 单测（{@code StreamAutoClaimerClaimTest}）和单实例死信集成测试都证明不了的 + * 真实并发竞态：mock 短路了 XCLAIM 的真实归属转移 + idle 重置语义，唯有真 Redis 能裁判。 + * + *

测试放在 {@code container} 包以调用 package-private 的 {@link StreamAutoClaimer#claimOne}， + * 直接复现 autoclaim 单条认领，不依赖定时器时序，结果确定（非 flaky）。 + * + *

Redis 不可达时整类禁用（skipped）。连接默认 {@code 127.0.0.1:6379}，用 + * {@code -Dspring.redis.host=… -Dspring.redis.port=…} 覆盖。本地起：{@code docker run -d -p 6379:6379 redis:6.2}。 + */ +@ExtendWith(StreamAutoClaimConcurrentClaimTest.RedisReachableCondition.class) +@SpringBootTest( + classes = StreamAutoClaimConcurrentClaimTest.TestApp.class, + properties = { + "redis-stream.env-isolation=false", + "redis-stream.read-count=10" + }) +class StreamAutoClaimConcurrentClaimTest { + + private static final String TOPIC = "IT_Concurrent_Topic"; + private static final String GROUP = "IT_Concurrent_Group"; + private static final long MIN_IDLE_MS = 500L; + + @Autowired + private RedisStreamEnhanceTemplate template; + + @Autowired + private StringRedisTemplate redis; + + @Autowired + private StreamKeyBuilder keyBuilder; + + @Test + void onlyOneInstanceProcessesUnderConcurrentClaim() throws Exception { + CountingHandler handler = new CountingHandler(); + String stream = keyBuilder.buildStream(TOPIC); + + // 清历史 + 重建组（delete 会连带删组，XADD 不重建，故须手动 MKSTREAM 建组）。 + redis.delete(stream); + createGroup(stream, GROUP); + + // 发一条真实消息（含 enc/body，可被 dispatchMessage 正常 decode→反序列化）。 + ConcurrentMsg msg = new ConcurrentMsg(); + msg.setText("process-once"); + template.send(TOPIC, msg, false); + + // 用 origin consumer XREADGROUP 读进 PEL（不 ACK），投递计数=1；之后 idle 从 0 起涨。 + List> recs = redis.opsForStream().read( + Consumer.from(GROUP, "origin-consumer"), + StreamReadOptions.empty().count(10), + StreamOffset.create(stream, ReadOffset.lastConsumed())); + assertEquals(1, recs == null ? 0 : recs.size(), "应有 1 条进入 PEL"); + String id = recs.get(0).getId().getValue(); + + // 等 idle 超过 minIdle，使首个认领者满足 XCLAIM 的 min-idle 门槛。 + Thread.sleep(MIN_IDLE_MS + 300L); + + StreamAutoClaimer claimerA = startedClaimer("concurrent-claimer-A", handler); + StreamAutoClaimer claimerB = startedClaimer("concurrent-claimer-B", handler); + try { + CyclicBarrier gate = new CyclicBarrier(2); + ExecutorService pool = Executors.newFixedThreadPool(2); + List> futures = pool.invokeAll(Arrays.asList( + claimTask(gate, claimerA, handler, stream, id), + claimTask(gate, claimerB, handler, stream, id))); + for (Future f : futures) { + f.get(10, TimeUnit.SECONDS); // 解包任务内异常，使其转为测试失败 + } + pool.shutdownNow(); + } finally { + claimerA.stop(); + claimerB.stop(); + } + + assertEquals(1, handler.handled.get(), + "两个实例并发认领同一条，业务只应被处理一次（多实例幂等）"); + + PendingMessages pel = redis.opsForStream().pending(stream, GROUP, Range.unbounded(), 100); + assertEquals(0L, pel == null ? 0 : pel.size(), "认领方 ACK 后 PEL 应清零"); + } + + /** 两个 claimer 用 deliveries=1（< maxRetry）走正常 dispatch 分支；barrier 让二者尽量同时发 XCLAIM。 */ + private Callable claimTask(CyclicBarrier gate, StreamAutoClaimer claimer, + CountingHandler handler, String stream, String id) { + return () -> { + gate.await(); + claimer.claimOne(handler, stream, GROUP, MIN_IDLE_MS, 100, id, 1); + return null; + }; + } + + /** new + start()：start() 初始化 claimConsumer；interval 设大，定时器在测试窗内不触发，避免干扰。 */ + private StreamAutoClaimer startedClaimer(String consumerName, CountingHandler handler) { + RedisStreamProperties props = new RedisStreamProperties(); + props.setConsumerName(consumerName); + props.getAutoclaim().setIntervalMs(600000); + StreamAutoClaimer claimer = new StreamAutoClaimer( + redis, keyBuilder, props, Collections.singletonList(handler)); + claimer.start(); + return claimer; + } + + private void createGroup(String stream, String group) { + redis.execute((RedisCallback) conn -> { + try { + return conn.execute("XGROUP", + "CREATE".getBytes(StandardCharsets.UTF_8), + stream.getBytes(StandardCharsets.UTF_8), + group.getBytes(StandardCharsets.UTF_8), + "$".getBytes(StandardCharsets.UTF_8), + "MKSTREAM".getBytes(StandardCharsets.UTF_8)); + } catch (RuntimeException e) { + return null; // BUSYGROUP：组已存在，忽略 + } + }); + } + + /** 不注册任何 handler bean：autoconfig 的 listener/autoclaimer 因 handlers 为空而空转，不干扰本测试。 */ + @SpringBootConfiguration + @EnableAutoConfiguration + static class TestApp { + } + + @Data + @EqualsAndHashCode(callSuper = true) + static class ConcurrentMsg extends StreamBaseMessage { + private String text; + } + + /** 真实 dispatch 路径计数（decode→反序列化→handleMessage），统计业务被处理的次数。 */ + static class CountingHandler extends EnhanceStreamConsumerHandler { + final AtomicInteger handled = new AtomicInteger(); + + @Override + public String topic() { + return TOPIC; + } + + @Override + public String group() { + return GROUP; + } + + @Override + public Class messageType() { + return ConcurrentMsg.class; + } + + @Override + public void handleMessage(ConcurrentMsg message) { + handled.incrementAndGet(); + } + + @Override + public boolean isRetry() { + return true; + } + + @Override + public boolean throwException() { + return false; + } + } + + /** Redis 不可达时禁用整个测试类（上下文加载即连 Redis，故须在加载前判断）。 */ + static class RedisReachableCondition implements ExecutionCondition { + @Override + public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) { + String host = System.getProperty("spring.redis.host", "127.0.0.1"); + int port = Integer.parseInt(System.getProperty("spring.redis.port", "6379")); + try (Socket s = new Socket()) { + s.connect(new InetSocketAddress(host, port), 500); + return ConditionEvaluationResult.enabled("redis reachable at " + host + ":" + port); + } catch (Exception e) { + return ConditionEvaluationResult.disabled( + "no redis at " + host + ":" + port + ", skip integration test"); + } + } + } +}