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