Spring Boot Redis哨兵集群主切换后订阅丢失，如何重订阅？(v2.7.8)

环境与现象 #

• 使用spring-boot-starter-data-redis v2.7.8
• 集群配置：3个Redis副本 + 3个Sentinel副本
• 问题：Redis主节点宕机，从节点晋升为主节点后，onMessage()方法不再接收消息；尝试重写RedisMessageListenerContainer的handleSubscriptionException()实现重订阅，但该方法从未被调用

错误日志 #

2023-11-03 04:31:48.784  INFO 1 --- [ter.local:5000]] r.c.j.p.SentineledConnectionProvider     : Created connection pool to master at imm-db-2.redis.imm-ns.svc.cluster.local:6379.
2023-11-03 04:31:58.855  INFO 1 --- [   scheduling-1] o.s.d.r.c.l.LettuceConnectionFactory     : Validation of shared connection failed. Creating a new connection.
2023-11-03 04:32:02.746  WARN 1 --- [ioEventLoop-4-1] i.l.c.m.MasterSlaveTopologyRefresh       : Unable to connect to imm-db-1.redis.imm-ns.svc.cluster.local:6379

现有代码 #

LettuceConnectionFactory配置 #

@Bean
public LettuceConnectionFactory lettuceConnectionFactory() {
    LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder().build();

    ClusterTopologyRefreshOptions topologyRefreshOptions = ClusterTopologyRefreshOptions.builder()
        .adaptiveRefreshTriggersTimeout(Duration.ofSeconds(60)).enableAllAdaptiveRefreshTriggers()
        .enablePeriodicRefresh().dynamicRefreshSources(true).build();

    ClusterClientOptions clusterClientOptions = ClusterClientOptions.builder()
        .topologyRefreshOptions(topologyRefreshOptions).build();
        
    RedisSentinelConfiguration sentinelConfig = new 
        RedisSentinelConfiguration().master("mymaster")
        .sentinel("imm-db-0.redis.imm-ns.svc.cluster.local", 5000)
        .sentinel("imm-db-1.redis.imm-ns.svc.cluster.local", 5000)
        .sentinel("imm-db-2.redis.imm-ns.svc.cluster.local", 5000);
        
    return new LettuceConnectionFactory(sentinelConfig, LettuceClientConfiguration
        .builder().readFrom(ReadFrom.MASTER).clientOptions(clusterClientOptions).build());
}

RedisMessageListenerContainer配置 #

@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,
        MessageListener messageListener) {

    log.info("********************RedisMessageListenerContainer**************************init done");
    RedisMessageListenerContainer container = new RedisMessageListenerContainer() {
        @Override
        protected void handleSubscriptionException(CompletableFuture<Void> future,
                BackOffExecution backOffExecution, Throwable ex) {
            super.handleSubscriptionException(future, backOffExecution, ex);

            if (ex instanceof RedisConnectionFailureException) {
                log.error("***************RedisConnectionFailureException :{}", ex.getMessage());

            } else if (ex instanceof InterruptedException) {
                log.error("***************InterruptedException :{}", ex.getMessage());
            } else if (ex instanceof TransientDataAccessException || ex instanceof RecoverableDataAccessException) {
                log.error("***************RecoverableDataAccessException :{} isRunning:{}", ex.getMessage(),
                        isRunning());
                log.error(
                        "***************Connection failure occurred. Restarting subscription task manually due to "
                                + ex,
                        ex);
                if (isRunning()) {
                    log.error(
                            "***************INSIDE Connection failure occurred. Restarting subscription task manually due to "
                                    + ex,
                            ex);
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        log.error("thred inturrepted....:{}", e.getMessage());
                    }
                    start(); // best we can do
                }
            } else {
                log.error("*****************Shutting down application due to unknown exception is running:{}", ex);
                if (isRunning()) {
                    log.warn("*****************Shutting down application due to unknown exception " + ex, ex);
                }
            }
        }
    };

    container.setConnectionFactory(connectionFactory);
    container.setErrorHandler(new ErrorHandler() {
        @Override
        public void handleError(Throwable t) {
            log.info("********Inside exception handler:{}", t.getMessage());
        }
    });
    container.addMessageListener(messageListener, new ChannelTopic("TOPIC_HTTP_TRIGGER"));
    return container;
}

