fix(mq-starter-core): 批量超时缺省时钟改单调源,免受 NTP 回拨影响
review 发现 #3:缺省 clock = System::currentTimeMillis,超时判断是差值 比较(now - batchStartAtMs),墙钟被 NTP 回拨会让差值变负,该批多躺回拨量, 突破"刷出延迟上界一个超时窗口"的文档承诺。 修复:缺省 clock 换 System.nanoTime()/1_000_000(单调,只用于相对差值)。 clock 注入口子不变,既有假时钟测试不受影响。 测试说明:回拨行为单测无法注入(系统时钟不可控),新增缺省时钟冒烟测试 守护纳秒→毫秒换算(错用裸 nanoTime 会让任何批立即被判超时),非严格 TDD RED——超时语义本身已有注入假时钟的用例覆盖。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -187,7 +187,9 @@ public class MqConsumeDispatcher {
|
|||||||
public static class Builder {
|
public static class Builder {
|
||||||
private int batchSize = 1;
|
private int batchSize = 1;
|
||||||
private int batchTimeoutMs; // 缺省 0 = 关闭(注解缺省 60s 由 registry 传入)
|
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 boolean idempotent = false;
|
||||||
private MqIdempotentStore store;
|
private MqIdempotentStore store;
|
||||||
private MqConsumeErrorHandler errorHandler;
|
private MqConsumeErrorHandler errorHandler;
|
||||||
|
|||||||
@@ -334,6 +334,22 @@ class MqConsumeDispatcherTest {
|
|||||||
assertFalse(d.needsTimeoutFlush(), "关闭超时的批量监听器不应纳入定时扫描");
|
assertFalse(d.needsTimeoutFlush(), "关闭超时的批量监听器不应纳入定时扫描");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void defaultClock_freshBatchNotDue() throws Exception {
|
||||||
|
// 不注入 clock,守护缺省单调时钟的纳秒→毫秒换算:错用裸 nanoTime 会让差值以纳秒计,
|
||||||
|
// 刚入缓冲的批立即被判超时刷出。
|
||||||
|
List<List<MqMessage>> 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
|
@Test
|
||||||
void forceFlush_flushesLeftovers() throws Exception {
|
void forceFlush_flushesLeftovers() throws Exception {
|
||||||
List<List<MqMessage>> consumed = new ArrayList<>();
|
List<List<MqMessage>> consumed = new ArrayList<>();
|
||||||
|
|||||||
Reference in New Issue
Block a user