From d5161fa6503af6ad754005cd99343594fb1b94a7 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Wed, 15 Jul 2026 08:27:10 +0800 Subject: [PATCH] =?UTF-8?q?fix(mq-starter-core):=20=E7=BB=88=E5=AE=A1?= =?UTF-8?q?=E4=B8=A4=E5=A4=84=E5=81=9C=E6=9C=BA=E7=BC=9D=E9=9A=99=E2=80=94?= =?UTF-8?q?=E2=80=94dispatcher=20=E6=89=93=E7=83=8A=E6=8B=92=E6=94=B6=20+?= =?UTF-8?q?=20=E5=AE=9A=E6=97=B6=E5=99=A8=E6=B8=A9=E5=92=8C=E5=85=B3?= =?UTF-8?q?=E9=97=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1) forceFlush 置 closed:driver.stop() 异步返回后残余线程再投递的消息一律 拒收抛 InProgress(不 ACK、下次启动重投),堵住正常停机丢最后一轮消息的缝隙; 检查先于去重,不留 processing 标记。 2) 定时器 shutdownNow→shutdown:不再打断正在执行的刷出,避免整批被误判失败; 后续执行照常取消,残留批由 forceFlush 接住(flushLock 串行兜底)。 终审 Important #1/#2,处置方案经用户确认。 Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 2 +- .../mq/container/MqConsumeDispatcher.java | 13 +++++- .../njcn/mq/container/MqListenerRegistry.java | 3 +- .../mq/container/MqConsumeDispatcherTest.java | 40 +++++++++++++++++++ 4 files changed, 55 insertions(+), 3 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index ae70949..2372c24 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -63,7 +63,7 @@ mq-spring-boot-starter 聚合 jar，业务方只引这一个（恒含 idemp - **去重的两类跳过**决定 driver 能否 ACK：已 `success` → 正常返回让 driver ACK 挡掉重复；仍 `processing`（并发中或消费者崩溃残留标记）→ 抛 `MqMessageInProgressException` 让 driver **不要 ACK**、保留消息待 `processing` 标记过期后重投。搞错就是**静默丢消息**。 - **单条消费失败**：先落库（**仅当配了 `errorHandler`**；`safeOnError` 对 null handler 静默跳过、不落库）+ `store.markFail()`（把去重标记从 processing 复位），再抛原异常交 driver 重投。跳过 `markFail` 会让标记卡在 processing，reclaim 重投时被去重挡掉并 ACK → 静默丢消息。 -- **批量消费**：满批 / 超时（`batchTimeoutMs` 缺省 60s，从本批首条算起；registry 全局单线程 `mq-batch-flush` 每 1s 扫）/ 停机（stop 时 `forceFlush`）三路径刷出，统一走 `doFlush`（`flushLock` 串行化，业务批量方法任一时刻至多一个调用）；kill -9 仍丢内存未刷批（上界 = 一个超时窗口）。批量失败因消息已逐条 ACK 无法精确重投整批，故落库（同样只在配了 `errorHandler` 时）+ 吞掉不抛。 +- **批量消费**：满批 / 超时（`batchTimeoutMs` 缺省 60s，从本批首条算起；registry 全局单线程 `mq-batch-flush` 每 1s 扫）/ 停机（stop 时 `forceFlush`）三路径刷出（forceFlush 同时打烊：停机后迟到的消息拒收抛 InProgress、不 ACK、下次启动重投，堵住 driver.stop 异步返回的缝隙），统一走 `doFlush`（`flushLock` 串行化，业务批量方法任一时刻至多一个调用）；kill -9 仍丢内存未刷批（上界 = 一个超时窗口）。批量失败因消息已逐条 ACK 无法精确重投整批，故落库（同样只在配了 `errorHandler` 时）+ 吞掉不抛。 driver 侧对 `MqMessageInProgressException` 的对应处理：`RedisStreamMqDriver.java` 捕获后不 `acknowledge()`（留 PEL 待 reclaim）；`RocketMqDriver.java` 在通用 `catch(Exception)` **之前**单独捕获并返回 `RECONSUME_LATER`（绝不进耗尽落库分支——catch 顺序不能动，它是 `Exception` 子类）。边缘（**两端共性**，2026-07-14 复核）：InProgress 白跑会烧重试计数——rocketmq 每次 `RECONSUME_LATER` 使 `reconsumeTimes` +1，烧光进原生 `%DLQ%`（进死信、非丢消息）；redis-stream 每次 reclaim（xClaim）使 PEL delivery +1，`deliveryCount > maxRetry` 时**不再 dispatch** 直接判耗尽（配了 `errorHandler` 落库，没配则静默丢弃——业务可能一次都没重跑）。缓解同构：`mq.idempotent.processing-ttl` 小于 maxRetry 轮重投累计时长（rocketmq 默认 ~100s / redis-stream 为 `maxRetry × reclaim-min-idle` 默认 90s）。 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 331a62a..c03fd27 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 @@ -34,6 +34,9 @@ public class MqConsumeDispatcher { private final List buffer = new ArrayList<>(); /** 串行化三条刷出路径（满批/超时/停机）的业务调用段：业务批量方法任一时刻至多一个调用在跑。 */ private final Object flushLock = new Object(); + /** 停机打烊标志:forceFlush 置位后 dispatch 一律拒收抛 InProgress——driver 不 ACK、下次启动重投, + * 堵住"driver.stop() 异步返回后残余线程仍入缓冲"的丢消息缝隙。volatile 供 driver 线程快速路径读取。 */ + private volatile boolean closed = false; private MqConsumeDispatcher(Builder b) { this.batchSize = b.batchSize; @@ -46,6 +49,10 @@ public class MqConsumeDispatcher { } public void dispatch(MqMessage msg) throws Exception { + if (closed) { + // 已打烊:抛 InProgress 信号让 driver 不 ACK,消息留在 MQ 下次启动重投(不碰去重标记)。 + throw new MqMessageInProgressException(msg.getKey()); + } if (idempotent && store != null && !store.tryAcquire(msg.getKey())) { // 跳过分两类，决定 driver 能否 ACK： // - 已 success：正常返回 → driver ACK，挡掉重复投递（正常去重）； @@ -77,6 +84,9 @@ public class MqConsumeDispatcher { // 攒批：满 batchSize 才 flush。锁内只攒/取快照，消费放锁外，避免持锁做 IO。 List flush = null; synchronized (buffer) { + if (closed) { + throw new MqMessageInProgressException(msg.getKey()); + } if (buffer.isEmpty()) { batchStartAtMs = clock.getAsLong(); } @@ -131,10 +141,11 @@ public class MqConsumeDispatcher { } } - /** 停机强制刷出残留批（非空即刷，不看超时）。 */ + /** 停机强制刷出残留批并打烊(非空即刷,不看超时;打烊后 dispatch 一律拒收,见 closed 字段)。 */ public void forceFlush() { List flush = null; synchronized (buffer) { + closed = true; if (!buffer.isEmpty()) { flush = new ArrayList<>(buffer); buffer.clear(); diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java index 8953e0e..bae3ad4 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java @@ -148,7 +148,8 @@ public class MqListenerRegistry implements SmartLifecycle { // 停机顺序：driver 先停（不再有新消息入缓冲）→ 定时器停 → 强刷残留批 → 置 running。 driver.stop(); if (flushScheduler != null) { - flushScheduler.shutdownNow(); + // 温和关闭(不打断正在执行的最后一次刷出;后续执行取消,残留批由下方 forceFlush 接住)。 + flushScheduler.shutdown(); flushScheduler = null; } for (MqConsumeDispatcher d : dispatchers) { 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 31c0315..3d5d088 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 @@ -4,6 +4,7 @@ import com.njcn.mq.core.MqMessage; import com.njcn.mq.spi.MqConsumeErrorHandler; import com.njcn.mq.spi.MqErrorIdentity; import com.njcn.mq.spi.MqIdempotentStore; +import com.njcn.mq.spi.MqMessageInProgressException; import org.junit.jupiter.api.Test; import java.util.ArrayList; @@ -395,4 +396,43 @@ class MqConsumeDispatcherTest { for (List batch : consumed) for (MqMessage m : batch) keys.add(m.getKey()); assertEquals(99, keys.size(), "99 个 key 应全部恰好出现一次"); } + + @Test + void forceFlush_thenDispatch_rejectsSoDriverWontAck() throws Exception { + List> consumed = new ArrayList<>(); + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(3).batchTimeoutMs(60_000) + .consumer(consumed::add) + .build(); + + d.dispatch(MqMessage.of("k1", "", "a")); + d.forceFlush(); // 停机强刷 + 打烊 + assertEquals(1, consumed.size(), "残留批应被停机强刷刷出"); + + assertThrows(MqMessageInProgressException.class, + () -> d.dispatch(MqMessage.of("k2", "", "b")), + "打烊后迟到的消息应拒收抛 InProgress,driver 不 ACK、下次启动重投"); + d.forceFlush(); + assertEquals(1, consumed.size(), "拒收的消息不得进入缓冲(再次强刷无动作)"); + } + + @Test + void closed_rejectsSingleModeToo_beforeTouchingDedupStore() { + List acquired = new ArrayList<>(); + MqIdempotentStore store = new MqIdempotentStore() { + @Override public boolean tryAcquire(String key) { acquired.add(key); return true; } + @Override public void markSuccess(String key) { } + @Override public void markFail(String key) { } + }; + MqConsumeDispatcher d = MqConsumeDispatcher.builder() + .batchSize(1).idempotent(true).store(store) + .consumer(batch -> { }) + .build(); + + d.forceFlush(); // 空缓冲仅打烊 + assertThrows(MqMessageInProgressException.class, + () -> d.dispatch(MqMessage.of("k1", "", "a")), + "打烊后单条路径同样拒收"); + assertEquals(0, acquired.size(), "拒收须发生在去重之前,不得留下 processing 标记"); + } }