Vert.x Redis Client实现Compare-and-Delete的WATCH事务优化咨询

核心疑问解答 #

• RedisAPI的会话跟踪：Vert.x的RedisAPI（即RedisClient）基于连接池实现，每次调用命令会从池中获取任意可用连接，不会跟踪单个会话。而Redis的WATCH命令绑定在单个连接上，因此必须使用同一个RedisConnection实例执行完整的WATCH/GET/MULTI/EXEC序列，才能保证WATCH的有效性。
• UNWATCH的处理逻辑：
  • 若GET或MULTI操作失败，需要主动调用UNWATCH，否则该连接上的WATCH会持续生效，影响后续操作。
  • 执行EXEC后，Redis会自动取消当前连接的WATCH状态；若连接意外断开，Redis也会自动清除该连接的WATCH状态。

优化后的CAD实现代码 #

以下是使用Vert.x原生Future API优化的实现，避免嵌套过深问题，同时正确处理连接、WATCH和异常：

import io.vertx.core.Future;
import io.vertx.redis.client.Redis;
import io.vertx.redis.client.RedisConnection;
import io.vertx.redis.client.Response;
import static io.vertx.redis.client.Command.*;
import static io.vertx.redis.client.Request.cmd;

public class RedisCadHandler {

    private final Redis redisClient;

    public RedisCadHandler(Redis redisClient) {
        this.redisClient = redisClient;
    }

    public Future<Boolean> compareAndDelete(String key, String expectedValue) {
        // 获取单个连接，全程复用保证WATCH有效性
        return redisClient.connect()
                .compose(conn -> {
                    // 执行WATCH监听目标key
                    return conn.send(cmd(WATCH).arg(key))
                            // 获取当前key的实际值
                            .compose(v -> conn.send(cmd(GET).arg(key)))
                            // 对比期望值与实际值
                            .compose(getResp -> {
                                if (getResp == null || !getResp.toString().equals(expectedValue)) {
                                    // 值不匹配或key不存在，取消WATCH并释放连接
                                    return conn.send(cmd(UNWATCH))
                                            .onComplete(ignored -> conn.close())
                                            .map(false);
                                }

                                // 值匹配，开启事务执行删除
                                return conn.send(cmd(MULTI))
                                        .compose(multiResp -> conn.send(cmd(DEL).arg(key)))
                                        .compose(delResp -> conn.send(cmd(EXEC)))
                                        .compose(execResp -> {
                                            // EXEC返回null表示事务被打断（key被其他客户端修改）
                                            boolean success = execResp != null && execResp.size() > 0 && execResp.get(0).toInteger() == 1;
                                            // EXEC后Redis自动取消WATCH，直接释放连接
                                            conn.close();
                                            return Future.succeededFuture(success);
                                        });
                            })
                            // 全局异常捕获：主动取消WATCH并释放连接
                            .recover(err -> {
                                return conn.send(cmd(UNWATCH))
                                        .onComplete(ignored -> conn.close())
                                        .compose(ignored -> Future.failedFuture(err));
                            });
                });
    }
}

代码说明 #

• 连接复用：通过redisClient.connect()获取单个连接，全程使用该连接执行所有命令，确保WATCH的绑定关系不失效。
• 链式调用：用compose替代嵌套结构，将每个操作步骤串联，代码逻辑更清晰易读。
• 异常处理：通过recover捕获所有异常场景，主动执行UNWATCH并关闭连接，避免连接资源泄漏和无效的WATCH状态残留。
• 事务结果判断：EXEC返回null代表事务被打断（WATCH的key在事务执行前被修改）；若返回的响应数组中DEL命令结果为1，则表示删除成功。

注意事项 #

• 该方案性能低于Lua脚本实现，涉及多次网络往返，且存在事务被打断的可能，若业务允许可在外层添加重试逻辑。

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