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

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/annotation/MqListener.java

@@ -13,4 +13,8 @@ public @interface MqListener {
     int batchSize() default 1;
     int maxRetry() default 3;
     String tag() default "*";
+    /** 是否启用消费去重（按 key）。默认关闭。 */
+    boolean idempotent() default false;
+    /** 异常落库回调的 bean 名；空表示不落库。 */
+    String errorHandler() default "";
 }

mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqCoreAutoConfiguration.java

@@ -3,7 +3,9 @@ import com.njcn.mq.container.MqListenerAnnotationBeanPostProcessor;
 import com.njcn.mq.container.MqListenerRegistry;
 import com.njcn.mq.core.DefaultMqTemplate;
 import com.njcn.mq.core.MqTemplate;
+import com.njcn.mq.spi.MqConsumeErrorHandler;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqIdempotentStore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
@@ -27,7 +29,13 @@ public class MqCoreAutoConfiguration {
     @Bean
     @ConditionalOnBean(MqDriver.class)
     @ConditionalOnMissingBean
-    public MqListenerRegistry mqListenerRegistry(MqDriver driver, MqListenerAnnotationBeanPostProcessor bpp) {
+    public MqListenerRegistry mqListenerRegistry(MqDriver driver,
-        return new MqListenerRegistry(driver, bpp.getEndpoints());
+                                                 MqListenerAnnotationBeanPostProcessor bpp,
+                                                 org.springframework.beans.factory.ObjectProvider<MqIdempotentStore> storeProvider,
+                                                 org.springframework.context.ApplicationContext appCtx) {
+        MqIdempotentStore store = storeProvider.getIfAvailable();
+        java.util.function.Function<String, MqConsumeErrorHandler> resolver =
+                name -> appCtx.getBean(name, MqConsumeErrorHandler.class);
+        return new MqListenerRegistry(driver, bpp.getEndpoints(), store, resolver);
     }
 }

mq-starter-core/src/main/java/com/njcn/mq/container/MqBatchConsumer.java

@@ -0,0 +1,14 @@
+package com.njcn.mq.container;
+
+import com.njcn.mq.core.MqMessage;
+import java.util.List;
+
+/**
+ * 业务消费回调（core 内部契约）。
+ * 由 {@link MqListenerRegistry} 把 @MqListener 方法的反射调用适配成统一的「批量入参」形态：
+ * 单条流传 size=1 的 list，批量流传攒满的 list。dispatcher 只面向它编排，不关心反射细节。
+ */
+@FunctionalInterface
+public interface MqBatchConsumer {
+    void consume(List<MqMessage> batch) throws Exception;
+}

mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java

@@ -0,0 +1,111 @@
+package com.njcn.mq.container;
+
+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;
+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;
+    private final MqConsumeErrorHandler errorHandler;
+    private final MqBatchConsumer consumer;
+    private final List<MqMessage> buffer = new ArrayList<>();
+
+    private MqConsumeDispatcher(Builder b) {
+        this.batchSize = b.batchSize;
+        this.idempotent = b.idempotent;
+        this.store = b.store;
+        this.errorHandler = b.errorHandler;
+        this.consumer = b.consumer;
+    }
+
+    public void dispatch(MqMessage msg) throws Exception {
+        if (idempotent && store != null && !store.tryAcquire(msg.getKey())) {
+            return; // processing 中或已 success，跳过
+        }
+        if (batchSize <= 1) {
+            try {
+                consumer.consume(Collections.singletonList(msg));
+            } catch (Exception e) {
+                // errorHandler 自身失败不得跳过 markFail：否则去重标记停在 processing，
+                // reclaim 重投时会被去重挡掉并 ACK → 静默丢消息。
+                safeOnError(msg, MqErrorIdentity.SINGLE, e);
+                if (idempotent && store != null) {
+                    store.markFail(msg.getKey());
+                }
+                throw e; // 抛原始异常，交 driver 重投
+            }
+            if (idempotent && store != null) {
+                store.markSuccess(msg.getKey());
+            }
+            return;
+        }
+        // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。
+        List<MqMessage> flush = null;
+        synchronized (buffer) {
+            buffer.add(msg);
+            if (buffer.size() >= batchSize) {
+                flush = new ArrayList<>(buffer);
+                buffer.clear();
+            }
+        }
+        if (flush != null) {
+            try {
+                consumer.consume(flush);
+            } catch (Exception e) {
+                // 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列：
+                // 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。
+                log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e);
+                for (MqMessage m : flush) {
+                    safeOnError(m, MqErrorIdentity.SINGLE, e);
+                }
+                return;
+            }
+            if (idempotent && store != null) {
+                for (MqMessage m : flush) {
+                    store.markSuccess(m.getKey());
+                }
+            }
+        }
+    }
+
+    /** 调用异常落库回调，并隔离其自身异常（落库是 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 {
+        private int batchSize = 1;
+        private boolean idempotent = false;
+        private MqIdempotentStore store;
+        private MqConsumeErrorHandler errorHandler;
+        private MqBatchConsumer consumer;
+        public Builder batchSize(int n) { this.batchSize = n; return this; }
+        public Builder idempotent(boolean v) { this.idempotent = v; return this; }
+        public Builder store(MqIdempotentStore s) { this.store = s; return this; }
+        public Builder errorHandler(MqConsumeErrorHandler h) { this.errorHandler = h; return this; }
+        public Builder consumer(MqBatchConsumer c) { this.consumer = c; return this; }
+        public MqConsumeDispatcher build() { return new MqConsumeDispatcher(this); }
+    }
+}

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java

@@ -26,9 +26,21 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor
             if (wantsContext && params[1] != MqContext.class) {
                 throw new IllegalStateException("@MqListener 第二参须为 MqContext: " + m);
             }
-            Class<?> payloadType = params[0];
+            Class<?> first = params[0];
+            boolean batch = List.class.isAssignableFrom(first);
+            Class<?> payloadType;
+            if (batch) {
+                // 批量方法 (List<T>)：取泛型实参 T 作为 payloadType，driver 仍按 T 逐条反序列化，dispatcher 负责攒批。
+                java.lang.reflect.Type gt = m.getGenericParameterTypes()[0];
+                if (!(gt instanceof java.lang.reflect.ParameterizedType)) {
+                    throw new IllegalStateException("@MqListener 批量方法须声明具体元素类型 List<T>: " + m);
+                }
+                payloadType = (Class<?>) ((java.lang.reflect.ParameterizedType) gt).getActualTypeArguments()[0];
+            } else {
+                payloadType = first;
+            }
             m.setAccessible(true);
-            endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext));
+            endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch));
         }
         return bean;
     }

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerEndpoint.java

@@ -12,4 +12,6 @@ public class MqListenerEndpoint {
     private MqListener meta;
     private Class<?> payloadType;
     private boolean wantsContext;
+    /** true=方法签名为 (List&lt;T&gt;)，批量消费；payloadType 此时为元素类型 T。 */
+    private boolean batch;
 }

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java

