From b4ec55baf518b54da2b5e2920600beff0cd0c3c7 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Wed, 15 Jul 2026 03:35:18 +0100 Subject: [PATCH] =?UTF-8?q?fix(mq-starter-core):=20=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E7=BC=BA=E7=9C=81=E6=97=B6=E9=92=9F=E6=94=B9?= =?UTF-8?q?=E5=8D=95=E8=B0=83=E6=BA=90,=E5=85=8D=E5=8F=97=20NTP=20?= =?UTF-8?q?=E5=9B=9E=E6=8B=A8=E5=BD=B1=E5=93=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit review 发现 #3:缺省 clock = System::currentTimeMillis,超时判断是差值 比较(now - batchStartAtMs),墙钟被 NTP 回拨会让差值变负,该批多躺回拨量, 突破"刷出延迟上界一个超时窗口"的文档承诺。 修复:缺省 clock 换 System.nanoTime()/1_000_000(单调,只用于相对差值)。 clock 注入口子不变,既有假时钟测试不受影响。 测试说明:回拨行为单测无法注入(系统时钟不可控),新增缺省时钟冒烟测试 守护纳秒→毫秒换算(错用裸 nanoTime 会让任何批立即被判超时),非严格 TDD RED——超时语义本身已有注入假时钟的用例覆盖。 Co-Authored-By: Claude Fable 5 --- .../njcn/mq/container/MqConsumeDispatcher.java | 4 +++- .../mq/container/MqConsumeDispatcherTest.java | 16 ++++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) 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 cd6dcf3..877c63d 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 @@ -187,7 +187,9 @@ public class MqConsumeDispatcher { public static class Builder { private int batchSize = 1; private int batchTimeoutMs; // 缺省 0 = 关闭（注解缺省 60s 由 registry 传入） - private LongSupplier clock = System::currentTimeMillis; + // 单调时钟:只做相对差值(now - batchStartAtMs),用 nanoTime 免受 NTP 回拨影响 + // (墙钟回拨会让差值变负,批多躺回拨量,突破"延迟上界一个超时窗口"承诺)。 + private LongSupplier clock = () -> System.nanoTime() / 1_000_000; private boolean idempotent = false; private MqIdempotentStore store; private MqConsumeErrorHandler errorHandler; 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 a5ab0ea..09305c6 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 @@ -334,6 +334,22 @@ class MqConsumeDispatcherTest { assertFalse(d.needsTimeoutFlush(), "关闭超时的批量监听器不应纳入定时扫描"); } + @Test + void defaultClock_freshBatchNotDue() throws Exception { + // 不注入 clock,守护缺省单调时钟的纳秒→毫秒换算:错用裸 nanoTime 会让差值以纳秒计, + // 刚入缓冲的批立即被判超时刷出。 + List> consumed = new ArrayList<>(); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(3).batchTimeoutMs(60_000) + .consumer(consumed::add) + .build(); + + d.dispatch(MqMessage.of("k1", "", "a")); + d.flushIfTimeout(); + + assertEquals(0, consumed.size(), "刚入缓冲远未到 60s,缺省时钟不应判超时"); + } + @Test void forceFlush_flushesLeftovers() throws Exception { List> consumed = new ArrayList<>();