From 8c39399d099422bdf039f2ef642e127633b4138d Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Tue, 14 Jul 2026 04:19:17 +0100 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-core):=20@MqListener=20batchSiz?= =?UTF-8?q?e=20=E4=B8=8E=E9=A6=96=E5=8F=82=20List=20=E9=94=99=E9=85=8D?= =?UTF-8?q?=E6=97=B6=E6=89=93=20WARN=20=E8=BD=AF=E6=A0=A1=E9=AA=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 走不走批量始终只由首参是否 List 决定，错配不阻断启动： - batchSize>1 但首参非 List → WARN 提示 batchSize 不生效、按单条逐一回调 - 首参 List 但 batchSize<=1 → WARN 提示将按 1 条一批回调 测试：ListAppender 捕获日志，覆盖两个错配方向 + 匹配时无 WARN。 Co-Authored-By: Claude Fable 5 --- ...MqListenerAnnotationBeanPostProcessor.java | 8 +++ .../njcn/mq/container/MqListenerScanTest.java | 62 +++++++++++++++++++ 2 files changed, 70 insertions(+) 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 a125b80..97fc9f8 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 @@ -1,6 +1,7 @@ package com.njcn.mq.container; import com.njcn.mq.annotation.MqListener; import com.njcn.mq.core.MqContext; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.annotation.AnnotatedElementUtils; @@ -8,6 +9,7 @@ import java.lang.reflect.Method; import java.util.ArrayList; import java.util.List; +@Slf4j public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor { private final List endpoints = new ArrayList<>(); @@ -39,6 +41,12 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor } else { payloadType = first; } + // 走不走批量只由首参是否 List 决定，batchSize 与之错配时静默按首参跑，这里 WARN 提示而不阻断启动。 + if (!batch && meta.batchSize() > 1) { + log.warn("[mq] @MqListener batchSize={} 但首参非 List，batchSize 不生效、将按单条逐一回调（如需批量攒批请把首参改为 List）: {}", meta.batchSize(), m); + } else if (batch && meta.batchSize() <= 1) { + log.warn("[mq] @MqListener 首参为 List 但 batchSize={}，将按 1 条一批回调（如需攒批请设置 batchSize>1）: {}", 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 c1336e2..74faba8 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 @@ -1,11 +1,35 @@ package com.njcn.mq.container; +import ch.qos.logback.classic.Level; +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; import com.njcn.mq.annotation.MqListener; import com.njcn.mq.core.MqContext; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.LoggerFactory; import java.util.List; import static org.junit.jupiter.api.Assertions.*; class MqListenerScanTest { + + private final ListAppender logAppender = new ListAppender<>(); + + @BeforeEach + void attachAppender() { + logAppender.start(); + ((Logger) LoggerFactory.getLogger(MqListenerAnnotationBeanPostProcessor.class)).addAppender(logAppender); + } + + @AfterEach + void detachAppender() { + ((Logger) LoggerFactory.getLogger(MqListenerAnnotationBeanPostProcessor.class)).detachAppender(logAppender); + } + + private long warnCount() { + return logAppender.list.stream().filter(e -> e.getLevel() == Level.WARN).count(); + } static class Biz { @MqListener(topic = "LN_Topic", group = "ln_consumer") public void onData(String msg) { } @@ -39,4 +63,42 @@ class MqListenerScanTest { assertTrue(ep.isBatch(), "List 参数应识别为批量"); assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T"); } + + static class BatchSizeButSinglePayload { + @MqListener(topic = "W1", group = "wg1", batchSize = 50) + public void onOne(String msg) { } + } + + static class ListButBatchSizeOne { + @MqListener(topic = "W2", group = "wg2") + public void onList(List msgs) { } + } + + @Test + void scan_batchSizeGtOneButSinglePayload_warnsAndRunsAsSingle() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new BatchSizeButSinglePayload(), "biz"); + MqListenerEndpoint ep = bpp.getEndpoints().get(0); + assertFalse(ep.isBatch(), "首参非 List 应按单条跑"); + assertEquals(1, warnCount(), "batchSize>1 但首参非 List 应打一条 WARN"); + assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchSize")); + } + + @Test + void scan_listParamButBatchSizeOne_warnsAndRunsAsBatch() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new ListButBatchSizeOne(), "biz"); + MqListenerEndpoint ep = bpp.getEndpoints().get(0); + assertTrue(ep.isBatch(), "首参 List 仍按批量跑"); + assertEquals(1, warnCount(), "首参 List 但 batchSize<=1 应打一条 WARN"); + assertTrue(logAppender.list.get(0).getFormattedMessage().contains("batchSize")); + } + + @Test + void scan_matchedSignatures_noWarn() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new Biz(), "biz"); + bpp.postProcessAfterInitialization(new BatchBiz(), "batchBiz"); + assertEquals(0, warnCount(), "batchSize 与首参匹配时不应有 WARN"); + } }