From 98731c3591ff604dc3e60e1e70bb8d98d2fce11c Mon Sep 17 00:00:00 2001
From: hongawen <83944980@qq.com>
Date: Tue, 14 Jul 2026 20:56:08 +0800
Subject: [PATCH] =?UTF-8?q?refactor(mq-starter-core):=20=E6=8A=BD=E5=8F=96?=
 =?UTF-8?q?=20doFlush=20=E7=BB=9F=E4=B8=80=E6=89=B9=E9=87=8F=E5=88=B7?=
 =?UTF-8?q?=E5=87=BA=E8=B7=AF=E5=BE=84=20+=20flushLock=20=E4=B8=B2?=
 =?UTF-8?q?=E8=A1=8C=E5=8C=96=E4=B8=9A=E5=8A=A1=E8=B0=83=E7=94=A8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
---
 .../njcn/mq/container/MqConsumeDispatcher.java   | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java
index 6f2622c..c6a81f4 100644
--- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java
+++ b/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);