diff --git a/mq-starter-core/pom.xml b/mq-starter-core/pom.xml index 81c26d0..a63dd49 100644 --- a/mq-starter-core/pom.xml +++ b/mq-starter-core/pom.xml @@ -8,6 +8,7 @@ com.alibabafastjson${fastjson.version} cn.hutoolhutool-all${hutool.version} org.projectlomboklombok${lombok.version}true + org.slf4jslf4j-api1.7.30provided org.springframework.bootspring-boot-configuration-processor${springboot.version}true org.springframework.bootspring-boot-starter-test${springboot.version}test 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 601eac2..8ede99d 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 lombok.extern.slf4j.Slf4j; import java.util.ArrayList; import java.util.Collections; @@ -13,8 +14,10 @@ import java.util.List; * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。 * 纯逻辑、不依赖 spring / redis，便于单测。重试本身交给 driver 的 MQ 原生重投。 */ +@Slf4j public class MqConsumeDispatcher { + // >1 时攒批；注意：无 idle flush，低流量下未满批会停滞、进程重启丢未入库批（与 legacy 一致，适合高流量流）。 private final int batchSize; private final boolean idempotent; private final MqIdempotentStore store; @@ -38,13 +41,13 @@ public class MqConsumeDispatcher { try { consumer.consume(Collections.singletonList(msg)); } catch (Exception e) { - if (errorHandler != null) { - errorHandler.onError(msg, MqErrorIdentity.SINGLE, e); - } + // errorHandler 自身失败不得跳过 markFail：否则去重标记停在 processing， + // reclaim 重投时会被去重挡掉并 ACK → 静默丢消息。 + safeOnError(msg, MqErrorIdentity.SINGLE, e); if (idempotent && store != null) { store.markFail(msg.getKey()); } - throw e; // 交 driver 重投 + throw e; // 抛原始异常，交 driver 重投 } if (idempotent && store != null) { store.markSuccess(msg.getKey()); @@ -66,10 +69,9 @@ public class MqConsumeDispatcher { } catch (Exception e) { // 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列： // 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。 - if (errorHandler != null) { - for (MqMessage m : flush) { - errorHandler.onError(m, MqErrorIdentity.SINGLE, e); - } + log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e); + for (MqMessage m : flush) { + safeOnError(m, MqErrorIdentity.SINGLE, e); } return; } @@ -81,6 +83,16 @@ public class MqConsumeDispatcher { } } + /** 调用异常落库回调，并隔离其自身异常（落库是 best-effort，不得掩盖/中断主流程）。 */ + private void safeOnError(MqMessage msg, MqErrorIdentity identity, Exception cause) { + if (errorHandler == null) return; + try { + errorHandler.onError(msg, identity, cause); + } catch (Exception ehEx) { + log.error("[mq] errorHandler 落库失败 key={}（不影响主流程）", msg.getKey(), ehEx); + } + } + public static Builder builder() { return new Builder(); } public static class Builder { diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java index 6a756ac..2e74217 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java @@ -8,10 +8,12 @@ import com.njcn.mq.spi.MqDriver; import com.njcn.mq.spi.MqIdempotentStore; import com.njcn.mq.spi.MqMessageListener; import com.njcn.mq.spi.MqSubscription; +import lombok.extern.slf4j.Slf4j; import org.springframework.context.SmartLifecycle; import java.util.ArrayList; import java.util.List; +@Slf4j public class MqListenerRegistry implements SmartLifecycle { private final MqDriver driver; private final List endpoints; @@ -34,6 +36,10 @@ public class MqListenerRegistry implements SmartLifecycle { if (running) return; for (MqListenerEndpoint ep : endpoints) { MqListener meta = ep.getMeta(); + if (meta.idempotent() && store == null) { + log.warn("[mq] @MqListener(idempotent=true) 但容器中无 MqIdempotentStore bean，去重不生效: topic={} group={}", + meta.topic(), meta.group()); + } MqConsumeErrorHandler eh = resolveErrorHandler(meta); MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder() @@ -64,7 +70,12 @@ public class MqListenerRegistry implements SmartLifecycle { private MqConsumeErrorHandler resolveErrorHandler(MqListener meta) { if (errorHandlerResolver == null || meta.errorHandler().isEmpty()) return null; - return errorHandlerResolver.apply(meta.errorHandler()); + try { + return errorHandlerResolver.apply(meta.errorHandler()); + } catch (Exception e) { + throw new IllegalStateException("@MqListener(errorHandler=\"" + meta.errorHandler() + + "\") 指定的 bean 未找到: topic=" + meta.topic() + " group=" + meta.group(), e); + } } /** 业务调用适配：批量 endpoint 调 (List<payload>[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */ 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 6afba25..1febbfd 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 @@ -142,4 +142,28 @@ class MqConsumeDispatcherTest { assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success"); } + + @Test + void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() { + List failed = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { return true; } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { failed.add(key); } + }; + // errorHandler 自身抛异常（落库 sink 挂掉，常与错误风暴同时发生） + MqConsumeErrorHandler eh = (m, id, ex) -> { throw new RuntimeException("sink down"); }; + RuntimeException boom = new RuntimeException("biz boom"); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store).errorHandler(eh) + .consumer(batch -> { throw boom; }) + .build(); + + Exception thrown = assertThrows(Exception.class, + () -> d.dispatch(MqMessage.of("k1", "", "a"))); + + assertSame(boom, thrown, "errorHandler 失败不应掩盖原始消费异常"); + assertEquals(1, failed.size(), + "errorHandler 抛异常时仍须 markFail，否则去重标记停在 processing 会导致 reclaim 时被去重挡掉而丢消息"); + } } 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 3abd48d..6eecf68 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 @@ -41,6 +41,8 @@ public class RedisMqIdempotentStore implements MqIdempotentStore { return true; } // key 已存在：fail（上次失败）允许重新消费；processing / success 一律挡回。 + // 注意：fail→reopen 的 get+set 非原子，跨实例对「共享同一 key 的不同消息」可能都放行； + // 同一条消息由 stream PEL/claim 序列化，故仅影响不同消息共享 key 的边缘场景。 if (FAIL.equals(redis.opsForValue().get(k))) { redis.opsForValue().set(k, PROCESSING, Duration.ofSeconds(processingTtl)); return true; 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 e6de53d..5da965d 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 @@ -28,7 +28,9 @@ public class RedisStreamMqDriver implements MqDriver { private volatile boolean running = true; public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) { - this(redis, keyBuilder, props, 5000L); + // reclaim 默认 idle 30s：须大于正常处理耗时，避免多实例误 reclaim 仍在处理(in-flight)的消息造成重复消费； + // 非幂等消费在多实例下为 at-least-once，可通过 4 参构造按业务处理时延调整。 + this(redis, keyBuilder, props, 30_000L); } public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) { @@ -132,8 +134,7 @@ public class RedisStreamMqDriver implements MqDriver { if (deliveryCount > maxRetry) { if (sub.getRetryExhaustedHandler() != null) { try { - sub.getRetryExhaustedHandler().accept(MqMessage.of( - fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), null)); + sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields)); } catch (Exception ex) { log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex); } @@ -147,6 +148,18 @@ public class RedisStreamMqDriver implements MqDriver { } } + /** 解码消息体为业务 payload（best-effort：毒消息时 payload 为 null），用于重试耗尽落库回调。 */ + private MqMessage buildMessage(MqSubscription sub, Map fields) { + Object payload = null; + try { + String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY)); + payload = JSON.parseObject(json, sub.getPayloadType()); + } catch (Exception ignore) { + // 毒消息：payload 留 null，至少保留 key/tag 供落库 + } + return MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload); + } + private void ensureGroup(String stream, String group) { try { redis.execute((RedisCallback) c -> c.execute("XGROUP", diff --git a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java index 17ec62f..d35d826 100644 --- a/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java +++ b/mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java @@ -32,18 +32,20 @@ class RedisStreamMqDriverRetryIT { String topic = "MQ_RETRY_IT_" + System.nanoTime(); AtomicInteger attempts = new AtomicInteger(); - BlockingQueue exhausted = new LinkedBlockingQueue<>(); + BlockingQueue exhausted = new LinkedBlockingQueue<>(); driver.subscribe(MqSubscription.builder() .topic(topic).group("g1").tag("*").payloadType(String.class) .concurrency(1).batchSize(1).maxRetry(2) .listener(m -> { attempts.incrementAndGet(); throw new RuntimeException("always fail"); }) - .retryExhaustedHandler(m -> exhausted.add(m.getKey())) + .retryExhaustedHandler(exhausted::add) .build()); driver.send(topic, MqMessage.of("rk1", "", "data")); - String key = exhausted.poll(12, TimeUnit.SECONDS); - assertEquals("rk1", key, "重试耗尽应回调 retryExhaustedHandler"); + MqMessage exhaustedMsg = exhausted.poll(12, TimeUnit.SECONDS); + assertNotNull(exhaustedMsg, "重试耗尽应回调 retryExhaustedHandler"); + assertEquals("rk1", exhaustedMsg.getKey()); + assertEquals("data", exhaustedMsg.getPayload(), "耗尽回调应携带完整 payload，而非 null"); assertTrue(attempts.get() >= 2, "应实际重试若干次后才耗尽, attempts=" + attempts.get()); Thread.sleep(500);