refactor(mq-starter-core): 抽取 doFlush 统一批量刷出路径 + flushLock 串行化业务调用

为 idle flush 铺路:满批/超时/停机三路径将共用同一刷出段,失败语义只写一份。
flushLock 使业务批量方法任一时刻至多一个调用(rocketmq 端原有多满批并发刷
顺带收紧为串行,设计已确认)。行为对既有测试不变。

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

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

@@ -25,6 +25,8 @@ public class MqConsumeDispatcher {
     private final MqConsumeErrorHandler errorHandler;
     private final MqBatchConsumer consumer;
     private final List<MqMessage> buffer = new ArrayList<>();
+    /** 串行化三条刷出路径（满批/超时/停机）的业务调用段：业务批量方法任一时刻至多一个调用在跑。 */
+    private final Object flushLock = new Object();
 
     private MqConsumeDispatcher(Builder b) {
         this.batchSize = b.batchSize;
@@ -73,11 +75,21 @@ public class MqConsumeDispatcher {
             }
         }
         if (flush != null) {
+            doFlush(flush);
+        }
+    }
+
+    /**
+     * 刷出一批：业务消费 → 成功逐条 markSuccess / 失败逐条 SINGLE 落库 + 吞。
+     * 满批 / 超时 / 停机三条路径都汇聚到这里，失败语义只写一份：
+     * 批量消息已逐条 ACK 无法精确重投整批 → 对齐 legacy 不卡队列：落库 + 吞掉，
+     * 不抛、不标 success（去重 processing 标记自然过期）。
+     */
+    private void doFlush(List<MqMessage> flush) {
+        synchronized (flushLock) {
             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);