【Redis】用Redis实现分布式锁、乐观锁

1、除了做缓存，Redis还能用来干什么？

Redis最主要的功能就是拿来做缓存，来提升系统的性能，但是除了做缓存以外，他还能做很多事（但是，能做并不代表就适合，并不代表就一定要用它）：

消息队列（不建议）：Redis 支持发布/订阅模式和Stream，可以作为轻量级消息队列使用，用于异步处理任务或处理高并发请求。

• 延迟消息（不建议）：Redis的ZSET可以用来实现延迟消息，也可以基于Key的过期消息实现延迟消息，还可以借助Redisson的RDelayQueue来实现延迟消息，都是可以的。
• 排行榜（建议）：利用Redis 的有序集合和列表结构，可以成为设计实时排行榜的绝佳选择，例如各类热门排行榜、热门商品列表等。
• 计数器（建议）：基于Redis可以实现一些计数器的功能，比如网站的访问量、朋友圈点赞等。通过 incr 命令就能实现原子性的自增操作，从而实现一个全局计数器。·
• 分布式ID（可以）：因为他有全局自增计数的功能，所以在分布式场景，我们也可以利用Redis来实现一个分布式ID来保障全局的唯一且自增。
• 分布式锁（建议）：Redis 的单线程特性可以保证多个客户端之间对同一把锁的操作是原子性的，可以轻松实现分布式锁，用于控制多个进程对共享资源的访问。
• 地理位置应用（建议）：Redis 支持GEO，支持地理位置定位和查询，可以存储地理位置信息并通过 Redis 的查询功能获取附近的位置信息。比如"附近的人"用它来实现就非常方便。
• 分布式限流（可以）：Redis提供了令牌桶和漏桶算法的实现，可以用于实现分布式限流。
• 分布式Session（建议）：可以使用Redis实现分布式Session管理，保证多台服务器之间用户的会话状态同步。
• 布隆过滤器（建议）：Redis提供了布隆过滤器（Bloom Filter）数据结构的实现，可以高效地检测一个元素是否存在于一个集合中
• 状态统计（数据量大建议用）：Redis中支持BitMap这种数据结构，它不仅查询和存储高效，更能节省很多空间，所以我们可以借助他做状态统计，比如记录亿级用户的登录状态，或者是拿他来做签到统计也比较常见。
• 共同关注（建议）：Redis中支持Set集合类型，这个类型非常适合我们做一些取并集、交集、差集等，基于这个特性，我们就能取交集的方式非常方便的实现共同好友、或者共同关注的功能。
• 推荐关注（可以）：和上面的共同关注类似，交集实现共同好友，那么并集或者差集就能实现推荐关注的功能。

2、分布式锁

2.1 如何用SETNX实现分布式锁？

利用Redis的单线程特性，在多个Redis客户端同时通过SETNX命令尝试获取锁，如果返回1表示获取锁成功，否则表示获取锁失败。

Redis Setnx（SET if Not eXists） 命令在指定的 key 不存在时，为 key 设置指定的值。设置成功，返回 1 。 设置失败，返回 0 。

因为Redis的单线程机制，所以可以保证只会有一个客户端成功获取到锁，而其他客户端则会失败。如果获取锁成功，则设置一个过期时间，防止该客户端挂了之后一直持有该锁。客户端释放锁的时候，需要先判断该锁是否仍然属于该客户端，如果是，则通过DEL命令释放锁。

public class RedisDistributedLock {private final JedisPool jedisPool;public RedisDistributedLock(JedisPool jedisPool) {this.jedisPool = jedisPool;}public boolean tryLock(String lockKey, String requestId, int expireTime) {try (Jedis jedis = jedisPool.getResource()) {String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);return "OK".equals(result);}}public boolean unlock(String lockKey, String requestId) {try (Jedis jedis = jedisPool.getResource()) {String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));return Long.parseLong(result.toString()) == 1L;}}
}

优点：
（1）实现简单：SETNX命令实现简单，易于理解和使用。
（2）性能较高：由于SETNX命令的执行原子性，保证了分布式锁的正确性，而且在Redis中，SETNX命令是单线程执行的，所以性能较高。

缺点：
（1）锁无法续期：如果加锁方在加锁后的执行时间较长，而锁的超时时间设置的较短，可能导致锁被误释放。
（2）无法避免死锁：如果加锁方在加锁后未能及时解锁（也未设置超时时间），且该客户端崩溃，可能导致死锁。
（3）存在竞争：由于SETNX命令是对Key的操作，所以在高并发情况下，多个客户端之间仍可能存在竞争，从而影响性能。
（4）setnx不支持可重入，可以借助redission封装的能力实现可重入锁。

