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 206db4e..0acd9bb 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 @@ -146,7 +146,13 @@ public class MqListenerRegistry implements SmartLifecycle { @Override public synchronized void stop() { // 停机顺序:driver 先停(不再有新消息入缓冲)→ 定时器停 → 强刷残留批 → 置 running。 - driver.stop(); + // driver.stop 异常必须吞掉:否则跳过后面整个强刷尾巴(缓冲里已 ACK 的批直接丢), + // 且 SmartLifecycle 回调没走完会卡 Spring 关闭。 + try { + driver.stop(); + } catch (Exception e) { + log.error("[mq] driver 停止失败(吞掉,继续停机流程强刷残留批)", e); + } if (flushScheduler != null) { // 温和关闭(不打断正在执行的最后一次刷出;后续执行取消,残留批由下方 forceFlush 接住)。 flushScheduler.shutdown(); 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 0b12de8..387f676 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 @@ -223,6 +223,33 @@ class MqListenerRegistryTest { assertNull(reg.getFlushScheduler(), "停机后定时器应已关闭置空"); } + /** fake driver:stop 时抛异常。 */ + static class FailingStopDriver implements MqDriver { + MqSubscription captured; + @Override public void send(String topic, MqMessage m) {} + @Override public void stop() { throw new RuntimeException("driver stop fail"); } + @Override public void subscribe(MqSubscription s) { this.captured = s; } + } + + @Test + void stop_whenDriverStopThrows_stillFlushesLeftoverAndCompletes() throws Exception { + List events = new ArrayList<>(); + LeftoverBiz biz = new LeftoverBiz(events); + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(biz, "biz"); + FailingStopDriver driver = new FailingStopDriver(); + MqListenerRegistry reg = new MqListenerRegistry(driver, bpp.getEndpoints()); + reg.start(); + driver.captured.getListener().onMessage(MqMessage.of("k1", "", "a")); + driver.captured.getListener().onMessage(MqMessage.of("k2", "", "b")); // batchSize=3,2 条不满批 + + reg.stop(); // driver.stop 抛异常不得中断停机流程 + + assertEquals(Arrays.asList("flush:2"), events, "driver.stop 抛异常也必须强刷残留批(否则已 ACK 的批静默丢)"); + assertNull(reg.getFlushScheduler(), "driver.stop 抛异常也应关闭定时器"); + assertFalse(reg.isRunning(), "driver.stop 抛异常也应完成停机置 running=false"); + } + @Test void stop_waitsForInFlightTimedFlushBeforeReturning() throws Exception { BatchBiz biz = new BatchBiz();