From 9af2ef06e222065dcc1a2bfadbe69d11d59767be Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Wed, 15 Jul 2026 03:37:53 +0100 Subject: [PATCH] =?UTF-8?q?refactor(mq-starter-core):=20doFlush=20?= =?UTF-8?q?=E7=9A=84=20markSuccess=20=E5=BE=AA=E7=8E=AF=E6=8C=AA=E5=87=BA?= =?UTF-8?q?=20flushLock?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit review 发现 #9:成功标记是逐条 store 往返(Redis 网络 IO),却跑在 flushLock 内——该锁本职只是串行化业务批量调用,批已出缓冲、key 逐条 独立,标记占锁只是白白让下一批的业务刷出排队(最多多等 batchSize 次 网络往返)。 修复:markSuccess 循环移到 synchronized 块外。失败路径(落库+吞)不变, 仍在锁内直接 return。去重正确性不受影响:同 key 不会同时出现在两批 (tryAcquire 挡住),store 本身并发安全。 测试:store 回调在首批标记进行中从另一线程投满一批,断言其刷出不被 flushLock 挡住(修复前该测试失败:另一线程阻塞在锁上)。 Co-Authored-By: Claude Fable 5 --- .../mq/container/MqConsumeDispatcher.java | 10 +++-- .../mq/container/MqConsumeDispatcherTest.java | 45 +++++++++++++++++++ 2 files changed, 51 insertions(+), 4 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 877c63d..21d1ccd 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 @@ -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()); } } } diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java index 09305c6..5e80cd8 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java +++ b/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> 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 会让差值以纳秒计,