From 646cc6c238775126d741ab7ed982ea71a6cb13f6 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 16 Jul 2026 12:21:10 +0100 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-core):=20=E5=90=8C=20group=20?= =?UTF-8?q?=E8=AE=A2=E9=98=85=20tag=20=E4=B8=8D=E4=B8=80=E8=87=B4=E5=90=AF?= =?UTF-8?q?=E5=8A=A8=20fail-fast=EF=BC=8C=E7=97=85=E6=80=81=20tag=20?= =?UTF-8?q?=E8=A1=A8=E8=BE=BE=E5=BC=8F=E6=89=93=20WARN?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 同组配不同 tag 时消息只投组内某一个消费者，不匹配方直接 ACK 吞掉、另一方 永远收不到（订阅关系不一致，静默丢消息级事故，README 原本只靠文字约定）。 现在注解扫描期强校验：同一 group 声明不等价的 tag 表达式直接 IllegalStateException， 报错点名 group、冲突表达式与两个方法。等价性按 MqTagMatcher.validParts 有效 tag 集合比较（"A || B" 与 "B||A" 等价，*/空/病态式同为全收），只覆盖单应用内。 - MqTagMatcher 新增 validParts / looksMisconfigured，matches 重构为基于 validParts，解析口径单一来源（每消息多一次小集合分配，可忽略） - 病态表达式（"||"、纯空白：写了内容却解析不出有效 tag）扫描期打 WARN 不阻断， redis 端静默全收 vs rocketmq 端行为不同的分叉维持现状（拍板记录不修） - README tag 小节、CLAUDE.md 签名约束一节同步 Co-Authored-By: Claude Fable 5 --- CLAUDE.md | 1 + README.md | 2 +- ...MqListenerAnnotationBeanPostProcessor.java | 27 +++++- .../java/com/njcn/mq/spi/MqTagMatcher.java | 44 +++++++--- .../njcn/mq/container/MqListenerScanTest.java | 86 +++++++++++++++++++ .../com/njcn/mq/spi/MqTagMatcherTest.java | 41 +++++++++ 6 files changed, 188 insertions(+), 13 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 3e31953..b0f1dde 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -85,6 +85,7 @@ driver 只需实现 `MqDriver`（`send` / `subscribe` / `stop`）。其余是可 - `batchSize > 1` 时首参要用 `List`，且**必须声明具体元素类型**（`List`，不能裸 `List`）—— driver 仍按 `T` 逐条反序列化，dispatcher 负责攒批。 - `batchSize` 与首参是否 `List` 的**错配只打 WARN、不阻断启动**（2026-07-14 决策，刻意软校验）：走不走批量始终只由首参决定——`batchSize>1` 配单 payload 按单条跑，`List` 首参配 `batchSize<=1` 按 1 条一批跑。 - `batchTimeoutMs` 仅批量监听器生效；`batchSize<=1` 且显式配了非缺省值时同样只打 WARN 软校验、不阻断（恰好显式写 60000 检测不出，已知可接受）。 +- **同 group 订阅 tag 一致性强校验**（2026-07-16）：同一 group 声明不等价的 tag 表达式直接抛 `IllegalStateException`（订阅关系不一致 = 静默丢消息；等价性按 `MqTagMatcher.validParts` 有效 tag 集合比较，`*`/空/病态式同为全收不算冲突）。只覆盖单应用内，跨应用/滚动发布窗口的不一致防不住。病态表达式（写了内容却解析不出有效 tag，如 `"||"`、纯空白）只打 WARN 不阻断——redis 端全收 vs rocketmq 端行为不同的分叉本身维持现状（2026-07-16 拍板记录不修）。 ## 启动期 fail-fast 校验（近期新增，改动时留意） diff --git a/README.md b/README.md index e6db6c8..2c26f09 100644 --- a/README.md +++ b/README.md @@ -129,7 +129,7 @@ public void onFile(FileData msg) { } // 只收 tag=node-01 的消息， - 消费端不写 `tag` 默认 `*`，整个 topic 全收；发送端用两参 `send` 则消息不带 tag——**对端若按 tag 过滤，不带标签的消息它收不到**，跨系统对接时这是个经典坑。 - rocketmq 模式：过滤发生在 **broker 服务端**，不匹配的消息根本不会投到消费者；tag 表达式直接透传 RocketMQ 原生语法，`"A || B"` 多选一也支持。 - redis-stream 模式：**消费端模拟过滤**——driver 收到消息后按订阅 tag 比对，不匹配的直接 ACK 丢弃、不进消费链（不碰去重、不算重试、不落库），支持与 rocketmq 相同的写法子集（`*` 全收、单 tag 精确、`A || B` 多选一）。与 rocketmq 的差别只剩过滤位置与成本：消息会先走一趟网络到消费端才被丢弃。 -- **同一 group 的所有订阅必须用同一个 tag 表达式**（含滚动发布期间的新旧实例）——要按 tag 分流，就各用各的 group。同组配不同 tag 时，消息只会投给组内某一个消费者，拿到"不归自己"消息的那个会直接 ACK 吞掉，对方永远收不到（两端皆然，rocketmq 生态称之为「订阅关系不一致」）。 +- **同一 group 的所有订阅必须用同一个 tag 表达式**（含滚动发布期间的新旧实例）——要按 tag 分流，就各用各的 group。同组配不同 tag 时，消息只会投给组内某一个消费者，拿到"不归自己"消息的那个会直接 ACK 吞掉，对方永远收不到（两端皆然，rocketmq 生态称之为「订阅关系不一致」）。**同应用内这条规则由框架启动时强校验**，配错直接启动失败（等价性按有效 tag 集合比较，`"A || B"` 与 `"B||A"` 不算冲突）；跨应用共用 group、滚动发布窗口内的新旧版本，仍需靠约定保持一致。另外，写了内容却解析不出任何有效 tag 的表达式（如 `"||"`、纯空白）启动时会打 WARN——这种写法 redis-stream 按全收处理、rocketmq 行为与之不同，多半是写错了，全收请用 `*` 或留空。 ## 进阶用法 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 8fddce4..402c6e2 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,18 +1,23 @@ package com.njcn.mq.container; import com.njcn.mq.annotation.MqListener; import com.njcn.mq.core.MqContext; +import com.njcn.mq.spi.MqTagMatcher; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.BeansException; import org.springframework.beans.factory.config.BeanPostProcessor; import org.springframework.core.annotation.AnnotatedElementUtils; import java.lang.reflect.Method; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; @Slf4j public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor { private final List endpoints = new ArrayList<>(); + /** group → 该 group 首个登记的 endpoint，用于同 group 订阅 tag 一致性校验。 */ + private final Map groupFirstSeen = new HashMap<>(); public List getEndpoints() { return endpoints; } @Override @@ -53,8 +58,28 @@ public class MqListenerAnnotationBeanPostProcessor implements BeanPostProcessor log.warn("[mq] @MqListener batchTimeoutMs={} 仅对批量监听器（batchSize>1）生效，当前 batchSize={} 下不生效: {}", meta.batchTimeoutMs(), meta.batchSize(), m); } + // tag 写了内容却解析不出任何有效 tag（如 "||"、纯空白）：redis-stream 端会静默按全收处理、 + // rocketmq 端透传原生语法行为不同——大概率写错，WARN 提示不阻断（全收请用 * 或留空）。 + if (MqTagMatcher.looksMisconfigured(meta.tag())) { + log.warn("[mq] @MqListener tag=\"{}\" 解析不出任何有效 tag：redis-stream 将按全收处理、rocketmq 行为与之不同，疑似配置错误（全收请用 * 或留空）: {}", + meta.tag(), m); + } m.setAccessible(true); - endpoints.add(new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch)); + MqListenerEndpoint endpoint = new MqListenerEndpoint(bean, m, meta, payloadType, wantsContext, batch); + // 同 group 的所有订阅必须用同一个 tag 表达式（订阅关系一致性，等价性按有效 tag 集合比较）。 + // 同组配不同 tag 时消息只投组内某一个消费者，拿到"不归自己"消息的会直接 ACK 丢弃、 + // 另一方永远收不到——静默丢消息级事故，必须在启动期拦下（README tag 一节的硬规则）。 + MqListenerEndpoint firstInGroup = groupFirstSeen.get(meta.group()); + if (firstInGroup == null) { + groupFirstSeen.put(meta.group(), endpoint); + } else if (!MqTagMatcher.validParts(firstInGroup.getMeta().tag()).equals(MqTagMatcher.validParts(meta.tag()))) { + throw new IllegalStateException(String.format( + "@MqListener 同一 group 的订阅 tag 必须一致: group=%s 已声明 tag=\"%s\" (%s)，此处又声明 tag=\"%s\" (%s)。" + + "同组不同 tag 时消息只会投给组内某一个消费者，不匹配方会直接 ACK 丢弃、另一方永远收不到（订阅关系不一致）；" + + "要按 tag 分流请各用各的 group，同一类消费请把 tag 配成一致。", + meta.group(), firstInGroup.getMeta().tag(), firstInGroup.getMethod(), meta.tag(), m)); + } + endpoints.add(endpoint); } return bean; } diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqTagMatcher.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqTagMatcher.java index 9e564f3..badc37c 100644 --- a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqTagMatcher.java +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqTagMatcher.java @@ -1,5 +1,9 @@ package com.njcn.mq.spi; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + /** * 订阅 tag 表达式匹配，语义对齐 RocketMQ 原生 tag 表达式**子集**： * {@code *}/空 = 全收、单 tag 精确匹配、{@code A || B} 多选一。不支持 SQL92 / 否定 / 通配。 @@ -18,25 +22,43 @@ public final class MqTagMatcher { * @return true = 该消息归本订阅消费 */ public static boolean matches(String subscriptionTag, String messageTag) { + // 空集 = 全收（*/空/全空段病态式）；有有效段则精确命中其一。解析口径单一来源于 validParts。 + Set parts = validParts(subscriptionTag); + return parts.isEmpty() || parts.contains(messageTag); + } + + /** + * 解析表达式中的有效 tag 集合。空集 = 全收（null / 空 / {@code *} / 解析不出有效段的病态表达式）， + * 判定口径与 {@link #matches} 完全一致。供订阅一致性比对等复用——等价性按集合比较， + * 顺序与空白差异（{@code "A || B"} vs {@code "B||A"}）不构成不一致。 + */ + public static Set validParts(String subscriptionTag) { if (subscriptionTag == null) { - return true; + return Collections.emptySet(); } String expr = subscriptionTag.trim(); if (expr.isEmpty() || "*".equals(expr)) { - return true; + return Collections.emptySet(); } - boolean hasValidPart = false; + Set parts = new HashSet<>(); for (String part : expr.split("\\|\\|")) { String p = part.trim(); - if (p.isEmpty()) { - continue; // 空段忽略（如 "A ||" 的尾段） - } - hasValidPart = true; - if (p.equals(messageTag)) { - return true; + if (!p.isEmpty()) { + parts.add(p); } } - // 全空段（如 "||"）= 没声明任何有效 tag = 视同全收；有有效段但都没命中 = 不匹配 - return !hasValidPart; + return parts; + } + + /** + * true = 表达式写了内容（非空串、非 {@code *}）却解析不出任何有效 tag（如 {@code "||"}、纯空白）。 + * 这种写法 redis-stream 端静默按全收处理、rocketmq 端行为不同（透传原生语法，可能启动失败或全不收）， + * 大概率是配置写错，调用方应打 WARN 提示；全收请用 {@code *} 或留空。 + */ + public static boolean looksMisconfigured(String subscriptionTag) { + if (subscriptionTag == null || subscriptionTag.isEmpty()) { + return false; + } + return !"*".equals(subscriptionTag.trim()) && validParts(subscriptionTag).isEmpty(); } } 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 905c370..8a6971a 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 @@ -145,4 +145,90 @@ class MqListenerScanTest { bpp.postProcessAfterInitialization(new TimeoutOffNegativeOnSingle(), "biz2"); assertEquals(0, warnCount(), "0/负数 = 关闭超时，单条监听器显式关闭语义无害，不应误报 WARN"); } + + // ---- 同 group 订阅 tag 一致性（订阅关系一致性）硬校验 ---- + + static class GroupTagA { + @MqListener(topic = "GT", group = "gt_consumer", tag = "A") + public void on(String msg) { } + } + static class GroupTagB { + @MqListener(topic = "GT", group = "gt_consumer", tag = "B") + public void on(String msg) { } + } + static class GroupTagAOrB { + @MqListener(topic = "GT", group = "gt_eq", tag = "A || B") + public void on(String msg) { } + } + static class GroupTagBOrA { + @MqListener(topic = "GT2", group = "gt_eq", tag = "B||A") + public void on(String msg) { } + } + static class GroupTagStar { + @MqListener(topic = "GT", group = "gt_all", tag = "*") + public void on(String msg) { } + } + static class GroupTagEmpty { + @MqListener(topic = "GT2", group = "gt_all", tag = "") + public void on(String msg) { } + } + static class OtherGroupTagB { + @MqListener(topic = "GT", group = "gt_other", tag = "B") + public void on(String msg) { } + } + + @Test + void scan_sameGroupDifferentTag_failsFast() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new GroupTagA(), "a"); + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> bpp.postProcessAfterInitialization(new GroupTagB(), "b"), + "同 group 配不同 tag 会静默丢消息（订阅关系不一致），必须启动失败"); + assertTrue(ex.getMessage().contains("gt_consumer"), "报错须点名 group：" + ex.getMessage()); + assertTrue(ex.getMessage().contains("\"A\"") && ex.getMessage().contains("\"B\""), + "报错须给出冲突的两个 tag 表达式：" + ex.getMessage()); + } + + @Test + void scan_sameGroupEquivalentTagExpressions_notBlocked() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new GroupTagAOrB(), "a"); + bpp.postProcessAfterInitialization(new GroupTagBOrA(), "b"); // "A || B" 与 "B||A" 集合等价 + bpp.postProcessAfterInitialization(new GroupTagStar(), "c"); + bpp.postProcessAfterInitialization(new GroupTagEmpty(), "d"); // "*" 与 "" 同为全收 + assertEquals(4, bpp.getEndpoints().size(), "等价表达式不应误报不一致"); + } + + @Test + void scan_differentGroupsDifferentTags_notBlocked() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new GroupTagA(), "a"); + bpp.postProcessAfterInitialization(new OtherGroupTagB(), "b"); + assertEquals(2, bpp.getEndpoints().size(), "按 tag 分流各用各的 group 是推荐用法，不应拦截"); + } + + // ---- 病态 tag 表达式软校验（WARN 不阻断）---- + + static class PathologicalTag { + @MqListener(topic = "PT", group = "pt_consumer", tag = "||") + public void on(String msg) { } + } + + @Test + void scan_pathologicalTagExpression_warnsButNotBlocks() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new PathologicalTag(), "biz"); + assertEquals(1, bpp.getEndpoints().size(), "病态 tag 软校验不应阻断启动"); + assertEquals(1, warnCount(), "解析不出有效 tag 的表达式应打一条 WARN"); + assertTrue(logAppender.list.get(0).getFormattedMessage().contains("tag"), + "WARN 应指向 tag 配置问题"); + } + + @Test + void scan_catchAllTag_noWarn() { + MqListenerAnnotationBeanPostProcessor bpp = new MqListenerAnnotationBeanPostProcessor(); + bpp.postProcessAfterInitialization(new GroupTagStar(), "a"); + bpp.postProcessAfterInitialization(new GroupTagEmpty(), "b"); + assertEquals(0, warnCount(), "*/空串是正常全收写法，不应告警"); + } } diff --git a/mq-starter-core/src/test/java/com/njcn/mq/spi/MqTagMatcherTest.java b/mq-starter-core/src/test/java/com/njcn/mq/spi/MqTagMatcherTest.java index 83e2c13..04e53ba 100644 --- a/mq-starter-core/src/test/java/com/njcn/mq/spi/MqTagMatcherTest.java +++ b/mq-starter-core/src/test/java/com/njcn/mq/spi/MqTagMatcherTest.java @@ -2,6 +2,11 @@ package com.njcn.mq.spi; import org.junit.jupiter.api.Test; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; + +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -42,4 +47,40 @@ class MqTagMatcherTest { assertFalse(MqTagMatcher.matches("A", "")); assertFalse(MqTagMatcher.matches("A || B", null)); } + + @Test + void validParts_parsesExpressionIntoTagSet() { + assertEquals(new HashSet<>(Collections.singletonList("A")), MqTagMatcher.validParts("A")); + assertEquals(new HashSet<>(Arrays.asList("A", "B")), MqTagMatcher.validParts("A || B")); + assertEquals(new HashSet<>(Arrays.asList("A", "B")), MqTagMatcher.validParts("B||A"), + "集合语义：顺序与空白差异不影响等价性"); + assertEquals(new HashSet<>(Collections.singletonList("A")), MqTagMatcher.validParts("A ||"), + "尾空段忽略，只留有效段"); + } + + @Test + void validParts_catchAllAndPathologicalExpressions_yieldEmptySet() { + assertEquals(Collections.emptySet(), MqTagMatcher.validParts(null)); + assertEquals(Collections.emptySet(), MqTagMatcher.validParts("")); + assertEquals(Collections.emptySet(), MqTagMatcher.validParts("*")); + assertEquals(Collections.emptySet(), MqTagMatcher.validParts(" * ")); + assertEquals(Collections.emptySet(), MqTagMatcher.validParts("||"), "全空段 = 全收，与 matches 语义一致"); + } + + @Test + void looksMisconfigured_flagsNonEmptyExpressionWithNoValidTag() { + assertTrue(MqTagMatcher.looksMisconfigured("||"), "写了 || 却没有任何有效 tag"); + assertTrue(MqTagMatcher.looksMisconfigured(" || ")); + assertTrue(MqTagMatcher.looksMisconfigured(" "), "纯空白：redis 端静默全收、rocketmq 端行为不同，疑似写错"); + } + + @Test + void looksMisconfigured_legitimateExpressions_notFlagged() { + assertFalse(MqTagMatcher.looksMisconfigured(null), "没写 = 缺省全收，正常"); + assertFalse(MqTagMatcher.looksMisconfigured(""), "显式空串按全收，两端语义一致，不告警"); + assertFalse(MqTagMatcher.looksMisconfigured("*")); + assertFalse(MqTagMatcher.looksMisconfigured(" * ")); + assertFalse(MqTagMatcher.looksMisconfigured("A")); + assertFalse(MqTagMatcher.looksMisconfigured("A ||"), "有有效段，尾空段无害"); + } }