@@ -2,40 +2,104 @@ package com.njcn.mq.container;
 import com.njcn.mq.annotation.MqListener;
 import com.njcn.mq.core.MqContext;
 import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqConsumeErrorHandler;
+import com.njcn.mq.spi.MqErrorIdentity;
 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;
+    private final MqIdempotentStore store;
+    private final java.util.function.Function<String, MqConsumeErrorHandler> errorHandlerResolver;
     private volatile boolean running = false;
 
     public MqListenerRegistry(MqDriver driver, List<MqListenerEndpoint> endpoints) {
+        this(driver, endpoints, null, null);
+    }
+
+    public MqListenerRegistry(MqDriver driver, List<MqListenerEndpoint> endpoints,
+                              MqIdempotentStore store,
+                              java.util.function.Function<String, MqConsumeErrorHandler> errorHandlerResolver) {
         this.driver = driver; this.endpoints = endpoints;
+        this.store = store; this.errorHandlerResolver = errorHandlerResolver;
     }
 
     @Override public synchronized void start() {
         if (running) return;
         for (MqListenerEndpoint ep : endpoints) {
             MqListener meta = ep.getMeta();
-            MqMessageListener listener = (MqMessage msg) -> invoke(ep, msg);
+            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()
+                    .batchSize(meta.batchSize())
+                    .consumer(batch -> consume(ep, batch));
+            if (meta.idempotent()) {
+                db.idempotent(true).store(store);
+            }
+            if (eh != null) {
+                db.errorHandler(eh);
+            }
+            MqConsumeDispatcher dispatcher = db.build();
+
+            java.util.function.Consumer<MqMessage> retryExhausted = eh == null ? null
+                    : msg -> eh.onError(msg, MqErrorIdentity.RETRY_EXHAUSTED, null);
+
             MqSubscription sub = MqSubscription.builder()
                     .topic(meta.topic()).group(meta.group()).tag(meta.tag())
                     .payloadType(ep.getPayloadType())
                     .concurrency(meta.concurrency()).batchSize(meta.batchSize()).maxRetry(meta.maxRetry())
-                    .listener(listener).build();
+                    .listener(dispatcher::dispatch)
+                    .retryExhaustedHandler(retryExhausted)
+                    .build();
             driver.subscribe(sub);
         }
         running = true;
     }
 
