Vert.x Java应用Redis连接后宕机无法自动重连问题求助

• 应用启动阶段若Redis未运行，会自动重试直至连接成功；
• 但已建立连接后Redis宕机，应用不会触发重连逻辑，直接返回500错误：[httpStatus: 500, message: "Redis connection is broken."]
• 使用io.vertx.redis.client包的RedisOptions配置重连逻辑，但未生效

核心成员变量与方法 #

// 注：原代码中RedesOptions应为RedisOptions笔误
private final RedisOptions options = new RedisOptions();
private Redis redisClient;

private void createRedisClient(Handler<AsyncResult<Redis>> handler) {
    Redis.createClient(vertx, options)
            .connect(onConnect -> {
                if (onConnect.succeeded()) {
                    this.redisClient = onConnect.result();
                    // 尝试在连接出错时触发重连
                    this.redisClient.exceptionHandler(e -> {
                        attemptReconnect(0);
                    });
                }
                handler.handle(onConnect);
            });
}

private void attemptReconnect(int retry) {
    if (retry > MAX_RECONNECT_RETRIES) {
        // 达到最大重试次数，停止尝试
        return;
    }
    System.out.println("连接出错 >>>> ");
    // 指数退避算法，最大延迟10240ms
    long backoff = (long) (Math.pow(2, Math.min(retry, 10)) * 10);

    vertx.setTimer(backoff, timer -> createRedisClient(onReconnect -> {
        if (onReconnect.failed()) {
            attemptReconnect(retry + 1);
        }
    }));
}

构造方法中的初始化调用 #

createRedisClient(onCreate -> {
    if (onCreate.succeeded()) {
        System.out.println("连接成功 >>>> " + onCreate.result());
        log.info("Redis Connected: " + onCreate.result());
    } else if (onCreate.failed()) {
        // 原代码中onCreate.result()在失败时为null，应打印异常原因
        System.out.println("连接出错 >>>> " + onCreate.cause().getMessage());
        attemptReconnect(0);
    }
});

核心问题 #

1. 重连成功后，业务逻辑仍可能在使用旧的失效redisClient实例；
2. 未保证redisClient实例的线程安全，并发场景下可能出现实例不一致问题；
3. 原代码未在重连成功后明确通知业务逻辑切换到新客户端。

修复后的代码示例 #

import java.util.concurrent.atomic.AtomicReference;
import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.redis.client.Redis;
import io.vertx.redis.client.RedisOptions;

// 用原子引用存储客户端，保证线程安全
private final AtomicReference<Redis> redisClient = new AtomicReference<>();
private final RedisOptions options = new RedisOptions();
private static final int MAX_RECONNECT_RETRIES = 10;

private void createRedisClient(Handler<AsyncResult<Redis>> handler) {
    Redis.createClient(vertx, options)
            .connect(onConnect -> {
                if (onConnect.succeeded()) {
                    Redis newClient = onConnect.result();
                    // 更新原子引用中的最新客户端实例
                    redisClient.set(newClient);
                    // 为新客户端绑定异常处理器，连接断开时自动触发重连
                    newClient.exceptionHandler(e -> {
                        System.err.println("Redis连接异常，触发重连: " + e.getMessage());
                        attemptReconnect(0);
                    });
                    System.out.println("Redis连接成功");
                } else {
                    System.err.println("Redis连接失败: " + onConnect.cause().getMessage());
                }
                handler.handle(onConnect);
            });
}

private void attemptReconnect(int retry) {
    if (retry > MAX_RECONNECT_RETRIES) {
        System.err.println("达到最大重连次数，停止尝试");
        return;
    }
    long backoff = (long) (Math.pow(2, Math.min(retry, 10)) * 10);
    System.out.println("准备重连，重试次数: " + retry + "，延迟: " + backoff + "ms");

    vertx.setTimer(backoff, timer -> createRedisClient(onReconnect -> {
        if (onReconnect.failed()) {
            attemptReconnect(retry + 1);
        }
    }));
}

// 初始化调用
public YourVerticle() {
    createRedisClient(onCreate -> {
        if (onCreate.failed()) {
            attemptReconnect(0);
        }
    });
}

// 业务中调用Redis的示例方法，确保获取最新客户端实例
public void executeRedisCommand(String key, Handler<AsyncResult<String>> resultHandler) {
    Redis client = redisClient.get();
    if (client == null) {
        resultHandler.handle(AsyncResult.failed(new IllegalStateException("Redis客户端未初始化")));
        return;
    }
    client.get(key, resultHandler);
}

额外优化建议 #

• 利用RedisOptions内置重连配置：减少自定义逻辑复杂度，示例如下：

  RedisOptions options = new RedisOptions()
          .setReconnectAttempts(MAX_RECONNECT_RETRIES)
          .setReconnectInterval(100); // 基础重连间隔，可结合自定义指数退避
  

• 添加详细监控日志：记录每次重连的状态、次数，便于排查问题；
• 业务请求中增加客户端有效性校验：若获取到的客户端为空，可触发即时重连或返回友好提示。

内容的提问来源于stack exchange，提问作者Henrique Ferraz