fix(mq-starter-core): 终审两处停机缝隙——dispatcher 打烊拒收 + 定时器温和关闭

1) forceFlush 置 closed:driver.stop() 异步返回后残余线程再投递的消息一律
   拒收抛 InProgress(不 ACK、下次启动重投),堵住正常停机丢最后一轮消息的缝隙;
   检查先于去重,不留 processing 标记。
2) 定时器 shutdownNow→shutdown:不再打断正在执行的刷出,避免整批被误判失败;
   后续执行照常取消,残留批由 forceFlush 接住(flushLock 串行兜底)。

终审 Important #1/#2,处置方案经用户确认。

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

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

@@ -34,6 +34,9 @@ public class MqConsumeDispatcher {
     private final List<MqMessage> buffer = new ArrayList<>();
     /** 串行化三条刷出路径（满批/超时/停机）的业务调用段：业务批量方法任一时刻至多一个调用在跑。 */
     private final Object flushLock = new Object();
+    /** 停机打烊标志:forceFlush 置位后 dispatch 一律拒收抛 InProgress——driver 不 ACK、下次启动重投,
+     *  堵住"driver.stop() 异步返回后残余线程仍入缓冲"的丢消息缝隙。volatile 供 driver 线程快速路径读取。 */
+    private volatile boolean closed = false;
 
     private MqConsumeDispatcher(Builder b) {
         this.batchSize = b.batchSize;
@@ -46,6 +49,10 @@ public class MqConsumeDispatcher {
     }
 
     public void dispatch(MqMessage msg) throws Exception {
+        if (closed) {
+            // 已打烊:抛 InProgress 信号让 driver 不 ACK,消息留在 MQ 下次启动重投(不碰去重标记)。
+            throw new MqMessageInProgressException(msg.getKey());
+        }
         if (idempotent && store != null && !store.tryAcquire(msg.getKey())) {
             // 跳过分两类，决定 driver 能否 ACK：
             //  - 已 success：正常返回 → driver ACK，挡掉重复投递（正常去重）；
@@ -77,6 +84,9 @@ public class MqConsumeDispatcher {
         // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。
         List<MqMessage> flush = null;
         synchronized (buffer) {
+            if (closed) {
+                throw new MqMessageInProgressException(msg.getKey());
+            }
             if (buffer.isEmpty()) {
                 batchStartAtMs = clock.getAsLong();
             }
@@ -131,10 +141,11 @@ public class MqConsumeDispatcher {
         }
     }
 
-    /** 停机强制刷出残留批（非空即刷，不看超时）。 */
+    /** 停机强制刷出残留批并打烊(非空即刷,不看超时;打烊后 dispatch 一律拒收,见 closed 字段)。 */
     public void forceFlush() {
         List<MqMessage> flush = null;
         synchronized (buffer) {
+            closed = true;
             if (!buffer.isEmpty()) {
                 flush = new ArrayList<>(buffer);
                 buffer.clear();

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

@@ -148,7 +148,8 @@ public class MqListenerRegistry implements SmartLifecycle {
         // 停机顺序：driver 先停（不再有新消息入缓冲）→ 定时器停 → 强刷残留批 → 置 running。
         driver.stop();
         if (flushScheduler != null) {
-            flushScheduler.shutdownNow();
+            // 温和关闭(不打断正在执行的最后一次刷出;后续执行取消,残留批由下方 forceFlush 接住)。
+            flushScheduler.shutdown();
             flushScheduler = null;
         }
         for (MqConsumeDispatcher d : dispatchers) {

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

@@ -4,6 +4,7 @@ import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqConsumeErrorHandler;
 import com.njcn.mq.spi.MqErrorIdentity;
 import com.njcn.mq.spi.MqIdempotentStore;
+import com.njcn.mq.spi.MqMessageInProgressException;
 import org.junit.jupiter.api.Test;
 
 import java.util.ArrayList;
@@ -395,4 +396,43 @@ class MqConsumeDispatcherTest {
         for (List<MqMessage> batch : consumed) for (MqMessage m : batch) keys.add(m.getKey());
         assertEquals(99, keys.size(), "99 个 key 应全部恰好出现一次");
     }
+
+    @Test
+    void forceFlush_thenDispatch_rejectsSoDriverWontAck() 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(), "残留批应被停机强刷刷出");
+
+        assertThrows(MqMessageInProgressException.class,
+                () -> d.dispatch(MqMessage.of("k2", "", "b")),
+                "打烊后迟到的消息应拒收抛 InProgress,driver 不 ACK、下次启动重投");
+        d.forceFlush();
+        assertEquals(1, consumed.size(), "拒收的消息不得进入缓冲(再次强刷无动作)");
+    }
+
+    @Test
+    void closed_rejectsSingleModeToo_beforeTouchingDedupStore() {
+        List<String> acquired = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { acquired.add(key); return true; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store)
+                .consumer(batch -> { })
+                .build();
+
+        d.forceFlush(); // 空缓冲仅打烊
+        assertThrows(MqMessageInProgressException.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")),
+                "打烊后单条路径同样拒收");
+        assertEquals(0, acquired.size(), "拒收须发生在去重之前,不得留下 processing 标记");
+    }
 }