+    private MqConsumeErrorHandler resolveErrorHandler(MqListener meta) {
+        if (errorHandlerResolver == null || meta.errorHandler().isEmpty()) return null;
+        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])。 */
+    private void consume(MqListenerEndpoint ep, List<MqMessage> batch) throws Exception {
+        if (ep.isBatch()) {
+            List<Object> payloads = new ArrayList<>(batch.size());
+            for (MqMessage m : batch) payloads.add(m.getPayload());
+            Object[] args = ep.isWantsContext()
+                    ? new Object[]{ payloads, new MqContext(batch.get(batch.size() - 1)) }
+                    : new Object[]{ payloads };
+            invokeReflect(ep, args);
+        } else {
+            for (MqMessage m : batch) invoke(ep, m);
+        }
+    }
+
     private void invoke(MqListenerEndpoint ep, MqMessage msg) throws Exception {
         Object[] args = ep.isWantsContext()
                 ? new Object[]{ msg.getPayload(), new MqContext(msg) }
                 : new Object[]{ msg.getPayload() };
+        invokeReflect(ep, args);
+    }
+
+    private void invokeReflect(MqListenerEndpoint ep, Object[] args) throws Exception {
         try { ep.getMethod().invoke(ep.getBean(), args); }
         catch (java.lang.reflect.InvocationTargetException e) {
             Throwable c = e.getCause();

mq-starter-core/src/main/java/com/njcn/mq/spi/MqConsumeErrorHandler.java

@@ -0,0 +1,18 @@
+package com.njcn.mq.spi;
+
+import com.njcn.mq.core.MqMessage;
+
+/**
+ * 消费异常落库回调 SPI。
+ *
+ * <p>业务实现（如调 feign 写错误日志表）；core 默认 no-op。对齐 legacy 的 saveExceptionMsgLog。
+ */
+public interface MqConsumeErrorHandler {
+
+    /**
+     * @param message   失败的消息
+     * @param identity  {@link MqErrorIdentity#SINGLE} 单次失败 / {@link MqErrorIdentity#RETRY_EXHAUSTED} 重试耗尽
+     * @param exception 失败异常（RETRY_EXHAUSTED 由 driver 触发时可能为 null）
+     */
+    void onError(MqMessage message, MqErrorIdentity identity, Exception exception);
+}

mq-starter-core/src/main/java/com/njcn/mq/spi/MqErrorIdentity.java

@@ -0,0 +1,9 @@
+package com.njcn.mq.spi;
+
+/** 消费异常落库的身份标识，对齐 legacy 的 IDENTITY_SINGLE / IDENTITY_RETRY。 */
+public enum MqErrorIdentity {
+    /** 单次消费失败。 */
+    SINGLE,
+    /** 重试耗尽（超过 maxRetry）。 */
+    RETRY_EXHAUSTED
+}

mq-starter-core/src/main/java/com/njcn/mq/spi/MqIdempotentStore.java

@@ -0,0 +1,24 @@
+package com.njcn.mq.spi;
+
+/**
+ * 消费去重 SPI（按消息 key 维度）。
+ *
+ * <p>状态机:processing / success / fail。语义对齐 legacy 的 filter+consumeSuccess+saveException,
+ * 但修正了 legacy 读写 key 不一致的缺陷(默认实现读写同一前缀键)。
+ */
+public interface MqIdempotentStore {
+
+    /**
+     * 尝试占用消费资格。
+     *
+     * @return true=可消费(并已标记 processing);false=应跳过(processing 中或已 success)。
+     *         无记录或上次 fail 时放行。
+     */
+    boolean tryAcquire(String key);
+
+    /** 消费成功,标记 success(较长 TTL,防重复消费)。 */
+    void markSuccess(String key);
+
+    /** 消费失败,标记 fail(允许后续重新消费)。 */
+    void markFail(String key);
+}

mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java

@@ -1,6 +1,8 @@
 package com.njcn.mq.spi;
+import com.njcn.mq.core.MqMessage;
 import lombok.Builder;
 import lombok.Data;
+import java.util.function.Consumer;
 
 @Data
 @Builder
@@ -13,4 +15,6 @@ public class MqSubscription {
     private int batchSize;
     private int maxRetry;
     private MqMessageListener listener;
+    /** 重试耗尽（超过 maxRetry）时由 driver 回调，用于落库 RETRY_EXHAUSTED；可空。 */
+    private Consumer<MqMessage> retryExhaustedHandler;
 }

mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqCoreAutoConfigurationTest.java

@@ -1,14 +1,19 @@
 package com.njcn.mq.autoconfig;
+import com.njcn.mq.annotation.MqListener;
 import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.core.MqTemplate;
 import com.njcn.mq.container.MqListenerRegistry;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqIdempotentStore;
 import com.njcn.mq.spi.MqSubscription;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import static org.assertj.core.api.Assertions.assertThat;
 
 class MqCoreAutoConfigurationTest {
@@ -32,4 +37,44 @@ class MqCoreAutoConfigurationTest {
             .withConfiguration(AutoConfigurations.of(MqCoreAutoConfiguration.class))
             .run(ctx -> assertThat(ctx).doesNotHaveBean(MqTemplate.class));
     }
+
+    static class InMemoryStore implements MqIdempotentStore {
+        final Map<String, String> m = new ConcurrentHashMap<>();
+        public boolean tryAcquire(String key) {
+            String s = m.get(key);
+            if (s == null || "fail".equals(s)) { m.put(key, "processing"); return true; }
+            return false;
+        }
+        public void markSuccess(String key) { m.put(key, "success"); }
+        public void markFail(String key) { m.put(key, "fail"); }
+    }
+    static class DedupListener {
+        final AtomicInteger count = new AtomicInteger();
+        @MqListener(topic = "D", group = "dg", idempotent = true)
+        public void on(String msg) { count.incrementAndGet(); }
+    }
+    @Configuration static class IdempotentSetup {
+        @Bean MqDriver driver() { return new MqDriver() {
+            public void send(String t, MqMessage m) {}
+            public void subscribe(MqSubscription s) {
+                try {
+                    s.getListener().onMessage(MqMessage.of("dup", "", "x"));
+                    s.getListener().onMessage(MqMessage.of("dup", "", "y"));
+                } catch (Exception e) { throw new RuntimeException(e); }
+            }
+            public void stop() {} }; }
+        @Bean MqIdempotentStore store() { return new InMemoryStore(); }
+        @Bean DedupListener dedupListener() { return new DedupListener(); }
+    }
+    @Test
+    void idempotentDedup_endToEnd_whenStoreBeanPresent() {
+        new ApplicationContextRunner()
+            .withConfiguration(AutoConfigurations.of(MqCoreAutoConfiguration.class))
+            .withUserConfiguration(IdempotentSetup.class)
+            .run(ctx -> {
+                assertThat(ctx).hasNotFailed();
+                DedupListener l = ctx.getBean(DedupListener.class);
+                assertThat(l.count.get()).isEqualTo(1);
+            });
+    }
 }

mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java

@@ -0,0 +1,169 @@
+package com.njcn.mq.container;
+
+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 org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+class MqConsumeDispatcherTest {
+
+    @Test
+    void singleMode_invokesConsumerWithMessage() throws Exception {
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "hello"));
+
+        assertEquals(1, consumed.size());
+        assertEquals("hello", consumed.get(0).get(0).getPayload());
+    }
+
+    @Test
+    void batchMode_flushesOnceWhenBufferReachesBatchSize() throws Exception {
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b"));
+        assertEquals(0, consumed.size(), "未满批不应消费");
+
+        d.dispatch(MqMessage.of("k3", "", "c"));
+        assertEquals(1, consumed.size(), "满批触发一次消费");
+        assertEquals(3, consumed.get(0).size());
+    }
+
+    @Test
+    void idempotent_skipsWhenStoreRejects() throws Exception {
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return !"dup".equals(key); }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("ok", "", "a"));
+        d.dispatch(MqMessage.of("dup", "", "b"));
+
+        assertEquals(1, consumed.size(), "被拒绝的 key 应跳过");
+        assertEquals("a", consumed.get(0).get(0).getPayload());
+    }
+
+    @Test
+    void idempotent_marksSuccessAfterConsume() throws Exception {
+        List<String> succeeded = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) { succeeded.add(key); }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(batch -> { })
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+
+        assertEquals(1, succeeded.size(), "消费成功后应标记 success");
+        assertEquals("k1", succeeded.get(0));
+    }
+
+    @Test
+    void singleMode_onException_callsErrorHandlerMarksFailAndRethrows() {
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        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); }
+        };
+        MqConsumeErrorHandler eh = (msg, identity, ex) -> errors.add(identity);
+        RuntimeException boom = new RuntimeException("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, "原异常应抛出交 driver 重投");
+        assertEquals(1, errors.size());
+        assertEquals(MqErrorIdentity.SINGLE, errors.get(0), "单次失败应落库 SINGLE");
+        assertEquals(1, failed.size());
+        assertEquals("k1", failed.get(0), "失败应标记 fail 允许重新消费");
+    }
+
+    @Test
+    void batchMode_onException_swallowsAndCallsErrorHandler() throws Exception {
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        MqConsumeErrorHandler eh = (msg, identity, ex) -> errors.add(identity);
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(2).errorHandler(eh)
+                .consumer(batch -> { throw new RuntimeException("db down"); })
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a")); // 攒批，未满
+        assertDoesNotThrow(() -> d.dispatch(MqMessage.of("k2", "", "b")),
+                "批量入库失败不应抛出（对齐 legacy 不卡队列）");
+        assertEquals(2, errors.size(), "批量失败应对每条落库");
+    }
+
+    @Test
+    void batchMode_marksSuccessForEachAfterFlush() throws Exception {
+        List<String> succeeded = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) { succeeded.add(key); }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(2).idempotent(true).store(store)
+                .consumer(batch -> { })
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b"));
+
+        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-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java

