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"); + } }