From d7ef8b5626eef8f9d48e49c9dacddb5a6bdf72b1 Mon Sep 17 00:00:00 2001 From: hongawen <83944980@qq.com> Date: Thu, 9 Jul 2026 11:05:00 +0800 Subject: [PATCH] =?UTF-8?q?feat(mq-starter-rocketmq):=20=E5=90=AF=E5=8A=A8?= =?UTF-8?q?=E6=9C=9F=20TCP=20=E6=8E=A2=E6=B5=8B=20NameServer=20=E8=BF=9E?= =?UTF-8?q?=E9=80=9A=E6=80=A7=EF=BC=88=E2=91=A0=E2=91=A1=20fail-fast?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...RocketMqStartupProbeAutoConfiguration.java | 81 +++++++++++++++++++ .../main/resources/META-INF/spring.factories | 3 +- ...etMqStartupProbeAutoConfigurationTest.java | 57 +++++++++++++ 3 files changed, 140 insertions(+), 1 deletion(-) create mode 100644 mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java create mode 100644 mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java diff --git a/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java new file mode 100644 index 0000000..57b35a7 --- /dev/null +++ b/mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java @@ -0,0 +1,81 @@ +package com.njcn.mq.driver.rocketmq; + +import com.njcn.mq.autoconfig.MqStartupChecks; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.env.Environment; +import org.springframework.util.StringUtils; + +import java.net.InetSocketAddress; +import java.net.Socket; + +/** + * ①② rocketmq 探测：mq.type=rocketmq 时，对 NameServer 地址做 TCP 短超时连接探测。 + * + *

不用 DefaultMQProducer.start()（它不主动连 NameServer、探不准且更慢）； + * 改用 Socket.connect + 可配超时：任一地址可达即通过，全不通则 fail-fast。 + */ +@Configuration +@ConditionalOnProperty(name = "mq.type", havingValue = "rocketmq") +public class RocketMqStartupProbeAutoConfiguration implements InitializingBean { + + private final Environment env; + + public RocketMqStartupProbeAutoConfiguration(Environment env) { + this.env = env; + } + + @Override + public void afterPropertiesSet() { + if (!MqStartupChecks.enabled(env)) { + return; + } + String nameServer = env.getProperty("rocketmq.name-server"); + if (!StringUtils.hasText(nameServer)) { + throw new IllegalStateException( + "MQ 启动失败：mq.type=rocketmq，但未配置 rocketmq.name-server。" + + "请配置 NameServer 地址（如 127.0.0.1:9876）。"); + } + + int timeoutMs = env.getProperty("mq.startup-check.rocketmq.timeout-ms", Integer.class, 1000); + Exception last = null; + for (String raw : nameServer.split("[;,]")) { + String addr = raw.trim(); + if (addr.isEmpty()) { + continue; + } + int idx = addr.lastIndexOf(':'); + if (idx < 0) { + last = new IllegalArgumentException("地址缺少端口: " + addr); + continue; + } + String host = addr.substring(0, idx); + int port; + try { + port = Integer.parseInt(addr.substring(idx + 1).trim()); + } catch (NumberFormatException e) { + last = e; + continue; + } + Socket socket = new Socket(); + try { + socket.connect(new InetSocketAddress(host, port), timeoutMs); + return; // 任一地址可达即通过 + } catch (Exception e) { + last = e; + } finally { + try { + socket.close(); + } catch (Exception ignore) { + // 关闭探测 socket 失败可忽略 + } + } + } + + throw new IllegalStateException( + "MQ 启动失败：mq.type=rocketmq，无法连接 NameServer（" + nameServer + "）。原因：" + + (last == null ? "未知" : last.getMessage()) + + "。请检查 rocketmq.name-server，或确认 NameServer 已启动。", last); + } +} diff --git a/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories b/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories index e317b90..d52dae1 100644 --- a/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories +++ b/mq-starter-rocketmq/src/main/resources/META-INF/spring.factories @@ -1,2 +1,3 @@ org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ -com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration +com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration,\ +com.njcn.mq.driver.rocketmq.RocketMqStartupProbeAutoConfiguration diff --git a/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java new file mode 100644 index 0000000..7d5989e --- /dev/null +++ b/mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java @@ -0,0 +1,57 @@ +package com.njcn.mq.driver.rocketmq; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.autoconfigure.AutoConfigurations; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; + +import java.net.ServerSocket; + +import static org.assertj.core.api.Assertions.assertThat; + +class RocketMqStartupProbeAutoConfigurationTest { + + private final ApplicationContextRunner runner = new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(RocketMqStartupProbeAutoConfiguration.class)); + + @Test + void reachableNameServer_startsUp() throws Exception { + // 起一个真实监听端口模拟"可达的 NameServer"，不依赖真 rocketmq + try (ServerSocket server = new ServerSocket(0)) { + int port = server.getLocalPort(); + runner.withPropertyValues( + "mq.type=rocketmq", + "rocketmq.name-server=127.0.0.1:" + port) + .run(ctx -> assertThat(ctx).hasNotFailed()); + } + } + + @Test + void unreachableNameServer_startupFails() { + runner.withPropertyValues( + "mq.type=rocketmq", + "rocketmq.name-server=127.0.0.1:1", + "mq.startup-check.rocketmq.timeout-ms=300") + .run(ctx -> { + assertThat(ctx).hasFailed(); + assertThat(ctx).getFailure().hasMessageContaining("127.0.0.1:1"); + }); + } + + @Test + void missingNameServer_startupFails() { + runner.withPropertyValues("mq.type=rocketmq") + .run(ctx -> { + assertThat(ctx).hasFailed(); + assertThat(ctx).getFailure().hasMessageContaining("name-server"); + }); + } + + @Test + void disabled_skipsProbe() { + runner.withPropertyValues( + "mq.type=rocketmq", + "rocketmq.name-server=127.0.0.1:1", + "mq.startup-check.enabled=false") + .run(ctx -> assertThat(ctx).hasNotFailed()); + } +}