Redis如何使用乐观锁(CAS)保证数据一致性

Iris ·

更新时间:2024-11-13

· 1886 次阅读

场景

问题模拟

CAS 来保证数据一致性

场景

在 Redis 中经常会存在这么一种情况，读取某一个 key 的值，做一些业务逻辑处理，然后根据读取到的值来计算出一个新的值，重新 set 进去。

如果客户端 A 刚读取到 key 值，紧接着客户端 B 就修改这个 key 的值，那么就会存在并发安全的问题。

问题模拟

假设 Redis Server 有个键名为 test 的key，里面存放的是一个 json 数组 [1, 2, 3]。

下面让我们模拟一下，客户端 A 与客户端 B 同时访问修改的情况，代码如下：

客户端 A：

class RedisClientA(username: String, password: String, host: String, port: Int) {
    val jedis: Jedis
    init {
        val pool = JedisPool(JedisPoolConfig(), host, port)
        jedis = pool.resource
        jedis.auth(username, password)
    }
    fun update(key: String) {
        val idStr = jedis.get(key)
        val idList = Json.decodeFromString<MutableList<Int>>(idStr)
        // 等待2秒，模拟业务
        TimeUnit.SECONDS.sleep(2L)
        idList.add(4)
        println("new id list: $idList")
        jedis.set(key, Json.encodeToString(idList))
    }
    fun getVal(key: String): String? {
        return jedis.get(key)
    }
}
fun main() {
    val key = "test"
    val redisClientA = RedisClientA("default", "123456", "127.0.0.1", 6379)
    redisClientA.update(key)
    val res = redisClientA.getVal(key)
    println("res: $res")
}

客户端 B:

class RedisClientB(username: String, password: String, host: String, port: Int) {
    val jedis: Jedis
    init {
        val pool = JedisPool(JedisPoolConfig(), host, port)
        jedis = pool.resource
        jedis.auth(username, password)
    }
    fun update(key: String) {
        val idStr = jedis.get(key)
        val idList = Json.decodeFromString<MutableList<Int>>(idStr)
        idList.add(5)
        println("new id list: $idList")
        jedis.set(key, Json.encodeToString(idList))
    }
    fun getVal(key: String): String? {
        return jedis.get(key)
    }
}
fun main() {
    val key = "test"
    val redisClientB = RedisClientB("default", "123456", "127.0.0.1", 6379)
    redisClientB.update(key)
    val res = redisClientB.getVal(key)
    println("res: $res")
}

客户端 A 阻塞了 2 秒，用来模拟耗时业务逻辑的处理。正在处理的时候，客户端 B 访问了 “test”，并增加了 id:5。

在客户端 A 耗时业务逻辑处理完的时候，增加了 id:4，并且会覆盖掉 id:5。

最终“test” 里的内容最终如下：

CAS 来保证数据一致性

WATCH 命令可以为 Redis 事务提供 check-and-set（CAS）行为。被 WATCH 的键会被监视，并会发觉这些键是否被改动过了。如果有至少一个被监视的建在 EXEC 执行之前被修改了，那么整个事务都会被取消，EXEC 返回空（Null replay）来表示事务执行失败。我们只需要重复操作，希望在这个时间段内不会有新的竞争。这种形式的锁被称作乐观锁，它是一种非常强大的锁机制。

那么 CAS 的方式如何实现呢？我们只需要把 RedisClientA 的 update() 方法中的代码修改如下：

fun update(key: String) {
    var flag = true
    while (flag) {
        jedis.watch(key)
        val idStr = jedis.get(key)
        val idList = Json.decodeFromString<MutableList<Int>>(idStr)
        // 等待2秒，模拟业务
        TimeUnit.SECONDS.sleep(2L)
        val transaction = jedis.multi()
        idList.add(4)
        println("new id list: $idList")
        transaction.set(key, Json.encodeToString(idList))
        transaction.exec()?.let {
            flag = false
        }
    }
}

最终 “test” 的内容如下：