From 0b950b3c35e23266d918a2667e1df2c962a8dd79 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Mon, 29 Jun 2026 11:16:02 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(message-boot):=20=E6=8E=A5=E5=85=A5=20?= =?UTF-8?q?mq-starter=20=E9=80=9A=E7=94=A8=20MQ=20=E6=8A=BD=E8=B1=A1?= =?UTF-8?q?=E5=B1=82=EF=BC=88Task=209=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 FrontDataMqListener：@MqListener(LN_Topic)，仅 redis-stream 模式激活 - 新增 RecallMqSender：MqTemplate.send 下行下发 Recall 消息 - FrontDataConsumer 加 @ConditionalOnProperty(mq.type=rocketmq, matchIfMissing=true) 互斥保护 - 新增 MqSliceWiringTest：4 个 ApplicationContextRunner 切片测试，全部通过 - 关键修复：用 withUserConfiguration(依赖顺序列举) 替代 AutoConfigurations.of()， 避免字母序排序导致 MqCoreAutoConfiguration 在 driver autoconfig 前处理， @ConditionalOnBean(MqDriver.class) 因此失败 Co-Authored-By: Claude Sonnet 4.6 --- message/message-boot/pom.xml | 22 ++ .../message/consumer/FrontDataConsumer.java | 2 + .../njcn/message/mq/FrontDataMqListener.java | 23 ++ .../com/njcn/message/mq/RecallMqSender.java | 19 ++ .../njcn/message/mq/MqSliceWiringTest.java | 200 ++++++++++++++++++ 5 files changed, 266 insertions(+) create mode 100644 message/message-boot/src/main/java/com/njcn/message/mq/FrontDataMqListener.java create mode 100644 message/message-boot/src/main/java/com/njcn/message/mq/RecallMqSender.java create mode 100644 message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java diff --git a/message/message-boot/pom.xml b/message/message-boot/pom.xml index a5e3fc3..0e5f597 100644 --- a/message/message-boot/pom.xml +++ b/message/message-boot/pom.xml @@ -100,12 +100,34 @@ spring-boot-starter-websocket 2.7.12 + + com.njcn + mq-spring-boot-starter + 1.0.0-SNAPSHOT + + + org.springframework.boot + spring-boot-starter-test + 2.7.12 + test + + + junit + junit + 4.13.2 + test + messageboot + + org.apache.maven.plugins + maven-surefire-plugin + 2.22.2 + org.apache.maven.plugins maven-compiler-plugin diff --git a/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumer.java b/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumer.java index a60a78c..f8d30fd 100644 --- a/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumer.java +++ b/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumer.java @@ -39,6 +39,8 @@ import java.util.concurrent.TimeUnit; * @createTime 2023/8/11 15:32 */ @Component +@org.springframework.boot.autoconfigure.condition.ConditionalOnProperty( + name = "mq.type", havingValue = "rocketmq", matchIfMissing = true) @RocketMQMessageListener( topic = "LN_Topic", consumerGroup = "ln_consumer", diff --git a/message/message-boot/src/main/java/com/njcn/message/mq/FrontDataMqListener.java b/message/message-boot/src/main/java/com/njcn/message/mq/FrontDataMqListener.java new file mode 100644 index 0000000..2219940 --- /dev/null +++ b/message/message-boot/src/main/java/com/njcn/message/mq/FrontDataMqListener.java @@ -0,0 +1,23 @@ +package com.njcn.message.mq; + +import com.njcn.message.messagedto.MessageDataDTO; +import com.njcn.mq.annotation.MqListener; +import com.njcn.stat.api.MessAnalysisFeignClient; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; +import java.util.Collections; + +@Component +@ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream") +public class FrontDataMqListener { + + @Resource + private MessAnalysisFeignClient messAnalysisFeignClient; + + @MqListener(topic = "LN_Topic", group = "ln_consumer") + public void onFrontData(MessageDataDTO msg) { + messAnalysisFeignClient.analysis(Collections.singletonList(msg)); + } +} diff --git a/message/message-boot/src/main/java/com/njcn/message/mq/RecallMqSender.java b/message/message-boot/src/main/java/com/njcn/message/mq/RecallMqSender.java new file mode 100644 index 0000000..22998e2 --- /dev/null +++ b/message/message-boot/src/main/java/com/njcn/message/mq/RecallMqSender.java @@ -0,0 +1,19 @@ +package com.njcn.message.mq; + +import com.njcn.message.constant.BusinessTopic; +import com.njcn.message.message.RecallMessage; +import com.njcn.mq.core.MqTemplate; +import org.springframework.stereotype.Component; + +import javax.annotation.Resource; + +@Component +public class RecallMqSender { + + @Resource + private MqTemplate mqTemplate; + + public void send(RecallMessage message, String nodeId) { + mqTemplate.send(nodeId + "_" + BusinessTopic.RECALL_TOPIC, message); + } +} diff --git a/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java b/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java new file mode 100644 index 0000000..d7157b8 --- /dev/null +++ b/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java @@ -0,0 +1,200 @@ +package com.njcn.message.mq; + +import com.alibaba.fastjson.JSON; +import com.njcn.message.message.RecallMessage; +import com.njcn.message.messagedto.MessageDataDTO; +import com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration; +import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; +import com.njcn.mq.driver.redis.RedisStreamMqDriver; +import com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration; +import com.njcn.mq.driver.rocketmq.RocketMqDriver; +import com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration; +import com.njcn.mq.spi.MqDriver; +import com.njcn.stat.api.MessAnalysisFeignClient; +import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; +import org.apache.rocketmq.spring.core.RocketMQTemplate; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamOffset; +import org.springframework.data.redis.connection.stream.StreamReadOptions; +import org.springframework.data.redis.connection.stream.StreamRecords; +import org.springframework.data.redis.core.StringRedisTemplate; + +import java.net.InetSocketAddress; +import java.net.Socket; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.timeout; +import static org.mockito.Mockito.verify; + +/** + * MQ starter 切片接入验证测试（ApplicationContextRunner，不启动全应用）。 + * 真实 Redis 要求：localhost:6379（docker redis-stream-it）。 + * 无 Redis 时，依赖 Redis 的用例通过 Assumptions 跳过。 + * + *

注意：withConfiguration(AutoConfigurations.of(...)) 会对 autoconfig 类名做字母序排序， + * 导致 MqCoreAutoConfiguration 在 driver autoconfig 之前处理，@ConditionalOnBean(MqDriver.class) + * 因此失败。改用 withUserConfiguration(...) 按依赖顺序显式列出，避免排序问题。 + */ +public class MqSliceWiringTest { + + private static boolean redisAvailable() { + try { + Socket s = new Socket(); + s.connect(new InetSocketAddress("localhost", 6379), 500); + s.close(); + return true; + } catch (Exception e) { + return false; + } + } + + /** + * redis-stream 模式的基础 runner。 + * 按依赖顺序列出：Redis → Phase1 stream → redis driver → rocket driver(inactive) → mq core → 业务 bean + * 这样 @ConditionalOnBean(MqDriver.class) 在 MqCoreAutoConfiguration 被解析时已能找到 MqDriver 定义。 + */ + private ApplicationContextRunner redisStreamRunner() { + return new ApplicationContextRunner() + .withUserConfiguration( + RedisAutoConfiguration.class, // 1. StringRedisTemplate + RedisStreamAutoConfiguration.class, // 2. StreamKeyBuilder, RedisStreamProperties + RedisStreamMqDriverAutoConfiguration.class, // 3. MqDriver(RedisStreamMqDriver) + RocketMqDriverAutoConfiguration.class, // 4. inactive for redis-stream + MqCoreAutoConfiguration.class, // 5. MqTemplate, MqListenerRegistry + FrontDataMqListener.class, // 6. 上行 listener + RecallMqSender.class // 7. 下行 sender + ) + .withBean(MessAnalysisFeignClient.class, + () -> mock(MessAnalysisFeignClient.class)) + .withPropertyValues( + "mq.type=redis-stream", + "spring.redis.host=localhost", + "spring.redis.port=6379" + ); + } + + // ------------------------------------------------------------------ + // 1. Driver 二选一 + // ------------------------------------------------------------------ + + @Test + void testRedisStreamDriverSelected() { + Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping"); + redisStreamRunner().run(ctx -> { + assertThat(ctx).hasSingleBean(MqDriver.class); + assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RedisStreamMqDriver.class); + }); + } + + @Test + void testRocketMqDriverSelected() { + // rocket 测试不依赖真实 Redis，仅引入 rocket 相关 + core + new ApplicationContextRunner() + .withUserConfiguration( + RocketMqDriverAutoConfiguration.class, // 1. MqDriver(RocketMqDriver) + RedisStreamMqDriverAutoConfiguration.class, // 2. inactive for rocketmq + MqCoreAutoConfiguration.class, // 3. MqTemplate, MqListenerRegistry + FrontDataMqListener.class, // 4. inactive for rocketmq + RecallMqSender.class // 5. sender + ) + .withBean(MessAnalysisFeignClient.class, + () -> mock(MessAnalysisFeignClient.class)) + .withBean(RocketMQTemplate.class, + () -> mock(RocketMQTemplate.class)) + .withBean(RocketMQProperties.class, RocketMQProperties::new) + .withPropertyValues("mq.type=rocketmq") + .run(ctx -> { + assertThat(ctx).hasSingleBean(MqDriver.class); + assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RocketMqDriver.class); + }); + } + + // ------------------------------------------------------------------ + // 2. 上行（需要真实 Redis） + // 向 LN_Topic stream XADD 一条消息，验证 FrontDataMqListener 触发了 + // messAnalysisFeignClient.analysis(...) + // ------------------------------------------------------------------ + + @Test + void testUpstreamFrontData() { + Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping"); + redisStreamRunner().run(ctx -> { + StringRedisTemplate redis = ctx.getBean(StringRedisTemplate.class); + MessAnalysisFeignClient mockClient = ctx.getBean(MessAnalysisFeignClient.class); + + // 构造 RedisStreamMqDriver 消费时期望的字段格式 + // enc=plain → body 即为原始 JSON（参见 MessageCodec.encodeBody(json, false)） + // stream key = StreamKeyBuilder.buildStream("LN_Topic") = "LN_Topic" + // （默认 envIsolation=false，无 env 前缀） + MessageDataDTO dto = new MessageDataDTO(); + dto.setDataType(1); + String json = JSON.toJSONString(dto); + + Map fields = new HashMap(); + fields.put("key", "test-key-upstream-1"); + fields.put("tag", ""); + fields.put("enc", "plain"); + fields.put("body", json); + fields.put("ts", String.valueOf(System.currentTimeMillis())); + + // XADD 到 LN_Topic（MqListenerRegistry.start() 已在 context 启动时订阅该 stream） + redis.opsForStream().add( + StreamRecords.newRecord().in("LN_Topic").ofMap(fields) + ); + + // 等待消费线程处理（blockMs 默认 2000ms，timeout=5000ms 足够） + verify(mockClient, timeout(5000)).analysis(anyList()); + }); + } + + // ------------------------------------------------------------------ + // 3. 下行（需要真实 Redis） + // 通过 RecallMqSender 发送 RecallMessage，验证 stream 内容正确 + // ------------------------------------------------------------------ + + @Test + void testDownstreamRecall() { + Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping"); + redisStreamRunner().run(ctx -> { + StringRedisTemplate redis = ctx.getBean(StringRedisTemplate.class); + RecallMqSender sender = ctx.getBean(RecallMqSender.class); + + // 清理，确保只有本次发送的消息 + redis.delete("n1_recall_Topic"); + + RecallMessage msg = new RecallMessage(); + msg.setGuid("g1"); + // MqTemplate.send("n1_recall_Topic", msg) → RedisStreamMqDriver XADD + // stream key = StreamKeyBuilder.buildStream("n1_recall_Topic") = "n1_recall_Topic" + sender.send(msg, "n1"); + + Long size = redis.opsForStream().size("n1_recall_Topic"); + assertThat(size).isGreaterThanOrEqualTo(1L); + + @SuppressWarnings("unchecked") + List> records = + redis.opsForStream().read( + StreamReadOptions.empty().count(1L), + StreamOffset.create("n1_recall_Topic", ReadOffset.from("0-0")) + ); + assertThat(records).isNotEmpty(); + + Map firstFields = records.get(0).getValue(); + // DefaultMqTemplate.send(topic, payload) 使用 encodeBody(json, false) → enc="plain" + // 因此 body 即为原始 JSON，无需 gzip 解压 + String body = (String) firstFields.get("body"); + RecallMessage result = JSON.parseObject(body, RecallMessage.class); + assertThat(result.getGuid()).isEqualTo("g1"); + }); + } +} From 79374ea6eca537222749815fc81f85e4cae2069a Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Mon, 29 Jun 2026 12:01:49 +0800 Subject: [PATCH 2/2] =?UTF-8?q?test(message):=20=E5=88=87=E7=89=87=20wirin?= =?UTF-8?q?g=20=E6=B5=8B=E8=AF=95=E6=94=B9=E7=94=A8=E7=9C=9F=E5=AE=9E=20au?= =?UTF-8?q?toconfig=20=E9=A1=BA=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MqSliceWiringTest 从 withUserConfiguration 手动排序改为 AutoConfigurations.of （等价真实 boot 的 AutoConfigurationSorter），并断言 MqTemplate / MqListenerRegistry 在真实顺序下装配，防止 core 装配顺序回归；配合 mq-starter driver autoconfig 的 @AutoConfigureBefore 修复。同时去掉 driver 选择用例多余的 Redis 守卫。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../njcn/message/mq/MqSliceWiringTest.java | 88 +++++++++++-------- 1 file changed, 49 insertions(+), 39 deletions(-) diff --git a/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java b/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java index d7157b8..de2d75f 100644 --- a/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java +++ b/message/message-boot/src/test/java/com/njcn/message/mq/MqSliceWiringTest.java @@ -5,6 +5,8 @@ import com.njcn.message.message.RecallMessage; import com.njcn.message.messagedto.MessageDataDTO; import com.njcn.middle.stream.autoconfig.RedisStreamAutoConfiguration; import com.njcn.mq.autoconfig.MqCoreAutoConfiguration; +import com.njcn.mq.container.MqListenerRegistry; +import com.njcn.mq.core.MqTemplate; import com.njcn.mq.driver.redis.RedisStreamMqDriver; import com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration; import com.njcn.mq.driver.rocketmq.RocketMqDriver; @@ -15,6 +17,7 @@ import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.AutoConfigurations; import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration; import org.springframework.boot.test.context.runner.ApplicationContextRunner; import org.springframework.data.redis.connection.stream.MapRecord; @@ -38,12 +41,16 @@ import static org.mockito.Mockito.verify; /** * MQ starter 切片接入验证测试（ApplicationContextRunner，不启动全应用）。 - * 真实 Redis 要求：localhost:6379（docker redis-stream-it）。 - * 无 Redis 时，依赖 Redis 的用例通过 Assumptions 跳过。 + * 真实 Redis 要求：localhost:6379（docker redis-stream-it）；仅上/下行收发用例需要， + * 纯装配用例不需要（Lettuce 懒连接，不触发实际连接）。 * - *

注意：withConfiguration(AutoConfigurations.of(...)) 会对 autoconfig 类名做字母序排序， - * 导致 MqCoreAutoConfiguration 在 driver autoconfig 之前处理，@ConditionalOnBean(MqDriver.class) - * 因此失败。改用 withUserConfiguration(...) 按依赖顺序显式列出，避免排序问题。 + *

装配一律用 {@code withConfiguration(AutoConfigurations.of(...))}，让 Spring 按 + * 真实 autoconfig 顺序（AutoConfigurationSorter）处理，与生产环境一致。以此 + * 验证 {@link MqCoreAutoConfiguration} 的 {@code @ConditionalOnBean(MqDriver)} 能在 + * 真实顺序下拿到 driver bean —— 这依赖 driver autoconfig 上的 + * {@code @AutoConfigureBefore(MqCoreAutoConfiguration.class)}。若缺该保障，按类名字母序 + * {@code com.njcn.mq.autoconfig.*} 先于 {@code com.njcn.mq.driver.*}，core 会在 driver + * bean 注册前求值条件失败，导致 MqTemplate / MqListenerRegistry 不装配。 */ public class MqSliceWiringTest { @@ -58,64 +65,67 @@ public class MqSliceWiringTest { } } - /** - * redis-stream 模式的基础 runner。 - * 按依赖顺序列出：Redis → Phase1 stream → redis driver → rocket driver(inactive) → mq core → 业务 bean - * 这样 @ConditionalOnBean(MqDriver.class) 在 MqCoreAutoConfiguration 被解析时已能找到 MqDriver 定义。 - */ + /** redis-stream 模式：真实 autoconfig 顺序装 5 个 autoconfig + 业务 bean（上/下行用）。 */ private ApplicationContextRunner redisStreamRunner() { return new ApplicationContextRunner() - .withUserConfiguration( - RedisAutoConfiguration.class, // 1. StringRedisTemplate - RedisStreamAutoConfiguration.class, // 2. StreamKeyBuilder, RedisStreamProperties - RedisStreamMqDriverAutoConfiguration.class, // 3. MqDriver(RedisStreamMqDriver) - RocketMqDriverAutoConfiguration.class, // 4. inactive for redis-stream - MqCoreAutoConfiguration.class, // 5. MqTemplate, MqListenerRegistry - FrontDataMqListener.class, // 6. 上行 listener - RecallMqSender.class // 7. 下行 sender - ) + .withConfiguration(AutoConfigurations.of( + RedisAutoConfiguration.class, + RedisStreamAutoConfiguration.class, + RedisStreamMqDriverAutoConfiguration.class, + RocketMqDriverAutoConfiguration.class, + MqCoreAutoConfiguration.class)) + .withUserConfiguration(FrontDataMqListener.class, RecallMqSender.class) .withBean(MessAnalysisFeignClient.class, () -> mock(MessAnalysisFeignClient.class)) .withPropertyValues( "mq.type=redis-stream", "spring.redis.host=localhost", - "spring.redis.port=6379" - ); + "spring.redis.port=6379"); } // ------------------------------------------------------------------ - // 1. Driver 二选一 + // 1. 装配：mq.type 二选一 + core bean 在真实 autoconfig 顺序下装配 // ------------------------------------------------------------------ @Test void testRedisStreamDriverSelected() { - Assumptions.assumeTrue(redisAvailable(), "Redis not available at localhost:6379, skipping"); - redisStreamRunner().run(ctx -> { - assertThat(ctx).hasSingleBean(MqDriver.class); - assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RedisStreamMqDriver.class); - }); + // 纯装配检查，不依赖真实 Redis（Lettuce 懒连接）。 + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of( + RedisAutoConfiguration.class, + RedisStreamAutoConfiguration.class, + RedisStreamMqDriverAutoConfiguration.class, + RocketMqDriverAutoConfiguration.class, + MqCoreAutoConfiguration.class)) + .withPropertyValues( + "mq.type=redis-stream", + "spring.redis.host=localhost", + "spring.redis.port=6379") + .run(ctx -> { + assertThat(ctx).hasSingleBean(MqDriver.class); + assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RedisStreamMqDriver.class); + // 真实顺序下 core 的 @ConditionalOnBean(MqDriver) 必须命中 + assertThat(ctx).hasSingleBean(MqTemplate.class); + assertThat(ctx).hasSingleBean(MqListenerRegistry.class); + }); } @Test void testRocketMqDriverSelected() { - // rocket 测试不依赖真实 Redis，仅引入 rocket 相关 + core + // rocket 测试不依赖真实 Redis；不装 RedisAutoConfiguration，redis driver 因 mq.type 不匹配而 inactive。 new ApplicationContextRunner() - .withUserConfiguration( - RocketMqDriverAutoConfiguration.class, // 1. MqDriver(RocketMqDriver) - RedisStreamMqDriverAutoConfiguration.class, // 2. inactive for rocketmq - MqCoreAutoConfiguration.class, // 3. MqTemplate, MqListenerRegistry - FrontDataMqListener.class, // 4. inactive for rocketmq - RecallMqSender.class // 5. sender - ) - .withBean(MessAnalysisFeignClient.class, - () -> mock(MessAnalysisFeignClient.class)) - .withBean(RocketMQTemplate.class, - () -> mock(RocketMQTemplate.class)) + .withConfiguration(AutoConfigurations.of( + RedisStreamMqDriverAutoConfiguration.class, + RocketMqDriverAutoConfiguration.class, + MqCoreAutoConfiguration.class)) + .withBean(RocketMQTemplate.class, () -> mock(RocketMQTemplate.class)) .withBean(RocketMQProperties.class, RocketMQProperties::new) .withPropertyValues("mq.type=rocketmq") .run(ctx -> { assertThat(ctx).hasSingleBean(MqDriver.class); assertThat(ctx.getBean(MqDriver.class)).isInstanceOf(RocketMqDriver.class); + assertThat(ctx).hasSingleBean(MqTemplate.class); + assertThat(ctx).hasSingleBean(MqListenerRegistry.class); }); }