fix(mq-starter-core): stop() 等待在途定时刷批,堵住停机静默丢批缝隙
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 <noreply@anthropic.com>
This commit is contained in:
@@ -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) {
|
||||
|
||||
@@ -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 返回前应等待定时线程在途的刷批任务完成");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user