feat(mq-starter-core): MqListenerRegistry 接线批量超时刷出定时器 + 停机强刷残留批

有"批量且 batchTimeoutMs>0"监听器才建全局单线程 daemon(mq-batch-flush)
每 1s 扫描 flushIfTimeout,任务体逐个 catch-all 防调度线程死亡;stop 顺序
改为 driver 停→定时器停→forceFlush 残留→running=false,正常停机不丢未满批。
两 driver 零改动,redis-stream/rocketmq 同时受益。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -12,6 +12,9 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.SmartLifecycle;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class MqListenerRegistry implements SmartLifecycle {
@@ -20,6 +23,9 @@ public class MqListenerRegistry implements SmartLifecycle {
     private final MqIdempotentStore store;
     private final java.util.function.Function<String, MqConsumeErrorHandler> errorHandlerResolver;
     private volatile boolean running = false;
+    private final List<MqConsumeDispatcher> dispatchers = new ArrayList<>();
+    /** 批量超时刷出定时器：仅存在"批量且 batchTimeoutMs>0"监听器时创建（全应用一条单线程 daemon）。 */
+    private ScheduledExecutorService flushScheduler;
 
     public MqListenerRegistry(MqDriver driver, List<MqListenerEndpoint> endpoints) {
         this(driver, endpoints, null, null);
@@ -44,6 +50,7 @@ public class MqListenerRegistry implements SmartLifecycle {
 
             MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder()
                     .batchSize(meta.batchSize())
+                    .batchTimeoutMs(meta.batchTimeoutMs())
                     .consumer(batch -> consume(ep, batch));
             if (meta.idempotent()) {
                 db.idempotent(true).store(store);
@@ -52,6 +59,7 @@ public class MqListenerRegistry implements SmartLifecycle {
                 db.errorHandler(eh);
             }
             MqConsumeDispatcher dispatcher = db.build();
+            dispatchers.add(dispatcher);
 
             java.util.function.Consumer<MqMessage> retryExhausted = eh == null ? null
                     : msg -> eh.onError(msg, MqErrorIdentity.RETRY_EXHAUSTED, null);
@@ -65,6 +73,7 @@ public class MqListenerRegistry implements SmartLifecycle {
                     .build();
             driver.subscribe(sub);
         }
+        startFlushScheduler();
         running = true;
     }
 
@@ -108,7 +117,50 @@ public class MqListenerRegistry implements SmartLifecycle {
         }
     }
 
-    @Override public synchronized void stop() { running = false; driver.stop(); }
+    /** 有需要超时刷出的批量监听器才起定时线程；每 1s 扫一遍（触发最多晚 1s，对 60s 级超时无感）。 */
+    private void startFlushScheduler() {
+        List<MqConsumeDispatcher> timed = new ArrayList<>();
+        for (MqConsumeDispatcher d : dispatchers) {
+            if (d.needsTimeoutFlush()) timed.add(d);
+        }
+        if (timed.isEmpty()) return;
+        flushScheduler = Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "mq-batch-flush");
+            t.setDaemon(true);
+            return t;
+        });
+        flushScheduler.scheduleWithFixedDelay(() -> {
+            // 任务体逐个 catch-all：异常不得逃逸，否则调度器取消后续执行，所有监听器的超时刷出静默失效。
+            for (MqConsumeDispatcher d : timed) {
+                try {
+                    d.flushIfTimeout();
+                } catch (Exception e) {
+                    log.error("[mq] 批量超时刷出异常（跳过本轮，不影响其他监听器）", e);
+                }
+            }
+        }, 1, 1, TimeUnit.SECONDS);
+    }
+
+    /** 包内可见，供单测断言定时器生命周期。 */
+    ScheduledExecutorService getFlushScheduler() { return flushScheduler; }
+
+    @Override public synchronized void stop() {
+        // 停机顺序：driver 先停（不再有新消息入缓冲）→ 定时器停 → 强刷残留批 → 置 running。
+        driver.stop();
+        if (flushScheduler != null) {
+            flushScheduler.shutdownNow();
+            flushScheduler = null;
+        }
+        for (MqConsumeDispatcher d : dispatchers) {
+            try {
+                d.forceFlush();
+            } catch (Exception e) {
+                log.error("[mq] 停机强制刷出失败（吞掉，不阻塞停机）", e);
+            }
+        }
+        dispatchers.clear();
+        running = false;
+    }
     @Override public boolean isRunning() { return running; }
     @Override public boolean isAutoStartup() { return true; }
     @Override public int getPhase() { return Integer.MAX_VALUE - 100; }

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