解决方案 #

1. 修正核心配置错误：混用Cluster与Sentinel配置 #

你当前错误地使用了ClusterClientOptions和ClusterTopologyRefreshOptions，但你的集群是Sentinel模式，并非Redis Cluster。这些Cluster相关配置仅适用于Redis Cluster架构，对Sentinel模式无效，会导致拓扑刷新、异常处理逻辑完全失效。

2. 重新配置LettuceConnectionFactory（Sentinel专属） #

替换Cluster配置为普通的ClientOptions，并添加重试、自动重连策略，确保Sentinel主从切换后能感知新主节点：

@Bean
public LettuceConnectionFactory lettuceConnectionFactory() {
    // 配置连接重试策略，失败后自动重试3次，间隔2秒
    Retry retry = Retry.fixedDelay(3, Duration.ofSeconds(2));
    
    // 配置客户端选项：自动重连、连接超时、断开后重连行为
    ClientOptions clientOptions = ClientOptions.builder()
            .socketOptions(SocketOptions.builder().connectTimeout(Duration.ofSeconds(5)).build())
            .connectionTimeout(Duration.ofSeconds(5))
            .disconnectedBehavior(ClientOptions.DisconnectedBehavior.RECONNECT)
            .autoReconnect(true)
            .build();

    // Sentinel集群配置
    RedisSentinelConfiguration sentinelConfig = new RedisSentinelConfiguration()
            .master("mymaster")
            .sentinel("imm-db-0.redis.imm-ns.svc.cluster.local", 5000)
            .sentinel("imm-db-1.redis.imm-ns.svc.cluster.local", 5000)
            .sentinel("imm-db-2.redis.imm-ns.svc.cluster.local", 5000);

    // 构建Lettuce客户端配置，绑定重试和客户端选项
    LettuceClientConfiguration clientConfig = LettuceClientConfiguration.builder()
            .readFrom(ReadFrom.MASTER) // 订阅必须连主节点
            .clientOptions(clientOptions)
            .retryPolicy(retry)
            .build();

    LettuceConnectionFactory factory = new LettuceConnectionFactory(sentinelConfig, clientConfig);
    // 添加连接监听器，感知连接开闭事件
    factory.addConnectionListener(new ConnectionListener() {
        @Override
        public void onOpen(RedisConnection connection) {
            log.info("Redis连接已建立，准备恢复订阅");
        }

        @Override
        public void onClose(RedisConnection connection) {
            log.warn("Redis连接已关闭，等待重连");
        }
    });
    return factory;
}

3. 优化RedisMessageListenerContainer的异常处理 #

修正handleSubscriptionException()逻辑，确保异常被正确捕获后重启订阅：

@Bean
public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory,
        MessageListener messageListener) {

    log.info("RedisMessageListenerContainer 初始化完成");
    RedisMessageListenerContainer container = new RedisMessageListenerContainer() {
        @Override
        protected void handleSubscriptionException(CompletableFuture<Void> future,
                BackOffExecution backOffExecution, Throwable ex) {
            log.error("订阅异常触发: {}", ex.getMessage(), ex);
            
            // 执行父类默认重试逻辑
            super.handleSubscriptionException(future, backOffExecution, ex);

            // 针对连接失败异常，强制重启容器（先停再启确保订阅完全重建）
            if (ex instanceof RedisConnectionFailureException && isRunning()) {
                log.info("开始重启订阅容器以恢复连接");
                stop();
                try {
                    // 等待3秒确保连接完全释放
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
                start();
            }
        }
    };

    container.setConnectionFactory(connectionFactory);
    container.setErrorHandler(t -> log.error("全局错误处理器捕获异常: {}", t.getMessage(), t));
    container.addMessageListener(messageListener, new ChannelTopic("TOPIC_HTTP_TRIGGER"));
    return container;
}

4. 额外验证项 #

• 确保Sentinel集群正常运行，能正确选举新主节点，且应用服务器可访问所有Sentinel节点
• 确认Redis主节点的订阅功能正常，新主节点晋升后未被禁用发布订阅权限
• 检查应用日志，主从切换后应能看到SentineledConnectionProvider创建新主节点连接的日志

内容的提问来源于stack exchange，提问作者Ganesh Gudghe