Compare commits
5 Commits
bd2e733d98
...
886c02ff96
| Author | SHA1 | Date | |
|---|---|---|---|
| 886c02ff96 | |||
| 0c6d37842e | |||
| d7ef8b5626 | |||
| 46b0c13462 | |||
| 5c4a29e1e1 |
@@ -0,0 +1,51 @@
|
|||||||
|
package com.njcn.mq.autoconfig;
|
||||||
|
|
||||||
|
import com.njcn.mq.spi.MqDriver;
|
||||||
|
import org.springframework.beans.factory.ObjectProvider;
|
||||||
|
import org.springframework.beans.factory.SmartInitializingSingleton;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
import org.springframework.core.env.Environment;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* ③ 通用校验:显式配了 {@code mq.type},却没有装配任何 {@link MqDriver},
|
||||||
|
* 说明忘引对应 starter 或 mq.type 拼错 —— 启动失败并给中文提示。
|
||||||
|
*
|
||||||
|
* <p>为何“有没有 MqDriver”就够:两个 driver 的 AutoConfiguration 都用
|
||||||
|
* {@code @ConditionalOnProperty(name="mq.type", havingValue=...)} 按 mq.type 条件装配,
|
||||||
|
* 因此“存在 MqDriver bean”已等价于“配的 mq.type 有对应 driver”。
|
||||||
|
*
|
||||||
|
* <p>仅当显式配了 {@code mq.type} 时装配;没配则完全不介入(存量零影响)。
|
||||||
|
* 放在 core 是因为:redis/rocketmq 包没引入时其模块内的类自己都不加载,无法自证缺失;
|
||||||
|
* 只有 core 有“到底有没有 MqDriver”的全局视角。
|
||||||
|
*/
|
||||||
|
@Configuration
|
||||||
|
@ConditionalOnProperty(name = "mq.type")
|
||||||
|
public class MqDriverPresenceAutoConfiguration implements SmartInitializingSingleton {
|
||||||
|
|
||||||
|
private final ObjectProvider<MqDriver> drivers;
|
||||||
|
private final Environment env;
|
||||||
|
|
||||||
|
public MqDriverPresenceAutoConfiguration(ObjectProvider<MqDriver> drivers, Environment env) {
|
||||||
|
this.drivers = drivers;
|
||||||
|
this.env = env;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void afterSingletonsInstantiated() {
|
||||||
|
if (!MqStartupChecks.enabled(env)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
String configured = env.getProperty("mq.type");
|
||||||
|
// 用 stream() 而非 getIfAvailable():同时引多个 driver 包时不会抛 NoUniqueBeanDefinitionException
|
||||||
|
if (drivers.stream().findAny().isPresent()) {
|
||||||
|
return; // 有 driver 装配,放行
|
||||||
|
}
|
||||||
|
throw new IllegalStateException(
|
||||||
|
"MQ 启动失败:已配置 mq.type=" + configured + ",但没有装配任何 MqDriver。\n"
|
||||||
|
+ "请检查:\n"
|
||||||
|
+ " 1) 是否引入对应 starter 依赖(redis-stream 需 mq-starter-redis-stream + spring-boot-starter-data-redis;\n"
|
||||||
|
+ " rocketmq 需 mq-starter-rocketmq + rocketmq starter);\n"
|
||||||
|
+ " 2) mq.type 是否拼写正确(支持:redis-stream / rocketmq)。");
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,22 @@
|
|||||||
|
package com.njcn.mq.autoconfig;
|
||||||
|
|
||||||
|
import org.springframework.core.env.Environment;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* MQ 启动校验的公共工具:统一读取逃生开关 {@code mq.startup-check.enabled}(默认 true)。
|
||||||
|
*
|
||||||
|
* <p>只接受严格的 {@code true}/{@code false}(忽略大小写、去首尾空白)。仅当显式配为
|
||||||
|
* {@code false} 时才关闭校验;缺省或任何非法值(如 {@code disabled})都按默认 {@code true}
|
||||||
|
* 继续校验——避免逃生开关自身因非法布尔值在启动期抛类型转换异常,反而把启动炸掉。
|
||||||
|
*/
|
||||||
|
public final class MqStartupChecks {
|
||||||
|
|
||||||
|
private MqStartupChecks() {
|
||||||
|
}
|
||||||
|
|
||||||
|
/** @return true 表示应执行启动校验;false 表示使用者主动关闭。 */
|
||||||
|
public static boolean enabled(Environment env) {
|
||||||
|
String value = env.getProperty("mq.startup-check.enabled");
|
||||||
|
return value == null || !"false".equalsIgnoreCase(value.trim());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,2 +1,3 @@
|
|||||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||||
com.njcn.mq.autoconfig.MqCoreAutoConfiguration
|
com.njcn.mq.autoconfig.MqCoreAutoConfiguration,\
|
||||||
|
com.njcn.mq.autoconfig.MqDriverPresenceAutoConfiguration
|
||||||
|
|||||||
@@ -0,0 +1,53 @@
|
|||||||
|
package com.njcn.mq.autoconfig;
|
||||||
|
|
||||||
|
import com.njcn.mq.spi.MqDriver;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||||
|
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
|
||||||
|
class MqDriverPresenceAutoConfigurationTest {
|
||||||
|
|
||||||
|
private final ApplicationContextRunner runner = new ApplicationContextRunner()
|
||||||
|
.withConfiguration(AutoConfigurations.of(MqDriverPresenceAutoConfiguration.class));
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void noMqType_doesNotFail() {
|
||||||
|
runner.run(ctx -> assertThat(ctx).hasNotFailed());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void mqTypeButNoDriver_fails() {
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream")
|
||||||
|
.run(ctx -> {
|
||||||
|
assertThat(ctx).hasFailed();
|
||||||
|
assertThat(ctx).getFailure().hasMessageContaining("redis-stream");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void mqTypeWithDriver_ok() {
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream")
|
||||||
|
.withBean("someDriver", MqDriver.class, () -> mock(MqDriver.class))
|
||||||
|
.run(ctx -> assertThat(ctx).hasNotFailed());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void illegalEnabledValue_stillChecks_insteadOfCrashing() {
|
||||||
|
// 逃生开关填非法布尔值不应炸在类型转换上,而应按默认继续校验:
|
||||||
|
// 此处无 driver,应走到中文③提示,而不是 Spring 的「Invalid boolean」类型转换异常。
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream", "mq.startup-check.enabled=disabled")
|
||||||
|
.run(ctx -> {
|
||||||
|
assertThat(ctx).hasFailed();
|
||||||
|
assertThat(ctx).getFailure().hasMessageContaining("没有装配任何 MqDriver");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void disabled_skipsCheck() {
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream", "mq.startup-check.enabled=false")
|
||||||
|
.run(ctx -> assertThat(ctx).hasNotFailed());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,69 @@
|
|||||||
|
package com.njcn.mq.driver.redis;
|
||||||
|
|
||||||
|
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.data.redis.connection.RedisConnection;
|
||||||
|
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||||
|
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* ①② redis 探测:mq.type=redis-stream 时,在 bean 初始化阶段主动 PING 一次 Redis。
|
||||||
|
*
|
||||||
|
* <p>lettuce 默认懒连接(启动不真正连、运行时才连);此处主动 PING,连不上就 fail-fast。
|
||||||
|
*/
|
||||||
|
@Configuration
|
||||||
|
@ConditionalOnProperty(name = "mq.type", havingValue = "redis-stream")
|
||||||
|
public class RedisStreamStartupProbeAutoConfiguration implements InitializingBean {
|
||||||
|
|
||||||
|
private final RedisConnectionFactory connectionFactory;
|
||||||
|
private final Environment env;
|
||||||
|
|
||||||
|
public RedisStreamStartupProbeAutoConfiguration(RedisConnectionFactory connectionFactory, Environment env) {
|
||||||
|
this.connectionFactory = connectionFactory;
|
||||||
|
this.env = env;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void afterPropertiesSet() {
|
||||||
|
if (!MqStartupChecks.enabled(env)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
RedisConnection conn = connectionFactory.getConnection();
|
||||||
|
try {
|
||||||
|
conn.ping();
|
||||||
|
} finally {
|
||||||
|
conn.close(); // 探测用连接必须释放,避免泄漏
|
||||||
|
}
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw new IllegalStateException(
|
||||||
|
"MQ 启动失败:mq.type=redis-stream,无法连接 Redis(" + target() + ")。原因:"
|
||||||
|
+ e.getMessage()
|
||||||
|
+ "。请检查 spring.data.redis.host/port/password,或确认 Redis 服务已启动。", e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 尽力取 host:port 用于提示;取不到(非 Lettuce / cluster / sentinel)则回退到有意义的文案,不再抛新异常。 */
|
||||||
|
private String target() {
|
||||||
|
try {
|
||||||
|
if (connectionFactory instanceof LettuceConnectionFactory) {
|
||||||
|
LettuceConnectionFactory lf = (LettuceConnectionFactory) connectionFactory;
|
||||||
|
// cluster/sentinel 模式下 getHostName()/getPort() 返回的是 standalone 默认值 localhost:6379
|
||||||
|
//(使用者从未配置过的假地址),会把排障引偏,故回退到指向真实配置项的文案。
|
||||||
|
if (lf.isClusterAware()) {
|
||||||
|
return "集群模式,见 spring.data.redis.cluster.nodes";
|
||||||
|
}
|
||||||
|
if (lf.isRedisSentinelAware()) {
|
||||||
|
return "哨兵模式,见 spring.data.redis.sentinel";
|
||||||
|
}
|
||||||
|
return lf.getHostName() + ":" + lf.getPort();
|
||||||
|
}
|
||||||
|
} catch (Exception ignore) {
|
||||||
|
// 取地址失败不能影响主流程
|
||||||
|
}
|
||||||
|
return "unknown";
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,2 +1,3 @@
|
|||||||
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
|
||||||
com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration
|
com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration,\
|
||||||
|
com.njcn.mq.driver.redis.RedisStreamStartupProbeAutoConfiguration
|
||||||
|
|||||||
@@ -0,0 +1,100 @@
|
|||||||
|
package com.njcn.mq.driver.redis;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.boot.autoconfigure.AutoConfigurations;
|
||||||
|
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
|
||||||
|
import org.springframework.data.redis.connection.RedisClusterConfiguration;
|
||||||
|
import org.springframework.data.redis.connection.RedisConnection;
|
||||||
|
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||||
|
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
|
||||||
|
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||||
|
|
||||||
|
import static org.assertj.core.api.Assertions.assertThat;
|
||||||
|
import static org.mockito.Mockito.mock;
|
||||||
|
import static org.mockito.Mockito.verify;
|
||||||
|
import static org.mockito.Mockito.when;
|
||||||
|
|
||||||
|
class RedisStreamStartupProbeAutoConfigurationTest {
|
||||||
|
|
||||||
|
private final ApplicationContextRunner runner = new ApplicationContextRunner()
|
||||||
|
.withConfiguration(AutoConfigurations.of(RedisStreamStartupProbeAutoConfiguration.class));
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void pingOk_startsUp_andReleasesConnection() {
|
||||||
|
RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
|
||||||
|
RedisConnection conn = mock(RedisConnection.class);
|
||||||
|
when(factory.getConnection()).thenReturn(conn);
|
||||||
|
when(conn.ping()).thenReturn("PONG");
|
||||||
|
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream")
|
||||||
|
.withBean(RedisConnectionFactory.class, () -> factory)
|
||||||
|
.run(ctx -> assertThat(ctx).hasNotFailed());
|
||||||
|
|
||||||
|
verify(conn).close();
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void pingFails_startupFails() {
|
||||||
|
RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
|
||||||
|
when(factory.getConnection()).thenThrow(new RuntimeException("Connection refused"));
|
||||||
|
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream")
|
||||||
|
.withBean(RedisConnectionFactory.class, () -> factory)
|
||||||
|
.run(ctx -> {
|
||||||
|
assertThat(ctx).hasFailed();
|
||||||
|
assertThat(ctx).getFailure().hasMessageContaining("Redis");
|
||||||
|
assertThat(ctx).getFailure().hasMessageContaining("Connection refused");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void pingFails_messageContainsHostPort() {
|
||||||
|
// 用真实 LettuceConnectionFactory(standalone) 覆盖 target() 的 host:port 提示分支;
|
||||||
|
// override getConnection() 让探测立即失败,无需真实 Redis / 网络。
|
||||||
|
LettuceConnectionFactory factory =
|
||||||
|
new LettuceConnectionFactory(new RedisStandaloneConfiguration("myhost", 1234)) {
|
||||||
|
@Override
|
||||||
|
public RedisConnection getConnection() {
|
||||||
|
throw new RuntimeException("Connection refused");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream")
|
||||||
|
.withBean(RedisConnectionFactory.class, () -> factory)
|
||||||
|
.run(ctx -> {
|
||||||
|
assertThat(ctx).hasFailed();
|
||||||
|
assertThat(ctx).getFailure().hasMessageContaining("myhost:1234");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void clusterMode_avoidsFakeLocalhostInMessage() {
|
||||||
|
// cluster 模式下 getHostName()/getPort() 会返回 standalone 默认值 localhost:6379(假地址);
|
||||||
|
// 修复后应回退到指向 cluster 配置项的文案,而不是打印使用者从未配过的 localhost:6379。
|
||||||
|
LettuceConnectionFactory factory =
|
||||||
|
new LettuceConnectionFactory(
|
||||||
|
new RedisClusterConfiguration(java.util.Collections.singletonList("127.0.0.1:7000"))) {
|
||||||
|
@Override
|
||||||
|
public RedisConnection getConnection() {
|
||||||
|
throw new RuntimeException("Connection refused");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream")
|
||||||
|
.withBean(RedisConnectionFactory.class, () -> factory)
|
||||||
|
.run(ctx -> {
|
||||||
|
assertThat(ctx).hasFailed();
|
||||||
|
assertThat(ctx).getFailure().hasMessageContaining("集群模式");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void disabled_skipsProbe() {
|
||||||
|
RedisConnectionFactory factory = mock(RedisConnectionFactory.class);
|
||||||
|
when(factory.getConnection()).thenThrow(new RuntimeException("should not be called"));
|
||||||
|
|
||||||
|
runner.withPropertyValues("mq.type=redis-stream", "mq.startup-check.enabled=false")
|
||||||
|
.withBean(RedisConnectionFactory.class, () -> factory)
|
||||||
|
.run(ctx -> assertThat(ctx).hasNotFailed());
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -3,6 +3,17 @@
|
|||||||
<modelVersion>4.0.0</modelVersion>
|
<modelVersion>4.0.0</modelVersion>
|
||||||
<parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
|
<parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
|
||||||
<artifactId>mq-starter-rocketmq</artifactId>
|
<artifactId>mq-starter-rocketmq</artifactId>
|
||||||
|
<dependencyManagement>
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.springframework.boot</groupId>
|
||||||
|
<artifactId>spring-boot-dependencies</artifactId>
|
||||||
|
<version>2.3.12.RELEASE</version>
|
||||||
|
<type>pom</type>
|
||||||
|
<scope>import</scope>
|
||||||
|
</dependency>
|
||||||
|
</dependencies>
|
||||||
|
</dependencyManagement>
|
||||||
<dependencies>
|
<dependencies>
|
||||||
<dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
|
<dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
|
||||||
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version></dependency>
|
<dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version></dependency>
|
||||||
|
|||||||
@@ -0,0 +1,85 @@
|
|||||||
|
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.boot.context.properties.bind.Binder;
|
||||||
|
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;
|
||||||
|
}
|
||||||
|
// 与 driver 同源取值:driver 走 RocketMQProperties(@ConfigurationProperties)的 getNameServer(),
|
||||||
|
// 享受 Spring relaxed binding;这里也用 Binder,避免 env.getProperty 精确键查找漏掉
|
||||||
|
// camelCase(rocketmq.nameServer)等 relaxed 变体,从而把「其实配好了」误判成「未配置」而错误 fail-fast。
|
||||||
|
String nameServer = Binder.get(env).bind("rocketmq.name-server", String.class).orElse(null);
|
||||||
|
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,70 @@
|
|||||||
|
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 reachableNameServer_camelCaseKey_startsUp() throws Exception {
|
||||||
|
// driver 通过 RocketMQProperties 走 relaxed binding,camelCase 的 rocketmq.nameServer 也能生效;
|
||||||
|
// 探测取值必须与之同源,否则会把配好的 NameServer 误判为未配置而错误 fail-fast。
|
||||||
|
try (ServerSocket server = new ServerSocket(0)) {
|
||||||
|
int port = server.getLocalPort();
|
||||||
|
runner.withPropertyValues(
|
||||||
|
"mq.type=rocketmq",
|
||||||
|
"rocketmq.nameServer=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