2.2 如何用Redisson实现分布式锁？

在使用SETNX实现的分布式锁中，因为存在锁无法续期导致并发冲突的问题，所以在真实的生产环境中用的并不是很多，其实，真正在使用Redis时，用的比较多的是基于Redisson实现分布式锁。

为了避免锁超时，Redisson中引入了看门狗的机制，他可以帮助我们在Redisson实例被关闭前，不断的延长锁的有效期。

自动续租：当一个Redisson客户端实例获取到一个分布式锁时，如果没有指定锁的超时时间，Watchdog会基于Netty的时间轮启动一个后台任务，定期向Redis发送命令，重新设置锁的过期时间，通常是锁的租约时间的1/3。这确保了即使客户端处理时间较长，所持有的锁也不会过期。

续期时长：默认情况下，每10s钟做一次续期，续期时长是30s。

停止续期：当锁被释放或者客户端实例被关闭时，Watchdog会自动停止对应锁的续租任务。

如果一个锁设置了超时时间，则不会续期。或者一个锁的unlock方法被调用了，那么就会停止续期。

2.3 watchdog一直续期，那客户端挂了怎么办？

如果，应用集群中的一台机器，拿到了分布式锁，但是在执行的过程中，他挂了，还没来得及把锁释放，那么会有问题么？

因为我们知道，锁的续期是Redisson实现的，而Redisson的后台任务是基于JVM运行的，也就是说，如果这台机器挂了，那么Redisson的后台任务也就没办法继续执行了。

那么他也就不会会再继续续期了，那么到了期限之后，锁就会自动解除了。这样就可以避免因为一个实例宕机导致分布式锁的不可用。

2.4 什么是RedLock，他解决了什么问题？

RedLock是Redis的作者提出的一个多节点分布式锁算法，旨在解决使用单节点Redis分布式锁可能存在的单点故障问题。

Redis的单点故障问题：
1、在使用单节点Redis实现分布式锁时，如果这个Redis实例挂掉，那么所有使用这个实例的客户端都会出现无法获取锁的情况。
2、当使用集群模式部署的时候，如果master一个客户端在master节点加锁成功了，然后没来得及同步数据到其他节点上，他就挂了， 那么这时候如果选出一个新的节点，再有客户端来加锁的时候，就也能加锁成功，因为数据没来得及同步，新的master会认为这个key是不存在的。

RedLock通过使用多个Redis节点，来提供一个更加健壮的分布式锁解决方案，能够在某些Redis节点故障的情况下，仍然能够保证分布式锁的可用性。

RedLock是通过引入多个Redis节点来解决单点故障的问题。

在进行加锁操作时，RedLock会向每个Redis节点发送相同的命令请求，每个节点都会去竞争锁，如果至少在大多数节点上成功获取了锁，那么就认为加锁成功。反之，如果大多数节点上没有成功获取锁，则加锁失败。这样就可以避免因为某个Redis节点故障导致加锁失败的情况发生。

3、介绍下Redis集群的脑裂问题？

Redis的脑裂问题可能发生在网络分区或者主节点出现问题的时候：

● 网络分区：网络故障或分区导致了不同子集之间的通信中断。
○ Master节点，哨兵和Slave节点被分割为了两个网络，Master处在一个网络中，Slave库和哨兵在另外一个网络中，此时哨兵发现和Master连不上了，就会发起主从切换，选一个新的Master，这时候就会出现两个主节点的情况。
● 主节点问题：集群中的主节点之间出现问题，导致不同的子集认为它们是正常的主节点。
○ Master节点有问题，哨兵就会开始选举新的主节点，但是在这个过程中，原来的那个Master节点又恢复了，这时候就可能会导致一部分Slave节点认为他是Master节点，而另一部分Slave新选出了一个Master

脑裂问题可能导致以下问题：
● 数据不一致：不同子集之间可能对同一数据进行不同的写入，导致数据不一致。
● 重复写入：在脑裂解决后，不同子集可能尝试将相同的写入操作应用到主节点上，导致数据重复。
● 数据丢失：新选出来的Master会向所有的实例发送slave of命令，让所有实例重新进行全量同步，而全量同步首先就会将实例上的数据先清空，所以在主从同步期间在原来那个Master上执行的命令将会被清空。

如何防止脑裂的发生？

