Middleware/mq-starter
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mq-starter-core): @MqListener 新增 batchTimeoutMs(批量超时刷出,缺省 60s)+ 启动软校验

Browse Source 
字段仅批量监听器生效;batchSize<=1 且显式配非缺省值时打 WARN 不阻断
(延续 batchSize 软校验风格;恰好显式写 60000 检测不出,已知可接受)。

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
hongawen
2026-07-14 20:49:48 +08:00
parent cfe9d0e994
commit ad864e5b93
3 changed files with 34 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java
View File

@@ -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;
}

5
mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerAnnotationBeanPostProcessor.java
View File

@@ -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));
}

26
mq-starter-core/src/test/java/com/njcn/mq/container/MqListenerScanTest.java
View File

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