diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java index 8ede99d..6f2622c 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java @@ -4,6 +4,7 @@ import com.njcn.mq.core.MqMessage; import com.njcn.mq.spi.MqConsumeErrorHandler; import com.njcn.mq.spi.MqErrorIdentity; import com.njcn.mq.spi.MqIdempotentStore; +import com.njcn.mq.spi.MqMessageInProgressException; import lombok.extern.slf4j.Slf4j; import java.util.ArrayList; @@ -35,7 +36,15 @@ public class MqConsumeDispatcher { public void dispatch(MqMessage msg) throws Exception { if (idempotent && store != null && !store.tryAcquire(msg.getKey())) { - return; // processing 中或已 success，跳过 + // 跳过分两类，决定 driver 能否 ACK： + // - 已 success：正常返回 → driver ACK，挡掉重复投递（正常去重）； + // - 仍 processing（并发处理中，或消费者在 mark 前崩溃残留的标记）：抛信号异常 → + // driver 不 ACK、保留消息，待 processing 标记过期后真正重投。否则该消息可能从未被 + // 成功消费却被去重挡回并 ACK → 静默丢消息。 + if (store.isCompleted(msg.getKey())) { + return; + } + throw new MqMessageInProgressException(msg.getKey()); } if (batchSize <= 1) { try { diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java index b04c4f8..06aeb5c 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java @@ -21,4 +21,22 @@ public interface MqIdempotentStore { /** 消费失败,标记 fail(允许后续重新消费)。 */ void markFail(String key); + + /** + * {@link #tryAcquire(String)} 返回 false(跳过)后,判断该 key 是否已消费成功。 + * + *

用于让 driver 区分两类"跳过",这决定了 driver 能否 ACK 该消息: + *

+ * + * @return true=已成功(driver 可 ACK);false=processing 中(driver 应保留待重投)。 + * 默认返回 true 以保持旧行为(向后兼容);具备状态机的实现应覆写。 + */ + default boolean isCompleted(String key) { + return true; + } } diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageInProgressException.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageInProgressException.java new file mode 100644 index 0000000..98f2ab7 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageInProgressException.java @@ -0,0 +1,21 @@ +package com.njcn.mq.spi; + +/** + * 控制流信号:消费去重标记仍处于 {@code processing}(并发处理中,或消费者在 mark 前崩溃残留的标记)。 + * + *

本次投递既不能视为成功(否则 driver 会 ACK,导致该消息若从未被成功消费即静默丢失), + * 也不属于业务处理失败。driver 捕获后应保留消息(不 ACK),待 processing 标记过期后由 MQ 真正重投。 + * + *

不携带堆栈(纯控制流,规避噪音与开销)。 + */ +public class MqMessageInProgressException extends RuntimeException { + + public MqMessageInProgressException(String key) { + super("message in-progress (dedup processing), leave for replay: key=" + key); + } + + @Override + public synchronized Throwable fillInStackTrace() { + return this; + } +} diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java index 1febbfd..f8f4924 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java +++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java @@ -143,6 +143,50 @@ class MqConsumeDispatcherTest { assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success"); } + @Test + void idempotent_skipInProgress_throwsSoDriverWontAckAndReplays() { + // 崩溃残留场景:去重标记停在 processing(tryAcquire=false 且 isCompleted=false)。 + // 消费者在 tryAcquire 之后、markSuccess/markFail 之前崩溃,消息从未被成功消费。 + // 若此时静默返回,driver 会把这条消息 ACK → 静默丢消息(Critical)。 + // 正确行为:抛异常,使 driver 不 ACK、保留消息,待 processing 标记过期后真正重投。 + List> consumed = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return false; } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { } + @Override public boolean isCompleted(String key) { return false; } + }; + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store) + .consumer(consumed::add) + .build(); + + assertThrows(Exception.class, + () -> d.dispatch(MqMessage.of("k1", "", "a")), + "processing 残留(非 success)跳过时应抛异常,使 driver 不 ACK、留待过期后重投,避免静默丢消息"); + assertEquals(0, consumed.size(), "不应消费"); + } + + @Test + void idempotent_skipCompleted_returnsNormallySoDriverAcks() { + // 已成功的重复投递:应被静默去重(driver 正常 ACK),不抛异常、不重复消费。 + List> consumed = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return false; } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { } + @Override public boolean isCompleted(String key) { return true; } + }; + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store) + .consumer(consumed::add) + .build(); + + assertDoesNotThrow(() -> d.dispatch(MqMessage.of("k1", "", "a")), + "已成功的重复投递应被静默去重(driver ACK),不抛异常"); + assertEquals(0, consumed.size(), "不应重复消费"); + } + @Test void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() { List failed = new ArrayList<>(); diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java index 6eecf68..fd10474 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java @@ -59,4 +59,11 @@ public class RedisMqIdempotentStore implements MqIdempotentStore { 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)); + } } diff --git a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java index 5da965d..c3da33b 100644 --- a/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java +++ b/mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java @@ -6,6 +6,7 @@ import com.njcn.middle.stream.constant.StreamMessageConstant; import com.njcn.middle.stream.support.StreamKeyBuilder; import com.njcn.mq.core.MqMessage; import com.njcn.mq.spi.MqDriver; +import com.njcn.mq.spi.MqMessageInProgressException; import com.njcn.mq.spi.MqSubscription; import lombok.extern.slf4j.Slf4j; import org.springframework.data.redis.connection.stream.*; @@ -98,6 +99,11 @@ public class RedisStreamMqDriver implements MqDriver { try { sub.getListener().onMessage(msg); return true; + } catch (MqMessageInProgressException inProgress) { + // 去重标记仍 processing（并发处理中/上次崩溃残留）：不 ACK，留 PEL， + // 待 processing 标记过期后由 reclaimPending 重新 acquire 并真正重投。 + log.debug("[mq-redis] in-progress, leave PEL key={}", msg.getKey()); + return false; } catch (Exception e) { log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e); return false; // 留 PEL，由 reclaimPending 重投或落库 diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java new file mode 100644 index 0000000..f11bc97 --- /dev/null +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreTest.java @@ -0,0 +1,34 @@ +package com.njcn.mq.driver.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）。 + * + *

isCompleted 必须仅在 key 值为 {@code success} 时返回 true：processing / 无记录 一律 false， + * 否则消费者崩溃后残留的 processing 标记会被 driver 当成"已完成"而 ACK，造成静默丢消息。 + */ +class RedisMqIdempotentStoreTest { + + @Test + void isCompleted_trueOnlyForSuccessValue() { + StringRedisTemplate redis = mock(StringRedisTemplate.class); + @SuppressWarnings("unchecked") + ValueOperations 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"), "无记录(标记已过期)不可视为完成"); + } +}