Redis 已经提供了两个配置项可以帮我们做这个事，分别是 min-slaves-to-write 和 min-slaves-max-lag：
min-slaves-to-write：主库能进行数据同步的最少从库数量；
min-slaves-max-lag：主从库间进行数据复制时，从库给主库发送 ACK 消息的最大延迟秒数。

这两个配置项必须同时满足，不然主节点拒绝写入。在期间满足min-slaves-to-write和min-slaves-max-lag的要求，那么主节点就会被禁止写入，脑裂造成的数据丢失情况自然也就解决了。

举个例子：

假设我们将 min-slaves-to-write 设置为 1，把 min-slaves-max-lag 设置为 10s。

如果Master节点因为某些原因挂了 12s，导致哨兵判断主库客观下线，开始进行主从切换。

同时，因为原Master宕机了 12s，没有一个（min-slaves-to-write）从库能和原主库在 10s（ min-slaves-max-lag） 内进行数据复制，这样一来，就因为不满足配置要求，原Master也就再也无法接收客户端请求了。

这样一来，主从切换完成后，也只有新主库能接收请求，这样就没有脑裂的发生了。

Redis脑裂可以采用min-slaves-to-write和min-slaves-max-lag合理配置尽量规避，但无法彻底解决。

4、Redisson的lock和tryLock有什么区别？

tryLock是尝试获取锁，如果能获取到直接返回true，如果无法获取到锁，他会按照我们指定的waitTime进行阻塞，在这个时间段内他还会再尝试获取锁。如果超过这个时间还没获取到则返回false。如果我们没有指定waitTime，那么他就在未获取到锁的时候，就直接返回false了。

lock的原理是以阻塞的方式去获取锁，如果获取锁失败会一直等待，直到获取成功。

所以，lock实现的是一个阻塞锁，而tryLock实现的是一个非阻塞锁（在没有指定waitTime的情况下）。

5、为什么Redis不支持回滚？

在Redis的官网文档中明确的提到过，不支持回滚。

不支持回滚主要的原因是支持回滚将对 Redis 的简洁性和性能产生重大影响。

Redis的设计就是简单、高效等，所以引入事务的回滚机制会让系统更加的复杂，并且影响性能。从使用场景上来说，Redis一般都是被用作缓存的，不太需要很复杂的事务支持，当人们需要复杂的事务时会考虑持久化的关系型数据库。相比于关系型数据库，Redis是通过单线程执行的，在执行过程中，出现错误的概率比较低，并且这些问题一般来编译阶段都应该被发现，所以就不太需要引入回滚机制。

6、如何用Redis实现乐观锁？

所谓乐观锁，其实就是基于CAS的机制，CAS的本质是Compare And Swap，就是需要知道一个key在修改前的值，去进行比较。

在Redis中，想要实现这个功能，可以依赖 WATCH 命令。这个命令一旦运行，他会确保只有在 WATCH 监视的键在调用 EXEC 之前没有改变时，后续的事务才会执行。

例如，如果没有 INCRBY，我们可以用下面的方式实现原子的增量操作：

WATCH counter
GET counter
MULTI
SET counter <从 GET 获得的值 + 任何增量>
EXEC

1. WATCH：使用 WATCH 命令监视一个或多个键。这个命令会监视给定键直到事务开始（即执行 MULTI 命令）。
2. GET：在事务开始之前，查询你需要的数据。
3. MULTI：使用 MULTI 命令开始事务。
4. SET：在事务中添加所有需要执行的命令。
5. EXEC：使用 EXEC 命令执行事务。如果自从事务开始以来监视的键被修改过，EXEC 将返回 nil，这表示事务中的命令没有被执行。

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;public class RedisOptimisticLock {public static void main(String[] args) {// 连接到 RedisJedis jedis = new Jedis("localhost");try {// 监视键String key = "myKey";jedis.watch(key);// 模拟从数据库读取最新值String value = jedis.get(key);int intValue = Integer.parseInt(value);// 开始事务Transaction t = jedis.multi();// 在事务中执行操作t.set(key, String.valueOf(intValue + 1));// 尝试执行事务if (t.exec() == null) {System.out.println("事务执行失败，数据已被其他客户端修改");} else {System.out.println("事务执行成功");}} catch (Exception e) {e.printStackTrace();} finally {jedis.close();}}
}

参考链接：
1、https://www.yuque.com/hollis666/wk6won/gxqm60
2、https://www.yuque.com/hollis666/wk6won/fg0f0wh41g8eu5ik
3、https://www.yuque.com/hollis666/wk6won/gt0qpqluiwb7bg70