@@ -149,4 +149,74 @@ class MqListenerRegistryTest {
         assertEquals(1, errors.size());
         assertEquals(MqErrorIdentity.RETRY_EXHAUSTED, errors.get(0), "重试耗尽应落库 RETRY_EXHAUSTED");
     }
+
+    static class NoTimeoutBatchBiz {
+        @MqListener(topic = "B0", group = "bg0", batchSize = 2, batchTimeoutMs = 0)
+        public void onBatch(List<String> msgs) { }
+    }
+    static class LeftoverBiz {
+        final List<String> events;
+        LeftoverBiz(List<String> events) { this.events = events; }
+        @MqListener(topic = "B", group = "bg", batchSize = 3)
+        public void onBatch(List<String> msgs) { events.add("flush:" + msgs.size()); }
+    }
+    /** fake driver：捕获 subscription，stop 时记录事件序。 */
+    static class OrderedDriver implements MqDriver {
+        final List<String> events;
+        MqSubscription captured;
+        OrderedDriver(List<String> events) { this.events = events; }
+        @Override public void send(String topic, MqMessage m) {}
+        @Override public void stop() { events.add("driverStop"); }
+        @Override public void subscribe(MqSubscription s) { this.captured = s; }
+    }
+
+    @Test
+    void start_withBatchListener_createsFlushScheduler() {
+        BatchBiz biz = new BatchBiz();
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        MqListenerRegistry reg = new MqListenerRegistry(new CaptureDriver(), bpp.getEndpoints());
+        reg.start();
+        try {
+            assertNotNull(reg.getFlushScheduler(), "有批量监听器应创建超时刷出定时器");
+        } finally {
+            reg.stop();
+        }
+    }
+
+    @Test
+    void start_withoutTimedBatchListener_noScheduler() {
+        Biz single = new Biz();
+        MqListenerAnnotationBeanPostProcessor bpp1 = new MqListenerAnnotationBeanPostProcessor();
+        bpp1.postProcessAfterInitialization(single, "biz");
+        MqListenerRegistry reg1 = new MqListenerRegistry(new CaptureDriver(), bpp1.getEndpoints());
+        reg1.start();
+        assertNull(reg1.getFlushScheduler(), "无批量监听器不应创建定时器");
+
+        NoTimeoutBatchBiz off = new NoTimeoutBatchBiz();
+        MqListenerAnnotationBeanPostProcessor bpp2 = new MqListenerAnnotationBeanPostProcessor();
+        bpp2.postProcessAfterInitialization(off, "biz");
+        MqListenerRegistry reg2 = new MqListenerRegistry(new CaptureDriver(), bpp2.getEndpoints());
+        reg2.start();
+        assertNull(reg2.getFlushScheduler(), "批量但 batchTimeoutMs=0 也不应创建定时器");
+    }
+
+    @Test
+    void stop_flushesLeftoverBatchAfterDriverStopped() throws Exception {
+        List<String> events = new ArrayList<>();
+        LeftoverBiz biz = new LeftoverBiz(events);
+        MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor();
+        bpp.postProcessAfterInitialization(biz, "biz");
+        OrderedDriver driver = new OrderedDriver(events);
+        MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints());
+        reg.start();
+        driver.captured.getListener().onMessage(MqMessage.of("k1", "", "a"));
+        driver.captured.getListener().onMessage(MqMessage.of("k2", "", "b")); // batchSize=3，2 条不满批
+
+        reg.stop();
+
+        assertEquals(Arrays.asList("driverStop", "flush:2"), events,
+                "停机应先停 driver 再强刷残留批（2 条未满批不丢）");
+        assertNull(reg.getFlushScheduler(), "停机后定时器应已关闭置空");
+    }
 }