refactor(mq-starter-core): doFlush 的 markSuccess 循环挪出 flushLock

review 发现 #9:成功标记是逐条 store 往返(Redis 网络 IO),却跑在
flushLock 内——该锁本职只是串行化业务批量调用,批已出缓冲、key 逐条
独立,标记占锁只是白白让下一批的业务刷出排队(最多多等 batchSize 次
网络往返)。

修复:markSuccess 循环移到 synchronized 块外。失败路径(落库+吞)不变,
仍在锁内直接 return。去重正确性不受影响:同 key 不会同时出现在两批
(tryAcquire 挡住),store 本身并发安全。

测试:store 回调在首批标记进行中从另一线程投满一批,断言其刷出不被
flushLock 挡住(修复前该测试失败:另一线程阻塞在锁上)。

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

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

@@ -129,10 +129,12 @@ public class MqConsumeDispatcher {
                 }
                 return;
             }
-            if (idempotent && store != null) {
-                for (MqMessage m : flush) {
-                    store.markSuccess(m.getKey());
-                }
+        }
+        // markSuccess 逐条 store 往返，放锁外：flushLock 只串行化业务调用，
+        // 批已出缓冲、key 逐条独立，标记不必让下一批的业务刷出等待。
+        if (idempotent && store != null) {
+            for (MqMessage m : flush) {
+                store.markSuccess(m.getKey());
             }
         }
     }

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

@@ -12,6 +12,9 @@ import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.junit.jupiter.api.Assertions.*;
 
@@ -334,6 +337,48 @@ class MqConsumeDispatcherTest {
         assertFalse(d.needsTimeoutFlush(), "关闭超时的批量监听器不应纳入定时扫描");
     }
 
+    @Test
+    void batchMarkSuccess_doesNotHoldFlushLock() throws Exception {
+        // #9:markSuccess 是逐条 store 往返,不得占着 flushLock(它只该串行化业务调用)——
+        // 慢 store 标记期间,其他线程的满批业务刷出不应被阻塞。
+        List<List<MqMessage>> consumed = new ArrayList<>();
+        AtomicBoolean first = new AtomicBoolean(true);
+        AtomicBoolean otherFlushCompletedDuringMark = new AtomicBoolean(false);
+        CountDownLatch otherFlushDone = new CountDownLatch(1);
+        MqConsumeDispatcher[] ref = new MqConsumeDispatcher[1];
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) {
+                if (first.compareAndSet(true, false)) {
+                    // 首批标记进行中:另一线程投满一批(其 doFlush 需要 flushLock)
+                    new Thread(() -> {
+                        try {
+                            ref[0].dispatch(MqMessage.of("k3", "", "c"));
+                            ref[0].dispatch(MqMessage.of("k4", "", "d"));
+                            otherFlushDone.countDown();
+                        } catch (Exception ignored) { }
+                    }).start();
+                    try {
+                        otherFlushCompletedDuringMark.set(otherFlushDone.await(800, TimeUnit.MILLISECONDS));
+                    } catch (InterruptedException ignored) { }
+                }
+            }
+            @Override public void markFail(String key) { }
+        };
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(2).idempotent(true).store(store)
+                .consumer(consumed::add)
+                .build();
+        ref[0] = d;
+
+        d.dispatch(MqMessage.of("k1", "", "a"));
+        d.dispatch(MqMessage.of("k2", "", "b")); // 满批 → doFlush → markSuccess(k1) 触发上面回调
+
+        assertTrue(otherFlushCompletedDuringMark.get(),
+                "markSuccess 期间另一线程的满批刷出不应被 flushLock 挡住(标记应在锁外)");
+        assertEquals(2, consumed.size(), "两批都应完成业务消费");
+    }
+
     @Test
     void defaultClock_freshBatchNotDue() throws Exception {
         // 不注入 clock,守护缺省单调时钟的纳秒→毫秒换算:错用裸 nanoTime 会让差值以纳秒计,