Redis如何使用乐观锁(CAS)保证数据一致性
时间：2022-08-30 09:28:54 | 栏目：Redis | 点击：次
场景
在 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” 的内容如下：
可见我们通过使用 WATCH 和 TRANACTION 命令，采用 CAS 乐观锁的方式实现了数据的一致性。