feat(mq-starter-core): 消费侧通用化 去重/批量/异常落库（注解+SPI 织入）

- 新增 MqConsumeDispatcher 消费处理链：去重→(攒批)→业务→成功标记/异常落库
- 新增 SPI：MqIdempotentStore（去重，修正 legacy 读写 key 不一致缺陷）、
  MqConsumeErrorHandler + MqErrorIdentity（异常落库 SINGLE/RETRY_EXHAUSTED）
- @MqListener 扩展 idempotent/errorHandler；BPP 识别 List<T> 批量签名并解析元素类型
- MqListenerRegistry 织入 dispatcher，按注解装配 store/errorHandler；
  autoconfig 经 ObjectProvider/ApplicationContext 注入（4 参构造，保留 2 参重载）
- 重试统一交 driver MQ 原生重投（单条异常 rethrow）；批量入库失败落库后吞掉不卡队列
- 19 个单测全绿（-DforkCount=0）

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

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,99 @@
+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 java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。
+ * 纯逻辑、不依赖 spring / redis，便于单测。重试本身交给 driver 的 MQ 原生重投。
+ */
+public class MqConsumeDispatcher {
+
+    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) {
+                if (errorHandler != null) {
+                    errorHandler.onError(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 标记自然过期）。
+                if (errorHandler != null) {
+                    for (MqMessage m : flush) {
+                        errorHandler.onError(m, MqErrorIdentity.SINGLE, e);
+                    }
+                }
+                return;
+            }
+            if (idempotent && store != null) {
+                for (MqMessage m : flush) {
+                    store.markSuccess(m.getKey());
+                }
+            }
+        }
+    }
+
+    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,82 @@ 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.MqIdempotentStore;
 import com.njcn.mq.spi.MqMessageListener;
 import com.njcn.mq.spi.MqSubscription;
 import org.springframework.context.SmartLifecycle;
+import java.util.ArrayList;
 import java.util.List;
 
 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);
+            MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder()
+                    .batchSize(meta.batchSize())
+                    .consumer(batch -> consume(ep, batch));
+            if (meta.idempotent()) {
+                db.idempotent(true).store(store);
+            }
+            if (errorHandlerResolver != null && !meta.errorHandler().isEmpty()) {
+                MqConsumeErrorHandler eh = errorHandlerResolver.apply(meta.errorHandler());
+                if (eh != null) {
+                    db.errorHandler(eh);
+                }
+            }
+            MqConsumeDispatcher dispatcher = db.build();
             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).build();
             driver.subscribe(sub);
         }
         running = true;
     }
 
+    /** 业务调用适配：批量 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/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,145 @@
+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");
+    }
+}

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,17 @@ class MqListenerRegistryTest {
             catch (Exception e) { throw new RuntimeException(e); }
         }
     }
+    /** 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 +83,44 @@ 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));
+    }
 }

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");
+    }
 }