feat(mq-starter-core): @MqListener batchSize 与首参 List 错配时打 WARN 软校验
走不走批量始终只由首参是否 List 决定,错配不阻断启动: - batchSize>1 但首参非 List → WARN 提示 batchSize 不生效、按单条逐一回调 - 首参 List 但 batchSize<=1 → WARN 提示将按 1 条一批回调 测试:ListAppender 捕获日志,覆盖两个错配方向 + 匹配时无 WARN。 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,6 +1,7 @@
|
|||||||
package com.njcn.mq.container;
|
package com.njcn.mq.container;
|
||||||
import com.njcn.mq.annotation.MqListener;
|
import com.njcn.mq.annotation.MqListener;
|
||||||
import com.njcn.mq.core.MqContext;
|
import com.njcn.mq.core.MqContext;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.beans.BeansException;
|
import org.springframework.beans.BeansException;
|
||||||
import org.springframework.beans.factory.config.BeanPostProcessor;
|
import org.springframework.beans.factory.config.BeanPostProcessor;
|
||||||
import org.springframework.core.annotation.AnnotatedElementUtils;
|
import org.springframework.core.annotation.AnnotatedElementUtils;
|
||||||
@@ -8,6 +9,7 @@ import java.lang.reflect.Method;
|
|||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
|
|
||||||
|
@Slf4j
|
||||||
public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor {
|
public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor {
|
||||||
|
|
||||||
private final List<MqListenerEndpoint> endpoints = new ArrayList<>();
|
private final List<MqListenerEndpoint> endpoints = new ArrayList<>();
|
||||||
@@ -39,6 +41,12 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor
|
|||||||
} else {
|
} else {
|
||||||
payloadType = first;
|
payloadType = first;
|
||||||
}
|
}
|
||||||
|
// 走不走批量只由首参是否 List 决定,batchSize 与之错配时静默按首参跑,这里 WARN 提示而不阻断启动。
|
||||||
|
if (!batch && meta.batchSize() > 1) {
|
||||||
|
log.warn("[mq] @MqListener batchSize={} 但首参非 List,batchSize 不生效、将按单条逐一回调(如需批量攒批请把首参改为 List<T>): {}", meta.batchSize(), m);
|
||||||
|
} else if (batch && meta.batchSize() <= 1) {
|
||||||
|
log.warn("[mq] @MqListener 首参为 List 但 batchSize={},将按 1 条一批回调(如需攒批请设置 batchSize>1): {}", meta.batchSize(), m);
|
||||||
|
}
|
||||||
m.setAccessible(true);
|
m.setAccessible(true);
|
||||||
endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch));
|
endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch));
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,11 +1,35 @@
|
|||||||
package com.njcn.mq.container;
|
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.annotation.MqListener;
|
||||||
import com.njcn.mq.core.MqContext;
|
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.junit.jupiter.api.Test;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
import java.util.List;
|
import java.util.List;
|
||||||
import static org.junit.jupiter.api.Assertions.*;
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
class MqListenerScanTest {
|
class MqListenerScanTest {
|
||||||
|
|
||||||
|
private final ListAppender<ILoggingEvent> 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 {
|
static class Biz {
|
||||||
@MqListener(topic = "LN_Topic", group = "ln_consumer")
|
@MqListener(topic = "LN_Topic", group = "ln_consumer")
|
||||||
public void onData(String msg) { }
|
public void onData(String msg) { }
|
||||||
@@ -39,4 +63,42 @@ class MqListenerScanTest {
|
|||||||
assertTrue(ep.isBatch(), "List 参数应识别为批量");
|
assertTrue(ep.isBatch(), "List 参数应识别为批量");
|
||||||
assertEquals(String.class, ep.getPayloadType(), "应解析 List 的元素类型 T");
|
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<String> 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");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user