From 02469dce11f220b65fc3595090d7ba787193b46f Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Wed, 15 Jul 2026 02:52:43 +0100 Subject: [PATCH] =?UTF-8?q?fix(mq-starter-core):=20stop()=20=E7=AD=89?= =?UTF-8?q?=E5=BE=85=E5=9C=A8=E9=80=94=E5=AE=9A=E6=97=B6=E5=88=B7=E6=89=B9?= =?UTF-8?q?,=E5=A0=B5=E4=BD=8F=E5=81=9C=E6=9C=BA=E9=9D=99=E9=BB=98?= =?UTF-8?q?=E4=B8=A2=E6=89=B9=E7=BC=9D=E9=9A=99?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit review 发现 #1:定时线程刚做完超时快照(buffer 已清空)、正在 doFlush 跑业务 方法时优雅停机——shutdown() 不等它,forceFlush() 见空 buffer 直接返回(连 flushLock 都不进),stop 走完后 JVM 退出杀掉 daemon 线程,该批消息早已逐条 ACK,静默丢失。这是终审 shutdownNow→shutdown 的对偶缝隙:温和关闭解决了 "不打断在刷批",但没人等它刷完。 修复:shutdown() 后补 awaitTermination(30s 上界,超时/中断打 WARN 继续停机, 不让挂死的业务方法永久卡住 Spring 关闭)。 测试:往真定时线程塞在途任务模拟"stop 瞬间定时线程正在刷批",断言 stop 返回前任务已完成(修复前该测试失败:stop 立即返回)。 Co-Authored-By: Claude Fable 5 --- .../njcn/mq/container/MqListenerRegistry.java | 10 +++++++ .../mq/container/MqListenerRegistryTest.java | 27 +++++++++++++++++++ 2 files changed, 37 insertions(+) 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 bae3ad4..206db4e 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 @@ -150,6 +150,16 @@ public class MqListenerRegistry implements SmartLifecycle { if (flushScheduler != null) { // 温和关闭(不打断正在执行的最后一次刷出;后续执行取消,残留批由下方 forceFlush 接住)。 flushScheduler.shutdown(); + // 必须等在途刷批跑完:超时快照已把批移出 buffer(下方 forceFlush 看不见它), + // 不等就返回会让 JVM 退出杀掉 daemon 线程,已逐条 ACK 的整批静默丢失。 + try { + if (!flushScheduler.awaitTermination(30, TimeUnit.SECONDS)) { + log.warn("[mq] 等待在途批量刷出超过 30s 仍未完成，放弃等待继续停机（该批可能丢失）"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.warn("[mq] 等待在途批量刷出被中断，继续停机（该批可能丢失）"); + } flushScheduler = null; } for (MqConsumeDispatcher d : dispatchers) { diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java index 66172d2..0b12de8 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java +++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerRegistryTest.java @@ -14,6 +14,9 @@ import java.util.List; import java.util.Map; import java.util.function.Function; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; @@ -219,4 +222,28 @@ class MqListenerRegistryTest { "停机应先停 driver 再强刷残留批（2 条未满批不丢）"); assertNull(reg.getFlushScheduler(), "停机后定时器应已关闭置空"); } + + @Test + void stop_waitsForInFlightTimedFlushBeforeReturning() throws Exception { + BatchBiz biz = new BatchBiz(); + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(biz, "biz"); + MqListenerRegistry reg = new MqListenerRegistry(new CaptureDriver(), bpp.getEndpoints()); + reg.start(); + + // 模拟停机瞬间定时线程正在 doFlush 跑业务方法（超时快照已清空 buffer，批只在该线程手上）： + // 直接往真定时线程塞一个在途任务。stop 若不等它跑完，JVM 退出即静默丢批。 + CountDownLatch entered = new CountDownLatch(1); + AtomicBoolean completed = new AtomicBoolean(false); + reg.getFlushScheduler().execute(() -> { + entered.countDown(); + try { Thread.sleep(300); } catch (InterruptedException ignored) { } + completed.set(true); + }); + assertTrue(entered.await(2, TimeUnit.SECONDS), "在途刷批任务应已开始执行"); + + reg.stop(); + + assertTrue(completed.get(), "stop 返回前应等待定时线程在途的刷批任务完成"); + } }