@@ -2,9 +2,19 @@ package com.njcn.mq.container;
 import com.njcn.mq.annotation.MqListener;
 import com.njcn.mq.core.MqContext;
 import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqConsumeErrorHandler;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqErrorIdentity;
+import com.njcn.mq.spi.MqIdempotentStore;
 import com.njcn.mq.spi.MqSubscription;
 import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -15,6 +25,33 @@ class MqListenerRegistryTest {
         @MqListener(topic = "LN_Topic", group = "ln_consumer")
         public void on(String payload, MqContext ctx) { got.set(payload); ctxKey.set(ctx.getKey()); }
     }
+    static class BatchBiz {
+        final List<String> received = new ArrayList<>();
+        @MqListener(topic = "B", group = "bg", batchSize = 2)
+        public void onBatch(List<String> msgs) { received.addAll(msgs); }
+    }
+    static class DedupBiz {
+        final AtomicInteger count = new AtomicInteger();
+        @MqListener(topic = "D", group = "dg", idempotent = true)
+        public void on(String msg) { count.incrementAndGet(); }
+    }
+    static class ErrBiz {
+        @MqListener(topic = "E", group = "eg", errorHandler = "myEH")
+        public void on(String msg) { throw new RuntimeException("biz fail"); }
+    }
+
+    /** 真实状态机的内存去重 store。 */
+    static class InMemoryStore implements MqIdempotentStore {
+        final Map<String, String> m = new ConcurrentHashMap<>();
+        @Override public boolean tryAcquire(String key) {
+            String s = m.get(key);
+            if (s == null || "fail".equals(s)) { m.put(key, "processing"); return true; }
+            return false;
+        }
+        @Override public void markSuccess(String key) { m.put(key, "success"); }
+        @Override public void markFail(String key) { m.put(key, "fail"); }
+    }
+
     /** fake driver：subscribe 时立刻把一条消息喂给 listener。 */
     static class ImmediateDriver implements MqDriver {
         @Override public void send(String topic, MqMessage m) {}
@@ -24,6 +61,24 @@ class MqListenerRegistryTest {
             catch (Exception e) { throw new RuntimeException(e); }
         }
     }
