feat(mq-starter-core): MqConsumeDispatcher 批量超时/停机强制刷出(idle flush)

从本批首条进入缓冲起算(linger 语义),flushIfTimeout 到点刷未满批、
forceFlush 停机非空即刷,均汇入 doFlush;时钟经 Builder 注入,单测用
假时钟全覆盖(不真等待)。Builder 缺省 batchTimeoutMs=0 保持旧行为。

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

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

@@ -10,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.function.LongSupplier;
 
 /**
  * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。
@@ -18,8 +19,14 @@ import java.util.List;
 @Slf4j
 public class MqConsumeDispatcher {
 
-    // >1 时攒批；注意：无 idle flush，低流量下未满批会停滞、进程重启丢未入库批（与 legacy 一致，适合高流量流）。
+    // >1 时攒批；满批 / 超时(batchTimeoutMs) / 停机三路径刷出。kill -9 仍丢内存未刷批（上界 = 一个超时窗口内的量）。
     private final int batchSize;
+    /** 批量超时刷出阈值（毫秒，从本批第一条进入缓冲起算）；<=0 表示关闭超时刷出。 */
+    private final int batchTimeoutMs;
+    /** 当前批第一条进入缓冲的时刻（buffer 锁保护；缓冲清空后由下一条进入时重新记录）。 */
+    private long batchStartAtMs;
+    /** 时钟（Builder 可注入，单测用假时钟免真等待）。 */
+    private final LongSupplier clock;
     private final boolean idempotent;
     private final MqIdempotentStore store;
     private final MqConsumeErrorHandler errorHandler;
