From ad864e5b93c00bee49ac972feb160a8dbbba38c7 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Tue, 14 Jul 2026 20:49:48 +0800 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-core):=20@MqListener=20?= =?UTF-8?q?=E6=96=B0=E5=A2=9E=20batchTimeoutMs(=E6=89=B9=E9=87=8F=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E5=88=B7=E5=87=BA,=E7=BC=BA=E7=9C=81=2060s)+=20?= =?UTF-8?q?=E5=90=AF=E5=8A=A8=E8=BD=AF=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 字段仅批量监听器生效;batchSize<=1 且显式配非缺省值时打 WARN 不阻断 (延续 batchSize 软校验风格;恰好显式写 60000 检测不出,已知可接受)。 Co-Authored-By: Claude Fable 5 --- .../com/njcn/mq/annotation/MqListener.java | 3 +++ ...MqListenerAnnotationBeanPostProcessor.java | 5 ++++ .../njcn/mq/container/MqListenerScanTest.java | 26 +++++++++++++++++++ 3 files changed, 34 insertions(+) diff --git a/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java b/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java index 52ac0b3..2fb3612 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java @@ -17,4 +17,7 @@ public @interface MqListener { boolean idempotent() default true; /** 异常落库回调的 bean 名；空表示不落库。 */ String errorHandler() default ""; + /** 批量攒批的最长等待毫秒数（从本批第一条进入缓冲起算，到点未满也强制刷出，与满批先到先刷）。 + * 仅批量监听器（batchSize>1 且首参 List）有意义；0 或负数 = 关闭超时刷出（回到攒满才刷）。缺省 60 秒。 */ + int batchTimeoutMs() default 60_000; } diff --git a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java index 97fc9f8..cf03222 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java @@ -47,6 +47,11 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor } else if (batch && meta.batchSize() <= 1) { log.warn("[mq] @MqListener 首参为 List 但 batchSize={}，将按 1 条一批回调（如需攒批请设置 batchSize>1）: {}", meta.batchSize(), m); } + // batchTimeoutMs 仅批量攒批时有意义；已知局限：显式写 60000（恰为缺省值）检测不出，可接受。 + if (meta.batchSize() <= 1 && meta.batchTimeoutMs() != 60_000) { + log.warn("[mq] @MqListener batchTimeoutMs={} 仅对批量监听器（batchSize>1）生效，当前 batchSize={} 下不生效: {}", + meta.batchTimeoutMs(), meta.batchSize(), m); + } m.setAccessible(true); endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch)); } diff --git a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java index 74faba8..dea9620 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java +++ b/mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java @@ -101,4 +101,30 @@ class MqListenerScanTest { bpp.postProcessAfterInitialization(new BatchBiz(), "batchBiz"); assertEquals(0, warnCount(), "batchSize 与首参匹配时不应有 WARN"); } + + static class TimeoutButSinglePayload { + @MqListener(topic = "W3", group = "wg3", batchTimeoutMs = 5000) + public void onOne(String msg) { } + } + + static class BatchWithCustomTimeout { + @MqListener(topic = "W4", group = "wg4", batchSize = 10, batchTimeoutMs = 5000) + public void onBatch(List msgs) { } + } + + @Test + void scan_batchTimeoutOnSingleListener_warnsButNotBlocks() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new TimeoutButSinglePayload(), "biz"); + assertEquals(1, bpp.getEndpoints().size(), "软校验不应阻断启动"); + assertEquals(1, warnCount(), "batchSize<=1 配非缺省 batchTimeoutMs 应打一条 WARN"); + assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchTimeoutMs")); + } + + @Test + void scan_batchTimeoutOnBatchListener_noWarn() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new BatchWithCustomTimeout(), "biz"); + assertEquals(0, warnCount(), "批量监听器配 batchTimeoutMs 属正常用法，不应 WARN"); + } }