fix(mq-starter): 修复启动 fail-fast 校验 review 发现的缺陷（relaxed binding / 逃生开关 / redis 假地址）

workflow 多智能体 review 后修复：

- rocketmq 探测改用 Binder 读 rocketmq.name-server，与 driver（RocketMQProperties）同源、
  享受 relaxed binding，修复 camelCase(rocketmq.nameServer) 被误判为「未配置」而错误 fail-fast
- MqStartupChecks 逃生开关只认严格 false（忽略大小写/空白），缺省或非法值一律继续校验，
  不再因非法布尔值（如 enabled=disabled）抛类型转换异常炸掉启动
- redis 探测 target() 对 cluster/sentinel 回退到指向真实配置项的文案，
  不再假报使用者从未配过的 localhost:6379
- 补齐回归测试：camelCase 键 / 非法开关值继续校验 / host:port 提示分支 / cluster 回退
  （core 5、redis-stream 5、rocketmq 5，全绿）

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-core/pom.xml

@@ -8,6 +8,7 @@
         <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
         <dependency><groupId>cn.hutool</groupId><artifactId>hutool-all</artifactId><version>${hutool.version}</version></dependency>
         <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
+        <dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.30</version><scope>provided</scope></dependency>
         <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-configuration-processor</artifactId><version>${springboot.version}</version><optional>true</optional></dependency>
         <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
     </dependencies>

mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqDriverPresenceAutoConfiguration.java

@@ -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）。");
+    }
+}

mq-starter-core/src/main/java/com/njcn/mq/autoconfig/MqStartupChecks.java

@@ -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());
+    }
+}

mq-starter-core/src/main/java/com/njcn/mq/container/MqConsumeDispatcher.java

@@ -4,6 +4,7 @@ import com.njcn.mq.core.MqMessage;
 import com.njcn.mq.spi.MqConsumeErrorHandler;
 import com.njcn.mq.spi.MqErrorIdentity;
 import com.njcn.mq.spi.MqIdempotentStore;
+import lombok.extern.slf4j.Slf4j;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -13,8 +14,10 @@ import java.util.List;
  * 消费处理链：编排 去重 →（攒批）→ 业务消费 → 成功标记 / 异常落库。
  * 纯逻辑、不依赖 spring / redis，便于单测。重试本身交给 driver 的 MQ 原生重投。
  */
