feat(mq-starter-redis-stream): 启动期 PING 探测 Redis 连通性(①② fail-fast)
This commit is contained in:
@@ -0,0 +1,61 @@
|
|||||||
|
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)则回退 unknown,不再抛新异常。 */
|
||||||
|
private String target() {
|
||||||
|
try {
|
||||||
|
if (connectionFactory instanceof LettuceConnectionFactory) {
|
||||||
|
LettuceConnectionFactory lf = (LettuceConnectionFactory) connectionFactory;
|
||||||
|
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,56 @@
|
|||||||
|
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.RedisConnection;
|
||||||
|
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||||
|
|
||||||
|
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 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());
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user