+    /** fake driver：捕获 subscription，不喂消息。 */
+    static class CaptureDriver implements MqDriver {
+        MqSubscription captured;
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() {}
+        @Override public void subscribe(MqSubscription s) { this.captured = s; }
+    }
+    /** fake driver：subscribe 时按序喂多条消息。 */
+    static class FeedDriver implements MqDriver {
+        private final MqMessage[] msgs;
+        FeedDriver(MqMessage... msgs) { this.msgs = msgs; }
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() {}
+        @Override public void subscribe(MqSubscription sub) {
+            try { for (MqMessage m : msgs) sub.getListener().onMessage(m); }
+            catch (Exception e) { throw new RuntimeException(e); }
+        }
+    }
 
     @Test
     void start_subscribesAndInvokesBusinessMethodWithPayloadAndContext() throws Exception {
@@ -35,4 +90,63 @@ class MqListenerRegistryTest {
         assertEquals("hi", biz.got.get());
         assertEquals("k9", biz.ctxKey.get());
     }
+
+    @Test
+    void batchListener_accumulatesAndInvokesWithPayloadList() {
+        BatchBiz biz = new BatchBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        MqListenerRegistry reg = new MqListenerRegistry(
+                new FeedDriver(MqMessage.of("k1", "", "a"), MqMessage.of("k2", "", "b")),
+                bpp.getEndpoints());
+        reg.start();
+        assertEquals(Arrays.asList("a", "b"), biz.received, "批量方法应收到攒满的 payload 列表");
+    }
+
+    @Test
+    void idempotentListener_dedupsSameKey() {
+        DedupBiz biz = new DedupBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        MqListenerRegistry reg = new MqListenerRegistry(
+                new FeedDriver(MqMessage.of("dupKey", "", "x"), MqMessage.of("dupKey", "", "y")),
+                bpp.getEndpoints(), new InMemoryStore(), null);
+        reg.start();
+        assertEquals(1, biz.count.get(), "同 key 重复消息应只消费一次");
+    }
+
+    @Test
+    void errorHandler_resolvedByNameAndCalledOnException() {
+        ErrBiz biz = new ErrBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        MqConsumeErrorHandler eh = (m, id, ex) -> errors.add(id);
+        Function<String, MqConsumeErrorHandler> resolver = name -> "myEH".equals(name) ? eh : null;
+        MqListenerRegistry reg = new MqListenerRegistry(
+                new FeedDriver(MqMessage.of("k1", "", "a")), bpp.getEndpoints(), null, resolver);
+
+        assertThrows(RuntimeException.class, reg::start, "单条异常应向 driver 抛出");
+        assertEquals(1, errors.size(), "按名解析的 errorHandler 应被调用");
+        assertEquals(MqErrorIdentity.SINGLE, errors.get(0));
+    }
+
+    @Test
+    void retryExhaustedHandler_wiredFromErrorHandler() {
+        ErrBiz biz = new ErrBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        MqConsumeErrorHandler eh = (m, id, ex) -> errors.add(id);
+        Function<String, MqConsumeErrorHandler> resolver = name -> "myEH".equals(name) ? eh : null;
+        CaptureDriver driver = new CaptureDriver();
+        MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints(), null, resolver);
+
+        reg.start();
+
+        assertNotNull(driver.captured.getRetryExhaustedHandler(), "应从 errorHandler 接线重试耗尽回调");
+        driver.captured.getRetryExhaustedHandler().accept(MqMessage.of("k", "", "p"));
+        assertEquals(1, errors.size());
+        assertEquals(MqErrorIdentity.RETRY_EXHAUSTED, errors.get(0), "重试耗尽应落库 RETRY_EXHAUSTED");
+    }
 }

mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java

@@ -13,6 +13,11 @@ class MqListenerScanTest {
         public void onWithCtx(String msg, MqContext ctx) { }
     }
 
+    static class BatchBiz {
+        @MqListener(topic = "B", group = "bg", batchSize = 10)
+        public void onBatch(List<String> msgs) { }
+    }
+
     @Test
     void scan_findsEndpoints_withPayloadTypeAndContextFlag() {
         MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
@@ -25,4 +30,13 @@ class MqListenerScanTest {
         MqListenerEndpoint e2 = eps.stream().filter(e -> e.getMeta().topic().equals("T2")).findFirst().orElseThrow(AssertionError::new);
         assertTrue(e2.isWantsContext());
     }
+
+    @Test
+    void scan_batchListener_resolvesElementTypeAndMarksBatch() {
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(new BatchBiz(), "biz");
+        MqListenerEndpoint ep = bpp.getEndpoints().get(0);
+        assertTrue(ep.isBatch(), "List 参数应识别为批量");
+        assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T");
+    }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java

@@ -0,0 +1,62 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.mq.spi.MqIdempotentStore;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.time.Duration;
+
+/**
+ * 基于 Redis 的去重默认实现。状态机 processing / success / fail，各带 TTL。
+ *
+ * <p>对齐 legacy filter+consumeSuccess+saveException 的语义，但修正了 legacy「读裸 key、写带前缀 key」
+ * 导致去重失效的缺陷：本实现读写统一使用 {@code prefix + key}。
+ */
+public class RedisMqIdempotentStore implements MqIdempotentStore {
+
+    private static final String PROCESSING = "processing";
+    private static final String SUCCESS = "success";
+    private static final String FAIL = "fail";
+
+    private final StringRedisTemplate redis;
+    private final String prefix;
+    private final long processingTtl;
+    private final long successTtl;
+    private final long failTtl;
+
+    public RedisMqIdempotentStore(StringRedisTemplate redis, String prefix,
+                                  long processingTtlSeconds, long successTtlSeconds, long failTtlSeconds) {
+        this.redis = redis;
+        this.prefix = prefix;
+        this.processingTtl = processingTtlSeconds;
+        this.successTtl = successTtlSeconds;
+        this.failTtl = failTtlSeconds;
+    }
+
+    @Override
+    public boolean tryAcquire(String key) {
+        String k = prefix + key;
+        // setIfAbsent 原子占用：key 不存在时写 processing 并放行。
+        Boolean acquired = redis.opsForValue().setIfAbsent(k, PROCESSING, Duration.ofSeconds(processingTtl));
+        if (Boolean.TRUE.equals(acquired)) {
+            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;
+        }
+        return false;
+    }
+
+    @Override
+    public void markSuccess(String key) {
+        redis.opsForValue().set(prefix + key, SUCCESS, Duration.ofSeconds(successTtl));
+    }
+
+    @Override
+    public void markFail(String key) {
+        redis.opsForValue().set(prefix + key, FAIL, Duration.ofSeconds(failTtl));
+    }
+}

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -9,6 +9,9 @@ import com.njcn.mq.spi.MqDriver;
 import com.njcn.mq.spi.MqSubscription;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.redis.connection.stream.*;
+import org.springframework.data.domain.Range;
+import org.springframework.data.redis.connection.stream.PendingMessage;
+import org.springframework.data.redis.connection.stream.PendingMessages;
 import org.springframework.data.redis.core.RedisCallback;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import java.nio.charset.StandardCharsets;
@@ -20,11 +23,19 @@ public class RedisStreamMqDriver implements MqDriver {
     private final StringRedisTemplate redis;
     private final StreamKeyBuilder keyBuilder;
     private final RedisStreamProperties props;
+    private final long reclaimMinIdleMs;
     private final List<Thread> workers = new ArrayList<>();
     private volatile boolean running = true;
 
     public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
+        // 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) {
         this.redis = redis; this.keyBuilder = keyBuilder; this.props = props;
+        this.reclaimMinIdleMs = reclaimMinIdleMs;
     }
 
     @Override public void send(String topic, MqMessage message) {
@@ -54,6 +65,7 @@ public class RedisStreamMqDriver implements MqDriver {
         int maxRetry = sub.getMaxRetry() > 0 ? sub.getMaxRetry() : props.getMaxRetry();
         while (running && !Thread.currentThread().isInterrupted()) {
             try {
+                reclaimPending(sub, stream, group, consumer, maxRetry);
                 List<MapRecord<String, Object, Object>> recs = redis.opsForStream().read(
                         Consumer.from(group, consumer),
                         StreamReadOptions.empty().count(props.getReadCount()).block(Duration.ofMillis(props.getBlockMs())),
@@ -88,10 +100,66 @@ public class RedisStreamMqDriver implements MqDriver {
             return true;
         } catch (Exception e) {
             log.warn("[mq-redis] handle failed key={}, leave PEL for retry (max={})", msg.getKey(), maxRetry, e);
-            return false; // 留 PEL；超 maxRetry 的兜底由后续 autoclaim 增强补（轻量版先靠重投）
+            return false; // 留 PEL，由 reclaimPending 重投或落库
         }
     }
 
+    /**
+     * 处理 PEL 中超过 {@code reclaimMinIdleMs} 仍未 ACK 的消息：
+     * delivery 次数未超 maxRetry 则重投消费；超过则回调 retryExhaustedHandler 落库并 ACK 终止。
+     */
+    private void reclaimPending(MqSubscription sub, String stream, String group, String consumer, int maxRetry) {
+        PendingMessages pendings;
+        try {
+            pendings = redis.opsForStream().pending(stream, group, Range.unbounded(), props.getReadCount());
+        } catch (Exception e) {
+            return; // group 尚未建立或暂无 PEL，忽略
+        }
+        if (pendings == null || pendings.isEmpty()) return;
+        Duration minIdle = Duration.ofMillis(reclaimMinIdleMs);
+        for (PendingMessage pm : pendings) {
+            if (pm.getElapsedTimeSinceLastDelivery().toMillis() < reclaimMinIdleMs) continue;
+            long deliveryCount = pm.getTotalDeliveryCount();
+            List<ByteRecord> claimed;
+            try {
+                // StreamOperations(2.3) 无 claim，走 connection 层 xClaim：转移所有权并使 delivery 计数 +1。
+                claimed = redis.execute((RedisCallback<List<ByteRecord>>) conn ->
+                        conn.streamCommands().xClaim(stream.getBytes(StandardCharsets.UTF_8), group, consumer, minIdle, pm.getId()));
+            } catch (Exception e) { continue; }
+            if (claimed == null || claimed.isEmpty()) continue; // 被其它消费者抢先 claim 或消息已删
+            ByteRecord rec = claimed.get(0);
+            Map<String, String> fields = new HashMap<>();
+            rec.getValue().forEach((k, v) -> fields.put(
+                    new String(k, StandardCharsets.UTF_8), new String(v, StandardCharsets.UTF_8)));
+            if (deliveryCount > maxRetry) {
+                if (sub.getRetryExhaustedHandler() != null) {
+                    try {
+                        sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields));
+                    } catch (Exception ex) {
+                        log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex);
+                    }
+                }
+                redis.opsForStream().acknowledge(stream, group, rec.getId());
+                log.warn("[mq-redis] retry exhausted, acked key={} delivery={}", fields.get(StreamMessageConstant.F_KEY), deliveryCount);
+            } else {
+                boolean ack = dispatch(sub, fields, maxRetry);
+                if (ack) redis.opsForStream().acknowledge(stream, group, rec.getId());
+            }
+        }
+    }
+
+    /** 解码消息体为业务 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/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriverAutoConfiguration.java

@@ -3,6 +3,7 @@ import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
 import com.njcn.middle.stream.support.StreamKeyBuilder;
 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
 import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqIdempotentStore;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
@@ -21,4 +22,11 @@ public class RedisStreamMqDriverAutoConfiguration {
     public MqDriver mqDriver(StringRedisTemplate redis, StreamKeyBuilder streamKeyBuilder, RedisStreamProperties props) {
         return new RedisStreamMqDriver(redis, streamKeyBuilder, props);
     }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqIdempotentStore mqIdempotentStore(StringRedisTemplate redis) {
+        // 默认 TTL：processing 60s / success 300s / fail 300s，对齐 legacy 量级；统一前缀（修正 key 一致性）。
+        return new RedisMqIdempotentStore(redis, "mq:idem:", 60, 300, 300);
+    }
 }

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisMqIdempotentStoreIT.java

@@ -0,0 +1,38 @@
+package com.njcn.mq.driver.redis;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class RedisMqIdempotentStoreIT {
+
+    @Test
+    void stateMachine_firstAcquire_thenBlocked_failReopens_successBlocks() {
+        StringRedisTemplate redis = newRedisOrSkip();
+        RedisMqIdempotentStore store = new RedisMqIdempotentStore(redis, "mq:idem:it:", 60, 300, 300);
+        String key = "k_" + System.nanoTime();
+
+        assertTrue(store.tryAcquire(key), "首次应放行并占用 processing");
+        assertFalse(store.tryAcquire(key), "processing 中应挡回");
+
+        store.markFail(key);
+        assertTrue(store.tryAcquire(key), "上次 fail 后应允许重新消费");
+
+        store.markSuccess(key);
+        assertFalse(store.tryAcquire(key), "success 后应挡回，避免重复消费");
+    }
+
+    private static StringRedisTemplate newRedisOrSkip() {
+        try {
+            LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);
+            cf.afterPropertiesSet();
+            StringRedisTemplate t = new StringRedisTemplate(cf);
+            t.afterPropertiesSet();
+            t.hasKey("ping");
+            return t;
+        } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; }
+    }
+}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java

@@ -0,0 +1,67 @@
+package com.njcn.mq.driver.redis;
+
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqSubscription;
+import com.njcn.middle.stream.autoconfig.RedisStreamProperties;
+import com.njcn.middle.stream.support.StreamKeyBuilder;
+import org.junit.jupiter.api.Test;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
+import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
+
+class RedisStreamMqDriverRetryIT {
+
+    @Test
+    void retryExhausted_afterMaxRetry_callsHandlerAndAcks() throws Exception {
+        StringRedisTemplate redis = newRedisOrSkip();
+        StreamKeyBuilder kb = new StreamKeyBuilder(false, "");
+        RedisStreamProperties props = new RedisStreamProperties();
+        props.setConsumerName("it-retry");
+        props.setReadCount(10);
+        props.setBlockMs(300);
+        props.setMaxRetry(2);
+        RedisStreamMqDriver driver = new RedisStreamMqDriver(redis, kb, props, 200L);
+
+        String topic = "MQ_RETRY_IT_" + System.nanoTime();
+        AtomicInteger attempts = new AtomicInteger();
+        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(exhausted::add)
+                .build());
+
+        driver.send(topic, MqMessage.of("rk1", "", "data"));
+
+        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);
+        PendingMessagesSummary summary = redis.opsForStream().pending(kb.buildStream(topic), "g1");
+        assertEquals(0L, summary.getTotalPendingMessages(), "耗尽后应 ACK，PEL 清空");
+        driver.stop();
+    }
+
+    private static StringRedisTemplate newRedisOrSkip() {
+        try {
+            LettuceConnectionFactory cf = new LettuceConnectionFactory("localhost", 6379);
+            cf.afterPropertiesSet();
+            StringRedisTemplate t = new StringRedisTemplate(cf);
+            t.afterPropertiesSet();
+            t.hasKey("ping");
+            return t;
+        } catch (Exception e) { assumeTrue(false, "no local redis, skip"); return null; }
+    }
+}

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverWiringTest.java

@@ -1,8 +1,13 @@
 package com.njcn.mq.driver.redis;
 
 import com.njcn.mq.autoconfig.MqCoreAutoConfiguration;
+import com.njcn.mq.spi.MqIdempotentStore;
 import org.junit.jupiter.api.Test;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.context.annotation.Bean;
+import org.springframework.data.redis.core.StringRedisTemplate;
+
+import java.lang.reflect.Method;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -30,4 +35,16 @@ class RedisStreamMqDriverWiringTest {
                 .as("@AutoConfigureBefore 必须指向 MqCoreAutoConfiguration")
                 .contains(MqCoreAutoConfiguration.class);
     }
+
+    @Test
+    void declaresIdempotentStoreBean() throws Exception {
+        Method m = RedisStreamMqDriverAutoConfiguration.class
+                .getDeclaredMethod("mqIdempotentStore", StringRedisTemplate.class);
+        assertThat(m.getAnnotation(Bean.class))
+                .as("redis-stream 模式应提供默认去重实现")
+                .isNotNull();
+        assertThat(MqIdempotentStore.class)
+                .as("默认去重实现须为 MqIdempotentStore 类型")
+                .isAssignableFrom(m.getReturnType());
+    }
 }