+@Slf4j
 public class MqConsumeDispatcher {
 
+    // >1 时攒批；注意：无 idle flush，低流量下未满批会停滞、进程重启丢未入库批（与 legacy 一致，适合高流量流）。
     private final int batchSize;
     private final boolean idempotent;
     private final MqIdempotentStore store;
@@ -38,13 +41,13 @@ public class MqConsumeDispatcher {
             try {
                 consumer.consume(Collections.singletonList(msg));
             } catch (Exception e) {
-                if (errorHandler != null) {
-                    errorHandler.onError(msg, MqErrorIdentity.SINGLE, e);
-                }
+                // errorHandler 自身失败不得跳过 markFail：否则去重标记停在 processing，
+                // reclaim 重投时会被去重挡掉并 ACK → 静默丢消息。
+                safeOnError(msg, MqErrorIdentity.SINGLE, e);
                 if (idempotent && store != null) {
                     store.markFail(msg.getKey());
                 }
-                throw e; // 交 driver 重投
+                throw e; // 抛原始异常，交 driver 重投
             }
             if (idempotent && store != null) {
                 store.markSuccess(msg.getKey());
@@ -66,10 +69,9 @@ public class MqConsumeDispatcher {
             } catch (Exception e) {
                 // 批量入库失败：消息已逐条 ACK，无法精确重投整批 → 对齐 legacy 不卡队列：
                 // 落库 + 吞掉，不抛、不标 success（去重 processing 标记自然过期）。
-                if (errorHandler != null) {
-                    for (MqMessage m : flush) {
-                        errorHandler.onError(m, MqErrorIdentity.SINGLE, e);
-                    }
+                log.warn("[mq] 批量消费失败，已落库并放弃该批 {} 条", flush.size(), e);
+                for (MqMessage m : flush) {
+                    safeOnError(m, MqErrorIdentity.SINGLE, e);
                 }
                 return;
             }
@@ -81,6 +83,16 @@ public class MqConsumeDispatcher {
         }
     }
 
+    /** 调用异常落库回调，并隔离其自身异常（落库是 best-effort，不得掩盖/中断主流程）。 */
+    private void safeOnError(MqMessage msg, MqErrorIdentity identity, Exception cause) {
+        if (errorHandler == null) return;
+        try {
+            errorHandler.onError(msg, identity, cause);
+        } catch (Exception ehEx) {
+            log.error("[mq] errorHandler 落库失败 key={}（不影响主流程）", msg.getKey(), ehEx);
+        }
+    }
+
     public static Builder builder() { return new Builder(); }
 
     public static class Builder {

mq-starter-core/src/main/java/com/njcn/mq/container/MqListenerRegistry.java

@@ -8,10 +8,12 @@ import com.njcn.mq.spi.MqDriver;
 import com.njcn.mq.spi.MqIdempotentStore;
 import com.njcn.mq.spi.MqMessageListener;
 import com.njcn.mq.spi.MqSubscription;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.SmartLifecycle;
 import java.util.ArrayList;
 import java.util.List;
 
+@Slf4j
 public class MqListenerRegistry implements SmartLifecycle {
     private final MqDriver driver;
     private final List<MqListenerEndpoint> endpoints;
@@ -34,6 +36,10 @@ public class MqListenerRegistry implements SmartLifecycle {
         if (running) return;
         for (MqListenerEndpoint ep : endpoints) {
             MqListener meta = ep.getMeta();
+            if (meta.idempotent() && store == null) {
+                log.warn("[mq] @MqListener(idempotent=true) 但容器中无 MqIdempotentStore bean，去重不生效: topic={} group={}",
+                        meta.topic(), meta.group());
+            }
             MqConsumeErrorHandler eh = resolveErrorHandler(meta);
 
             MqConsumeDispatcher.Builder db = MqConsumeDispatcher.builder()
@@ -64,7 +70,12 @@ public class MqListenerRegistry implements SmartLifecycle {
 
     private MqConsumeErrorHandler resolveErrorHandler(MqListener meta) {
         if (errorHandlerResolver == null || meta.errorHandler().isEmpty()) return null;
-        return errorHandlerResolver.apply(meta.errorHandler());
+        try {
+            return errorHandlerResolver.apply(meta.errorHandler());
+        } catch (Exception e) {
+            throw new IllegalStateException("@MqListener(errorHandler=\"" + meta.errorHandler()
+                    + "\") 指定的 bean 未找到: topic=" + meta.topic() + " group=" + meta.group(), e);
+        }
     }
 
     /** 业务调用适配：批量 endpoint 调 (List&lt;payload&gt;[,ctx])；单条 endpoint 逐条调 (payload[,ctx])。 */

mq-starter-core/src/main/resources/META-INF/spring.factories

@@ -1,2 +1,3 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-com.njcn.mq.autoconfig.MqCoreAutoConfiguration
+com.njcn.mq.autoconfig.MqCoreAutoConfiguration,\
+com.njcn.mq.autoconfig.MqDriverPresenceAutoConfiguration

mq-starter-core/src/test/java/com/njcn/mq/autoconfig/MqDriverPresenceAutoConfigurationTest.java

@@ -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());
+    }
+}

mq-starter-core/src/test/java/com/njcn/mq/container/MqConsumeDispatcherTest.java

@@ -142,4 +142,28 @@ class MqConsumeDispatcherTest {
 
         assertEquals(2, succeeded.size(), "flush 成功后应对每条标记 success");
     }
+
+    @Test
+    void singleMode_whenErrorHandlerThrows_stillMarksFailAndRethrowsOriginal() {
+        List<String> failed = new ArrayList<>();
+        MqIdempotentStore store = new MqIdempotentStore() {
+            @Override public boolean tryAcquire(String key) { return true; }
+            @Override public void markSuccess(String key) { }
+            @Override public void markFail(String key) { failed.add(key); }
+        };
+        // errorHandler 自身抛异常（落库 sink 挂掉，常与错误风暴同时发生）
+        MqConsumeErrorHandler eh = (m, id, ex) -> { throw new RuntimeException("sink down"); };
+        RuntimeException boom = new RuntimeException("biz boom");
+        MqConsumeDispatcher d = MqConsumeDispatcher.builder()
+                .batchSize(1).idempotent(true).store(store).errorHandler(eh)
+                .consumer(batch -> { throw boom; })
+                .build();
+
+        Exception thrown = assertThrows(Exception.class,
+                () -> d.dispatch(MqMessage.of("k1", "", "a")));
+
+        assertSame(boom, thrown, "errorHandler 失败不应掩盖原始消费异常");
+        assertEquals(1, failed.size(),
+                "errorHandler 抛异常时仍须 markFail，否则去重标记停在 processing 会导致 reclaim 时被去重挡掉而丢消息");
+    }
 }

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisMqIdempotentStore.java

@@ -41,6 +41,8 @@ public class RedisMqIdempotentStore implements MqIdempotentStore {
             return true;
         }
         // key 已存在：fail（上次失败）允许重新消费；processing / success 一律挡回。
+        // 注意：fail→reopen 的 get+set 非原子，跨实例对「共享同一 key 的不同消息」可能都放行；
+        // 同一条消息由 stream PEL/claim 序列化，故仅影响不同消息共享 key 的边缘场景。
         if (FAIL.equals(redis.opsForValue().get(k))) {
             redis.opsForValue().set(k, PROCESSING, Duration.ofSeconds(processingTtl));
             return true;

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamMqDriver.java

@@ -28,7 +28,9 @@ public class RedisStreamMqDriver implements MqDriver {
     private volatile boolean running = true;
 
     public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props) {
-        this(redis, keyBuilder, props, 5000L);
+        // reclaim 默认 idle 30s：须大于正常处理耗时，避免多实例误 reclaim 仍在处理(in-flight)的消息造成重复消费；
+        // 非幂等消费在多实例下为 at-least-once，可通过 4 参构造按业务处理时延调整。
+        this(redis, keyBuilder, props, 30_000L);
     }
 
     public RedisStreamMqDriver(StringRedisTemplate redis, StreamKeyBuilder keyBuilder, RedisStreamProperties props, long reclaimMinIdleMs) {
@@ -132,8 +134,7 @@ public class RedisStreamMqDriver implements MqDriver {
             if (deliveryCount > maxRetry) {
                 if (sub.getRetryExhaustedHandler() != null) {
                     try {
-                        sub.getRetryExhaustedHandler().accept(MqMessage.of(
-                                fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), null));
+                        sub.getRetryExhaustedHandler().accept(buildMessage(sub, fields));
                     } catch (Exception ex) {
                         log.error("[mq-redis] retryExhausted handler error key={}", fields.get(StreamMessageConstant.F_KEY), ex);
                     }
@@ -147,6 +148,18 @@ public class RedisStreamMqDriver implements MqDriver {
         }
     }
 
+    /** 解码消息体为业务 payload（best-effort：毒消息时 payload 为 null），用于重试耗尽落库回调。 */
+    private MqMessage buildMessage(MqSubscription sub, Map<String, String> fields) {
+        Object payload = null;
+        try {
+            String json = MessageCodec.decodeBody(fields.get(StreamMessageConstant.F_ENC), fields.get(StreamMessageConstant.F_BODY));
+            payload = JSON.parseObject(json, sub.getPayloadType());
+        } catch (Exception ignore) {
+            // 毒消息：payload 留 null，至少保留 key/tag 供落库
+        }
+        return MqMessage.of(fields.get(StreamMessageConstant.F_KEY), fields.get(StreamMessageConstant.F_TAG), payload);
+    }
+
     private void ensureGroup(String stream, String group) {
         try {
             redis.execute((RedisCallback<Object>) c -> c.execute("XGROUP",

mq-starter-redis-stream/src/main/java/com/njcn/mq/driver/redis/RedisStreamStartupProbeAutoConfiguration.java

@@ -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";
+    }
+}

mq-starter-redis-stream/src/main/resources/META-INF/spring.factories

@@ -1,2 +1,3 @@
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration
+com.njcn.mq.driver.redis.RedisStreamMqDriverAutoConfiguration,\
+com.njcn.mq.driver.redis.RedisStreamStartupProbeAutoConfiguration

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamMqDriverRetryIT.java

@@ -32,18 +32,20 @@ class RedisStreamMqDriverRetryIT {
 
         String topic = "MQ_RETRY_IT_" + System.nanoTime();
         AtomicInteger attempts = new AtomicInteger();
-        BlockingQueue<String> exhausted = new LinkedBlockingQueue<>();
+        BlockingQueue<MqMessage> exhausted = new LinkedBlockingQueue<>();
         driver.subscribe(MqSubscription.builder()
                 .topic(topic).group("g1").tag("*").payloadType(String.class)
                 .concurrency(1).batchSize(1).maxRetry(2)
                 .listener(m -> { attempts.incrementAndGet(); throw new RuntimeException("always fail"); })
-                .retryExhaustedHandler(m -> exhausted.add(m.getKey()))
+                .retryExhaustedHandler(exhausted::add)
                 .build());
 
         driver.send(topic, MqMessage.of("rk1", "", "data"));
 
-        String key = exhausted.poll(12, TimeUnit.SECONDS);
-        assertEquals("rk1", key, "重试耗尽应回调 retryExhaustedHandler");
+        MqMessage exhaustedMsg = exhausted.poll(12, TimeUnit.SECONDS);
+        assertNotNull(exhaustedMsg, "重试耗尽应回调 retryExhaustedHandler");
+        assertEquals("rk1", exhaustedMsg.getKey());
+        assertEquals("data", exhaustedMsg.getPayload(), "耗尽回调应携带完整 payload，而非 null");
         assertTrue(attempts.get() >= 2, "应实际重试若干次后才耗尽, attempts=" + attempts.get());
 
         Thread.sleep(500);

mq-starter-redis-stream/src/test/java/com/njcn/mq/driver/redis/RedisStreamStartupProbeAutoConfigurationTest.java

@@ -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());
+    }
+}

mq-starter-rocketmq/pom.xml

@@ -3,6 +3,17 @@
     <modelVersion>4.0.0</modelVersion>
     <parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
     <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>
         <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>

mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfiguration.java

@@ -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);
+    }
+}

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

mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqStartupProbeAutoConfigurationTest.java

@@ -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());
+    }
+}