fix(stream): autoclaim 先 XCLAIM 抢占再处理,杜绝多实例重复消费/重复死信

claimOne 此前死信分支不抢占、正常分支抢占但不校验结果,多实例并发认领同一条
PEL 消息时会被重复 dispatch、重复落死信。改 claim 返回是否真抢到(XCLAIM 不带
JUSTID 返回认领到的消息列表,空即未抢到),claimOne 先抢占成功才 dispatch / 落死信。
补 claimOne mock 单测(抢到/抢不到 × 正常/死信 共 4 例)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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 抢占 → 死信判定 → 认领重投。
      *
-     * <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数；因 {@link #claim} 用<b>不带 JUSTID</b>
+     * <p><b>先抢占再处理</b>：{@link #claim} 抢不到（idle 不足 minIdle，或并发实例已认领并重置 idle）
-     * 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。
+     * 即直接退出,保证多实例下同一条消息只被一个实例 dispatch / 落死信(幂等),不重复消费、不重复落死信。
+     *
+     * <p>{@code deliveries} 为本轮 {@code XCLAIM} 之前的投递计数（XPENDING 报告值）；因 {@link #claim}
+     * 用<b>不带 JUSTID</b> 的 XCLAIM，每次认领会把投递计数 +1，故 deliveries 随 autoclaim 轮次单调增长。
      * {@code maxRetry} 语义 = 最多投递 maxRetry 次（含首投），第 maxRetry 次仍 RETRY 即落死信。
-     * 死信判定<b>前置于 dispatch</b>：达上限直接落异常 + 强制 ACK，不再多跑一次业务副作用。
+     * 死信分支抢占后<b>只落异常 + 强制 ACK，不再 dispatch</b>，不跑业务副作用。
      */
-    private void claimOne(EnhanceStreamConsumerHandler<?> handler, String stream, String group,
+    void claimOne(EnhanceStreamConsumerHandler<?> handler, String stream, String group,
-                          long minIdle, int maxRetry, String id, long deliveries) {
+                  long minIdle, int maxRetry, String id, long deliveries) {
         List<MapRecord<String, Object, Object>> records =
                 redis.opsForStream().range(stream, Range.closed(id, id));
         if (records == null || records.isEmpty()) {
@@ -185,6 +189,12 @@ public class StreamAutoClaimer implements SmartLifecycle {
         MapRecord<String, Object, Object> rec = records.get(0);
         Map<String, String> 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），再重投
+        // 已认领（投递计数 +1、idle 重置）→ 重投
-        claim(stream, group, minIdle, id);
         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 {
      * <p>不带 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) {
+    private boolean claim(String stream, String group, long minIdle, String id) {
-        redis.execute((RedisCallback<Object>) connection -> connection.execute("XCLAIM",
+        Object claimed = redis.execute((RedisCallback<Object>) 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<String, String> toStringMap(Map<Object, Object> raw) {

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}）的多实例幂等。
+ *
+ * <p>核心: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<String, Object, Object> ops = mock(StreamOperations.class);
+
+    @SuppressWarnings("unchecked")
+    private StreamAutoClaimer newClaimer(RecordingHandler handler) {
+        when(redis.opsForStream()).thenReturn(ops);
+        Map<Object, Object> fields = new HashMap<>();
+        fields.put("enc", "plain");
+        fields.put("body", "{}");
+        MapRecord<String, Object, Object> 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<TestMsg> {
+        boolean dispatched = false;
+        boolean deadLettered = false;
+
+        @Override public String topic() { return STREAM; }
+        @Override public String group() { return GROUP; }
+        @Override public Class<TestMsg> 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<String, String> fields) {
+            dispatched = true;
+            return ConsumeResult.ACK;
+        }
+
+        @Override
+        public void saveExceptionFromFields(Map<String, String> fields, Exception cause) {
+            deadLettered = true;
+        }
+    }
+}