Redis集群实现分布式锁-RedLock

一、redis集群分布式锁

Redis单节点实现分布式锁 ，如果通过sentinel保证高可用，如果master节点由于某些原因发生了主从切换，那么就会出现锁丢失的情况：

1. 客户端1在Redis的master节点上拿到了锁。

2. Master宕机了，存储锁的key还没有来得及同步到Slave上。

3. master故障，发生故障转移，slave节点升级为master节点。

4. 客户端2从新的Master获取到了对应同一个资源的锁。

于是，客户端1和客户端2同时持有了同一个资源的锁。锁的安全性被打破了。针对这个问题。Redis作者antirez提出了RedLock算法来解决这个问题。

二、RedLock简介

在不同进程需要互斥地访问共享资源时，分布式锁是一种非常有用的技术手段。实现高效的分布式锁有三个属性需要考虑：

1、安全属性：互斥，不管什么时候，只有一个客户端持有锁。
2、效率属性A:不会死锁。
3、效率属性B：容错，只要大多数redis节点能够正常工作，客户端端都能获取和释放锁。

三、RedLock算法

在分布式版本的算法里我们假设我们有N个Redis master节点，这些节点都是完全独立的，我们不用任何复制或者其他隐含的分布式协调算法。我们已经描述了如何在单节点环境下安全地获取和释放锁。因此我们理所当然地应当用这个方法在每个单节点里来获取和释放锁。在我们的例子里面我们把N设成5，这个数字是一个相对比较合理的数值，因此我们需要在不同的计算机或者虚拟机上运行5个master节点来保证他们大多数情况下都不会同时宕机。一个客户端需要做如下操作来获取锁：

1、获取当前时间（单位是毫秒）。

2、轮流用相同的key和随机值在N个节点上请求锁，在这一步里，客户端在每个master上请求锁时，会有一个和总的锁释放时间相比小的多的超时时间。比如如果锁自动释放时间是10秒钟，那每个节点锁请求的超时时间可能是5-50毫秒的范围，这个可以防止一个客户端在某个宕掉的master节点上阻塞过长时间，如果一个master节点不可用了，我们应该尽快尝试下一个master节点。

3、客户端计算第二步中获取锁所花的时间，只有当客户端在大多数master节点上成功获取了锁（在这里是3个），而且总共消耗的时间不超过锁释放时间，这个锁就认为是获取成功了。

4、如果锁获取成功了，那现在锁自动释放时间就是最初的锁释放时间减去之前获取锁所消耗的时间。

5、如果锁获取失败了，不管是因为获取成功的锁不超过一半（N/2+1)还是因为总消耗时间超过了锁释放时间，客户端都会到每个master节点上释放锁，即便是那些他认为没有获取成功的锁。

四、Redissson实现RedLock

redisson包已经有对redlock算法封装，接下来就具体看看使用redisson包来实现分布式锁的正确姿势。

<!-- JDK 1.8+ compatible -->
<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>3.9.0</version>
</dependency>  
 
<!-- JDK 1.6+ compatible -->
<dependency>
   <groupId>org.redisson</groupId>
   <artifactId>redisson</artifactId>
   <version>2.14.0</version>
</dependency>

Redisson管理类：

import org.redisson.Redisson;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
 
 
 
public class RedissonManager {
    private static Config config = new Config();
    private static RedissonClient redisson = null;
    private static final String RAtomicName = "genId_";
    public static void init(){
        try{
            config.useClusterServers()
                    .setScanInterval(200000)//设置集群状态扫描间隔
                    .setMasterConnectionPoolSize(10000)//设置对于master节点的连接池中连接数最大为10000
                    .setSlaveConnectionPoolSize(10000)//设置对于slave节点的连接池中连接数最大为500
                    .setIdleConnectionTimeout(10000)//如果当前连接池里的连接数量超过了最小空闲连接数，而同时有连接空闲时间超过了该数值，那么这些连接将会自动被关闭，并从连接池里去掉。时间单位是毫秒。
                    .setConnectTimeout(30000)//同任何节点建立连接时的等待超时。时间单位是毫秒。
                    .setTimeout(3000)//等待节点回复命令的时间。该时间从命令发送成功时开始计时。
                    .setRetryInterval(3000)//当与某个节点的连接断开时，等待与其重新建立连接的时间间隔。时间单位是毫秒。
                    .addNodeAddress("redis://127.0.0.1:7000","redis://127.0.0.1:7001","redis://127.0.0.1:7002","redis://127.0.0.1:7003","redis://127.0.0.1:7004","redis://127.0.0.1:7005");
            redisson = Redisson.create(config);
 
            RAtomicLong atomicLong = redisson.getAtomicLong(RAtomicName);
            atomicLong.set(0);//自增设置为从0开始
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    public static RedissonClient getRedisson(){
        if(redisson == null){
            RedissonManager.init(); //初始化
        }
        return redisson;
    }

我们配置了很多参数，其实一共有十来种参数，我们只是设置几个比较重要的而已。

getRedisson  方法是使用者初始化 Redisson。

nextID 方法返回一共为 RAtomicName 变量操作了多少次，也就是我成功使用分布式锁的次数。

分布式锁操作类：

import com.config.RedissonManager;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;
 
 
@Component
public class RedissonLock {
    private static final Logger LOGGER = LoggerFactory.getLogger(RedissonLock.class);
 
    private static RedissonClient redissonClient = RedissonManager.getRedisson();
 
 
    public void lock(String lockName) {
        String key = lockName;
        RLock myLock = redissonClient.getLock(key);
        //lock提供带timeout参数，timeout结束强制解锁，防止死锁
        myLock.lock(2, TimeUnit.SECONDS);
        // 1. 最常见的使用方法
        //lock.lock();
        // 2. 支持过期解锁功能,10秒以后自动解锁, 无需调用unlock方法手动解锁
        //lock.lock(10, TimeUnit.SECONDS);
        // 3. 尝试加锁，最多等待3秒，上锁以后10秒自动解锁
//        try {
//            boolean res = mylock.tryLock(3, 10, TimeUnit.SECONDS);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        System.err.println("======lock======" + Thread.currentThread().getName());
    }
 
    public void unLock(String lockName) {
        String key = lockName;
        RLock myLock = redissonClient.getLock(key);
        myLock.unlock();
        System.err.println("======unlock======" + Thread.currentThread().getName());
    }
}

lock 方法是加锁操作，unLock 方法是解锁操作。

注释中的代码列举类 3中lock 的方法，大家学习更多操作请查看下面博客

https://blog.csdn.net/l1028386804/article/details/73523810

教你 RedissonClient 所有操作。