Merge: mq-starter 消费侧通用化 + redis-stream 可放量（去重/批量/异常落库/重试 reclaim + opus review 修复）

- [Critical] dispatcher errorHandler 抛异常时仍保证 markFail + 抛原始异常（safeOnError 隔离）：
  否则去重标记停在 processing，reclaim 重投被去重挡掉并 ACK → 静默丢消息
- [Important] 重试耗尽回调携带完整 payload（reclaim 解码 body，不再传 null）
- [Important] reclaim 默认 idle 5s→30s，避免多实例误 reclaim in-flight 消息；注释 at-least-once 语义
- [Important] idempotent=true 但无 MqIdempotentStore bean 时 start() 告警
- core 引入 slf4j-api(provided)；批量失败 / errorHandler 失败均记日志
- errorHandler bean 未找到时抛带 topic/group 的清晰错误
- 文档化已知限制（批量无 idle flush、去重 fail→reopen 非原子）
- 全仓 29 测试绿（含连真实 redis 的 IT）

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

mq-starter-core/pom.xml

@@ -8,6 +8,7 @@
         <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
         <dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>${hutool.version}</version></dependency>
         <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
+        <dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.30</version><scope>provided</scope></dependency>
         <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><version>${springboot.version}</version><optional>true</optional></dependency>
         <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
     </dependencies>

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 {

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<MqListenerEndpoint> 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&lt;payload&gt;[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */

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<String> 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 时被去重挡掉而丢消息");
+    }
 }

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;

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<String, String> 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<Object>) c -> c.execute("XGROUP",

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<String> exhausted = new LinkedBlockingQueue<>();
+        BlockingQueue<MqMessage> 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);