@@ -30,6 +37,8 @@ public class MqConsumeDispatcher {
 
     private MqConsumeDispatcher(Builder b) {
         this.batchSize = b.batchSize;
+        this.batchTimeoutMs = b.batchTimeoutMs;
+        this.clock = b.clock;
         this.idempotent = b.idempotent;
         this.store = b.store;
         this.errorHandler = b.errorHandler;
@@ -68,6 +77,9 @@ public class MqConsumeDispatcher {
         // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。
         List<MqMessage> flush = null;
         synchronized (buffer) {
+            if (buffer.isEmpty()) {
+                batchStartAtMs = clock.getAsLong();
+            }
             buffer.add(msg);
             if (buffer.size() >= batchSize) {
                 flush = new ArrayList<>(buffer);
@@ -104,6 +116,40 @@ public class MqConsumeDispatcher {
         }
     }
 
+    /** 供定时线程调用：本批首条进入缓冲已满 batchTimeoutMs 则强制刷出未满批；未到点 / 空缓冲 / 未启用超时直接返回。 */
+    public void flushIfTimeout() {
+        if (batchTimeoutMs <= 0) return;
+        List<MqMessage> flush = null;
+        synchronized (buffer) {
+            if (!buffer.isEmpty() && clock.getAsLong() - batchStartAtMs >= batchTimeoutMs) {
+                flush = new ArrayList<>(buffer);
+                buffer.clear();
+            }
+        }
+        if (flush != null) {
+            doFlush(flush);
+        }
+    }
+
+    /** 停机强制刷出残留批（非空即刷，不看超时）。 */
+    public void forceFlush() {
+        List<MqMessage> flush = null;
+        synchronized (buffer) {
+            if (!buffer.isEmpty()) {
+                flush = new ArrayList<>(buffer);
+                buffer.clear();
+            }
+        }
+        if (flush != null) {
+            doFlush(flush);
+        }
+    }
+
+    /** registry 据此决定是否把本 dispatcher 纳入定时超时扫描。 */
+    public boolean needsTimeoutFlush() {
+        return batchSize > 1 && batchTimeoutMs > 0;
+    }
+
     /** 调用异常落库回调，并隔离其自身异常（落库是 best-effort，不得掩盖/中断主流程）。 */
     private void safeOnError(MqMessage msg, MqErrorIdentity identity, Exception cause) {
         if (errorHandler == null) return;
@@ -118,11 +164,15 @@ public class MqConsumeDispatcher {
 
     public static class Builder {
         private int batchSize = 1;
+        private int batchTimeoutMs;                              // 缺省 0 = 关闭（注解缺省 60s 由 registry 传入）
+        private LongSupplier clock = System::currentTimeMillis;
         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 batchTimeoutMs(int ms) { this.batchTimeoutMs = ms; return this; }
+        public Builder clock(LongSupplier c) { this.clock = c; 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; }

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

@@ -7,7 +7,10 @@ import com.njcn.mq.spi.MqIdempotentStore;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Set;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -210,4 +213,186 @@ class MqConsumeDispatcherTest {
         assertEquals(1, failed.size(),
                 "errorHandler 抛异常时仍须 markFail，否则去重标记停在 processing 会导致 reclaim 时被去重挡掉而丢消息");
     }
+
+    @Test
+    void timeoutFlush_notDue_doesNothing() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        now[0] = 59_999;
+        d.flushIfTimeout();
+
+        assertEquals(0, consumed.size(), "未到超时不应刷出");
+    }
+
+    @Test
+    void timeoutFlush_due_flushesPartialBatchAndMarksSuccess() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        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(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b"));
+        now[0] = 60_000;
+        d.flushIfTimeout();
+
+        assertEquals(1, consumed.size(), "到点应刷出未满批");
+        assertEquals(2, consumed.get(0).size());
+        assertEquals(Arrays.asList("k1", "k2"), succeeded, "超时刷出成功后应逐条 markSuccess");
+    }
+
+    @Test
+    void timeoutFlush_onException_swallowsAndFilesSinglePerMessage() throws Exception {
+        long[] now = {0};
+        List<MqErrorIdentity> errors = new ArrayList<>();
+        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) { }
+        };
+        MqConsumeErrorHandler eh = (msg, identity, ex) -> errors.add(identity);
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .idempotent(true).store(store).errorHandler(eh)
+                .consumer(batch -> { throw new RuntimeException("db down"); })
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b"));
+        now[0] = 60_000;
+        assertDoesNotThrow(d::flushIfTimeout, "超时刷出失败应吞掉（消息已 ACK 无法重投）");
+
+        assertEquals(2, errors.size(), "失败应逐条落库 SINGLE");
+        assertEquals(MqErrorIdentity.SINGLE, errors.get(0));
+        assertEquals(0, succeeded.size(), "失败不得 markSuccess");
+    }
+
+    @Test
+    void fullBatchFlush_resetsTimer_noDoubleFlush() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(2).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b")); // 满批刷出
+        assertEquals(1, consumed.size());
+
+        now[0] = 120_000;
+        d.flushIfTimeout();
+        assertEquals(1, consumed.size(), "满批刷出后缓冲已空，超时扫描不应重复刷");
+    }
+
+    @Test
+    void emptyBuffer_timeoutAndForceFlush_doNothing() {
+        long[] now = {999_999};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.flushIfTimeout();
+        d.forceFlush();
+
+        assertEquals(0, consumed.size(), "空缓冲任何刷出都应是无动作");
+    }
+
+    @Test
+    void timeoutDisabled_neverTimeoutFlushes() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(0).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        now[0] = Long.MAX_VALUE / 2;
+        d.flushIfTimeout();
+
+        assertEquals(0, consumed.size(), "batchTimeoutMs=0 应关闭超时刷出");
+        assertFalse(d.needsTimeoutFlush(), "关闭超时的批量监听器不应纳入定时扫描");
+    }
+
+    @Test
+    void forceFlush_flushesLeftovers() throws Exception {
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000)
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.forceFlush();
+
+        assertEquals(1, consumed.size(), "停机强刷应刷出残留批");
+        assertEquals(1, consumed.get(0).size());
+    }
+
+    @Test
+    void timeoutFlush_newMessageAfterFlush_restartsTimer() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(3).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        now[0] = 60_000;
+        d.flushIfTimeout();                       // 第一批超时刷出
+        assertEquals(1, consumed.size());
+
+        now[0] = 61_000;
+        d.dispatch(MqMessage.of("k2", "", "b"));  // 新批从 61s 重新计时
+        now[0] = 120_000;
+        d.flushIfTimeout();
+        assertEquals(1, consumed.size(), "新批仅 59s 未到超时，不应刷出");
+
+        now[0] = 121_000;
+        d.flushIfTimeout();
+        assertEquals(2, consumed.size(), "新批满 60s 应刷出");
+    }
+
+    @Test
+    void timeoutThenFullBatch_noDupNoLoss() throws Exception {
+        long[] now = {0};
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(50).batchTimeoutMs(60_000).clock(() -> now[0])
+                .consumer(consumed::add)
+                .build();
+
+        for (int i = 0; i < 49; i++) d.dispatch(MqMessage.of("t" + i, "", "x"));
+        now[0] = 60_000;
+        d.flushIfTimeout();                       // 超时刷出 49 条
+
+        for (int i = 0; i < 50; i++) d.dispatch(MqMessage.of("f" + i, "", "y")); // 满批刷出 50 条
+
+        assertEquals(2, consumed.size());
+        assertEquals(49, consumed.get(0).size(), "超时批应恰好 49 条");
+        assertEquals(50, consumed.get(1).size(), "满批应恰好 50 条，无重复无丢失");
+        Set<String> keys = new HashSet<>();
+        for (List<MqMessage> batch : consumed) for (MqMessage m : batch) keys.add(m.getKey());
+        assertEquals(99, keys.size(), "99 个 key 应全部恰好出现一次");
+    }
 }