From a2965dd80ef2f3d5cbaf0cd67ae89dc4c5837db9 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 26 Jun 2026 13:21:32 +0800 Subject: [PATCH] =?UTF-8?q?fix(stream):=20autoclaim=20=E5=85=88=20XCLAIM?= =?UTF-8?q?=20=E6=8A=A2=E5=8D=A0=E5=86=8D=E5=A4=84=E7=90=86,=E6=9D=9C?= =?UTF-8?q?=E7=BB=9D=E5=A4=9A=E5=AE=9E=E4=BE=8B=E9=87=8D=E5=A4=8D=E6=B6=88?= =?UTF-8?q?=E8=B4=B9/=E9=87=8D=E5=A4=8D=E6=AD=BB=E4=BF=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit claimOne 此前死信分支不抢占、正常分支抢占但不校验结果,多实例并发认领同一条 PEL 消息时会被重复 dispatch、重复落死信。改 claim 返回是否真抢到(XCLAIM 不带 JUSTID 返回认领到的消息列表,空即未抢到),claimOne 先抢占成功才 dispatch / 落死信。 补 claimOne mock 单测(抢到/抢不到 × 正常/死信 共 4 例)。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../stream/container/StreamAutoClaimer.java | 39 +++-- .../container/StreamAutoClaimerClaimTest.java | 141 ++++++++++++++++++ 2 files changed, 170 insertions(+), 10 deletions(-) create mode 100644 src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerClaimTest.java diff --git a/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java b/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java index 89406f1..a12784f 100644 --- a/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java +++ b/src/main/java/com/njcn/middle/stream/container/StreamAutoClaimer.java @@ -17,6 +17,7 @@ import org.springframework.data.redis.core.RedisCallback; import org.springframework.data.redis.core.StringRedisTemplate; import java.nio.charset.StandardCharsets; +import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -166,15 +167,18 @@ public class StreamAutoClaimer implements SmartLifecycle { } /** - * 处理单条 PEL 消息：取内容 → 死信判定前置 → 认领重投。 + * 处理单条 PEL 消息：取内容 → XCLAIM 抢占 → 死信判定 → 认领重投。 * - *

{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数；因 {@link #claim} 用不带 JUSTID - * 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。 + *

先抢占再处理：{@link #claim} 抢不到（idle 不足 minIdle，或并发实例已认领并重置 idle） + * 即直接退出,保证多实例下同一条消息只被一个实例 dispatch / 落死信(幂等),不重复消费、不重复落死信。 + * + *

{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数（XPENDING 报告值）；因 {@link #claim} + * 用不带 JUSTID 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。 * {@code maxRetry} 语义 = 最多投递 maxRetry 次（含首投），第 maxRetry 次仍 RETRY 即落死信。 - * 死信判定前置于 dispatch：达上限直接落异常 + 强制 ACK，不再多跑一次业务副作用。 + * 死信分支抢占后只落异常 + 强制 ACK，不再 dispatch，不跑业务副作用。 */ - private void claimOne(EnhanceStreamConsumerHandler handler, String stream, String group, - long minIdle, int maxRetry, String id, long deliveries) { + void claimOne(EnhanceStreamConsumerHandler handler, String stream, String group, + long minIdle, int maxRetry, String id, long deliveries) { List> records = redis.opsForStream().range(stream, Range.closed(id, id)); if (records == null || records.isEmpty()) { @@ -185,6 +189,12 @@ public class StreamAutoClaimer implements SmartLifecycle { MapRecord rec = records.get(0); Map fields = toStringMap(rec.getValue()); + // 先 XCLAIM 抢占：仅当真正认领到（idle 足够且未被并发实例抢走）才由本实例独占处理； + // 抢不到说明已有其它实例认领，直接退出——杜绝多实例对同一条消息重复 dispatch 或重复落死信。 + if (!claim(stream, group, minIdle, id)) { + return; + } + if (deliveries >= maxRetry) { // 投递计数已达上限 → 落异常 + 强制 ACK（死信，不再无限滞留 PEL） // 由 handler 内部反序列化后回调,使死信回调能拿到业务消息对象（解析失败兜底 null） @@ -194,8 +204,7 @@ public class StreamAutoClaimer implements SmartLifecycle { return; } - // 认领到本 claimer（同时把投递计数 +1、重置 idle），再重投 - claim(stream, group, minIdle, id); + // 已认领（投递计数 +1、idle 重置）→ 重投 ConsumeResult result = handler.dispatchMessage(fields); if (result == ConsumeResult.ACK) { redis.opsForStream().acknowledge(stream, group, rec.getId()); @@ -209,14 +218,24 @@ public class StreamAutoClaimer implements SmartLifecycle { *

不带 JUSTID 时 XCLAIM 会转移归属、重置 idle 并把投递计数 +1——后者是 maxRetry 死信判定的 * 依据：listener 用 {@code XREADGROUP lastConsumed()} 永不重读 PEL，若此处用 JUSTID 则投递计数 * 永久冻结在首投值 1，{@code deliveries >= maxRetry} 恒不成立、死信分支永不触发。 + * + * @return 是否真正认领到本条：XCLAIM 抢到返回非空消息列表，没抢到返回空。返回类型未知时保守当作 + * 抢到（宁可退化回"可能重复"也不让消息因误判没抢到而永久卡 PEL）。 */ - private void claim(String stream, String group, long minIdle, String id) { - redis.execute((RedisCallback) connection -> connection.execute("XCLAIM", + private boolean claim(String stream, String group, long minIdle, String id) { + Object claimed = redis.execute((RedisCallback) connection -> connection.execute("XCLAIM", stream.getBytes(StandardCharsets.UTF_8), group.getBytes(StandardCharsets.UTF_8), claimConsumer.getBytes(StandardCharsets.UTF_8), String.valueOf(minIdle).getBytes(StandardCharsets.UTF_8), id.getBytes(StandardCharsets.UTF_8))); + if (claimed instanceof Collection) { + return !((Collection) claimed).isEmpty(); + } + if (claimed instanceof Object[]) { + return ((Object[]) claimed).length > 0; + } + return claimed != null; } private static Map toStringMap(Map raw) { diff --git a/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerClaimTest.java b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerClaimTest.java new file mode 100644 index 0000000..6cbd428 --- /dev/null +++ b/src/test/java/com/njcn/middle/stream/container/StreamAutoClaimerClaimTest.java @@ -0,0 +1,141 @@ +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.ConsumeResult; +import com.njcn.middle.stream.handler.EnhanceStreamConsumerHandler; +import com.njcn.middle.stream.support.StreamKeyBuilder; +import org.junit.jupiter.api.Test; +import org.springframework.data.domain.Range; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.RecordId; +import org.springframework.data.redis.core.RedisCallback; +import org.springframework.data.redis.core.StreamOperations; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * 单元测试:autoclaim 认领单条 PEL 消息（{@link StreamAutoClaimer#claimOne}）的多实例幂等。 + * + *

核心:XCLAIM 真正抢到（返回非空）才由本实例 dispatch / 落死信;抢不到（并发实例已认领、 + * idle 被重置)直接退出,杜绝重复消费与重复死信回调。 + */ +class StreamAutoClaimerClaimTest { + + private static final String STREAM = "T"; + private static final String GROUP = "G"; + private static final String ID = "1-0"; + + private final StringRedisTemplate redis = mock(StringRedisTemplate.class); + @SuppressWarnings("unchecked") + private final StreamOperations ops = mock(StreamOperations.class); + + @SuppressWarnings("unchecked") + private StreamAutoClaimer newClaimer(RecordingHandler handler) { + when(redis.opsForStream()).thenReturn(ops); + Map fields = new HashMap<>(); + fields.put("enc", "plain"); + fields.put("body", "{}"); + MapRecord rec = mock(MapRecord.class); + when(rec.getValue()).thenReturn(fields); + when(rec.getId()).thenReturn(RecordId.of(ID)); + when(ops.range(eq(STREAM), any(Range.class))).thenReturn(Collections.singletonList(rec)); + return new StreamAutoClaimer(redis, new StreamKeyBuilder(false, ""), + new RedisStreamProperties(), Collections.singletonList(handler)); + } + + @SuppressWarnings("unchecked") + private void stubClaimAcquired(boolean acquired) { + when(redis.execute(any(RedisCallback.class))) + .thenReturn(acquired ? Collections.singletonList(new Object()) : Collections.emptyList()); + } + + @Test + void skipsDispatchWhenClaimLost() { + RecordingHandler handler = new RecordingHandler(); + StreamAutoClaimer claimer = newClaimer(handler); + stubClaimAcquired(false); // 并发实例已抢走，本实例 XCLAIM 抢不到 + + claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 0); + + assertFalse(handler.dispatched, "未抢到 XCLAIM 不应 dispatch"); + verify(ops, never()).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class)); + } + + @Test + void skipsDeadLetterWhenClaimLost() { + RecordingHandler handler = new RecordingHandler(); + StreamAutoClaimer claimer = newClaimer(handler); + stubClaimAcquired(false); + + claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 9); // deliveries 已超限 + + assertFalse(handler.deadLettered, "未抢到 XCLAIM 不应重复落死信"); + verify(ops, never()).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class)); + } + + @Test + void dispatchesWhenClaimAcquired() { + RecordingHandler handler = new RecordingHandler(); + StreamAutoClaimer claimer = newClaimer(handler); + stubClaimAcquired(true); + + claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 0); + + assertTrue(handler.dispatched, "抢到 XCLAIM 应 dispatch"); + verify(ops, times(1)).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class)); + } + + @Test + void deadLettersWhenClaimAcquiredAndOverLimit() { + RecordingHandler handler = new RecordingHandler(); + StreamAutoClaimer claimer = newClaimer(handler); + stubClaimAcquired(true); + + claimer.claimOne(handler, STREAM, GROUP, 1000L, 3, ID, 9); + + assertTrue(handler.deadLettered, "抢到且超限应落死信"); + assertFalse(handler.dispatched, "死信不应再 dispatch 业务"); + verify(ops, times(1)).acknowledge(eq(STREAM), eq(GROUP), any(RecordId.class)); + } + + static class TestMsg extends StreamBaseMessage { + } + + /** 记录 dispatch / 死信回调是否被触发的测试桩。 */ + static class RecordingHandler extends EnhanceStreamConsumerHandler { + boolean dispatched = false; + boolean deadLettered = false; + + @Override public String topic() { return STREAM; } + @Override public String group() { return GROUP; } + @Override public Class messageType() { return TestMsg.class; } + @Override public void handleMessage(TestMsg message) { } + @Override public boolean isRetry() { return true; } + @Override public boolean throwException() { return false; } + + @Override + public ConsumeResult dispatchMessage(Map fields) { + dispatched = true; + return ConsumeResult.ACK; + } + + @Override + public void saveExceptionFromFields(Map fields, Exception cause) { + deadLettered = true; + } + } +}