feat(mq-starter-rocketmq): 启动期 TCP 探测 NameServer 连通性(①② fail-fast)
This commit is contained in:
@@ -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 短超时连接探测。
|
||||||
|
*
|
||||||
|
* <p>不用 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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,2 +1,3 @@
|
|||||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||||
com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration
|
com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration,\
|
||||||
|
com.njcn.mq.driver.rocketmq.RocketMqStartupProbeAutoConfiguration
|
||||||
|
|||||||
@@ -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());
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user