分布式应用中的分布式锁

雨点打透心脏的1/2处 2022-03-01 11:18 197阅读 0赞

分布式锁是控制分布式系统之间同步访问共享资源的一种方式。如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要通过一些互斥手段来防止彼此之间的干扰，以保证一致性，在这种情况下，就需要使用分布式锁了。

简而言之，分布式锁就是指在分布式的部署环境下，通过锁机制来让多客户端互斥的对共享资源进行访问。

分布式锁要满足哪些要求呢？

1、互斥性–在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行；  
2、高可用的获取锁与释放锁；  
3、高性能的获取锁与释放锁；  
4、具备可重入特性–当一个客户端获取锁之后，这个客户端可以再次获取锁。  
5、具备锁失效机制，防止死锁；  
6、具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败。  
7、持锁人解锁–加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。

一般有三种实现方案

数据库(乐观锁/悲观锁)；  
Redis缓存；  
Zookeeper  
三种方案比较

从性能角度：缓存>Zookeeper>MySQL；  
从实现角度：MySQL>缓存>Zookeeper；  
从可靠性角度：Zookeeper>缓存>MySQL。

【1】基于数据库的分布式锁  
基于数据库来做分布式锁的话，通常有两种做法：

基于数据库的乐观锁

基于数据库的悲观锁

① 乐观锁机制

乐观锁机制其实就是在数据库表中引入一个版本号（version）字段来实现的。

当我们要从数据库中读取数据的时候，同时把这个version字段也读出来，如果要对读出来的数据进行更新后写回数据库，则需要将version加1，同时将新的数据与新的version更新到数据表中，且必须在更新的时候同时检查目前数据库里version值是不是之前的那个version，如果是，则正常更新。如果不是，则更新失败，说明在这个过程中有其它的进程去更新过数据了。

② 悲观锁机制

在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁 （需要注意的是，在InnoDB中只有字段加了索引的，才会是行级锁，否者是表级锁，所以就要给要执行的方法字段名添加索引），当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。我们可以认为获得排他锁的线程即可获得分布式锁，当获取到锁之后，可以执行方法的业务逻辑，执行完方法之后释放锁。

③ 使用数据库实现分布式锁相关问题探讨

1.因为是基于数据库实现的，数据库的可用性和性能将直接影响分布式锁的可用性及性能，所以，数据库需要双机部署、数据同步、主备切换。

2.不具备可重入的特性。在表中新增一列，用于记录当前获取到锁的机器和线程信息，在再次获取锁的时候，先查询表中机器和线程信息是否和当前机器和线程相同，若相同则直接获取锁。

3.没有锁失效机制。在表中新增一列，用于记录失效时间，并且需要有定时任务清除这些失效的数据。

4.不具备阻塞锁特性。获取不到锁直接返回失败，所以需要优化获取逻辑，循环多次去获取。

5.非公平的。再建一张中间表，将等待锁的线程全记录下来，并根据创建时间排序，只有最先创建的允许获取锁。

【2】基于 ZooKeeper 做分布式锁  
① ZooKeeper 锁相关基础知识

zk 一般由多个节点构成（单数），采用 zab 一致性协议。因此可以将 zk 看成一个单点结构，对其修改数据其内部自动将所有节点数据进行修改而后才提供查询服务。  
zk 的数据以目录树的形式，每个目录称为 znode， znode 中可存储数据（一般不超过 1M），还可以在其中增加子节点。  
子节点有三种类型。序列化节点，每在该节点下增加一个节点自动在该节点的名称上自增。临时节点，一旦创建这个 znode 的客户端与服务器失去联系，这个 znode 也将自动删除。最后就是普通节点。  
Watch 机制，client 可以监控每个节点的变化，当产生变化会给 client 产生一个事件。  
② ZK 基本锁

原理：利用临时节点与 watch 机制。每个锁占用一个普通节点 /lock，当需要获取锁时在 /lock 目录下创建一个临时节点，创建成功则表示获取锁成功，失败则 watch/lock 节点，有删除操作后再去争锁。临时节点好处在于当进程挂掉后能自动上锁的节点自动删除即取消锁。

缺点：所有取锁失败的进程都监听父节点，很容易发生羊群效应，即当释放锁后所有等待进程一起来创建节点，并发量很大。

③ ZK 锁优化

原理：上锁改为创建临时有序节点，每个上锁的节点均能创建节点成功，只是其序号不同。只有序号最小的可以拥有锁，如果这个节点序号不是最小的则 watch 序号比本身小的前一个节点 (公平锁)。

步骤：

1.在 /lock 节点下创建一个有序临时节点 (EPHEMERAL\_SEQUENTIAL)。  
2.判断创建的节点序号是否最小，如果是最小则获取锁成功。不是则取锁失败，然后 watch 序号比本身小的前一个节点。  
3.当取锁失败，设置 watch 后则等待 watch 事件到来后，再次判断是否序号最小。  
4.取锁成功则执行代码，最后释放锁（删除该节点）。  
④ 代码实现

import java.io.IOException;  
import java.util.ArrayList;  
import java.util.Collections;  
import java.util.List;  
import java.util.concurrent.CountDownLatch;  
import java.util.concurrent.TimeUnit;  
import java.util.concurrent.locks.Condition;  
import java.util.concurrent.locks.Lock;

import org.apache.zookeeper.CreateMode;  
import org.apache.zookeeper.KeeperException;  
import org.apache.zookeeper.WatchedEvent;  
import org.apache.zookeeper.Watcher;  
import org.apache.zookeeper.ZooDefs;  
import org.apache.zookeeper.ZooKeeper;  
import org.apache.zookeeper.data.Stat;

public class DistributedLock implements Lock, Watcher\{  
    private ZooKeeper zk;  
    private String root = "/locks";//根  
    private String lockName;//竞争资源的标志  
    private String waitNode;//等待前一个锁  
    private String myZnode;//当前锁  
    private CountDownLatch latch;//计数器  
    private int sessionTimeout = 30000;  
    private List<Exception> exception = new ArrayList<Exception>();

/\*\*  
     \* 创建分布式锁,使用前请确认config配置的zookeeper服务可用  
     \* @param config 127.0.0.1:2181  
     \* @param lockName 竞争资源标志,lockName中不能包含单词lock  
     \*/  
    public DistributedLock(String config, String lockName)\{  
        this.lockName = lockName;  
        // 创建一个与服务器的连接  
        try \{  
            zk = new ZooKeeper(config, sessionTimeout, this);  
            Stat stat = zk.exists(root, false);  
            if(stat == null)\{  
                // 创建根节点  
                zk.create(root, new byte\[0\], ZooDefs.Ids.OPEN\_ACL\_UNSAFE,CreateMode.PERSISTENT);  
            \}  
        \} catch (IOException e) \{  
            exception.add(e);  
        \} catch (KeeperException e) \{  
            exception.add(e);  
        \} catch (InterruptedException e) \{  
            exception.add(e);  
        \}  
    \}

/\*\*  
     \* zookeeper节点的监视器  
     \*/  
    public void process(WatchedEvent event) \{  
        if(this.latch != null) \{  
            this.latch.countDown();  
        \}  
    \}

public void lock() \{  
        if(exception.size() > 0)\{  
            throw new LockException(exception.get(0));  
        \}  
        try \{  
            if(this.tryLock())\{  
                System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true");  
                return;  
            \}  
            else\{  
                waitForLock(waitNode, sessionTimeout);//等待锁  
            \}  
        \} catch (KeeperException e) \{  
            throw new LockException(e);  
        \} catch (InterruptedException e) \{  
            throw new LockException(e);  
        \}  
    \}

public boolean tryLock() \{  
        try \{  
            String splitStr = "\_lock\_";  
            if(lockName.contains(splitStr))  
                throw new LockException("lockName can not contains \\\\u000B");  
            //创建临时子节点  
            myZnode = zk.create(root + "/" + lockName + splitStr, new byte\[0\], ZooDefs.Ids.OPEN\_ACL\_UNSAFE,CreateMode.EPHEMERAL\_SEQUENTIAL);  
            System.out.println(myZnode + " is created ");  
            //取出所有子节点  
            List<String> subNodes = zk.getChildren(root, false);  
            //取出所有lockName的锁  
            List<String> lockObjNodes = new ArrayList<String>();  
            for (String node : subNodes) \{  
                String \_node = node.split(splitStr)\[0\];  
                if(\_node.equals(lockName))\{  
                    lockObjNodes.add(node);  
                \}  
            \}  
            Collections.sort(lockObjNodes);  
            System.out.println(myZnode + "==" + lockObjNodes.get(0));  
            if(myZnode.equals(root+"/"+lockObjNodes.get(0)))\{  
                //如果是最小的节点,则表示取得锁  
                return true;  
            \}  
            //如果不是最小的节点，找到比自己小1的节点  
            String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);  
            waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);  
        \} catch (KeeperException e) \{  
            throw new LockException(e);  
        \} catch (InterruptedException e) \{  
            throw new LockException(e);  
        \}  
        return false;  
    \}

public boolean tryLock(long time, TimeUnit unit) \{  
        try \{  
            //获取锁成功，直接返回true  
            if(this.tryLock())\{  
                return true;  
            \}  
            //否则就等待  
            return waitForLock(waitNode,time);  
        \} catch (Exception e) \{  
            e.printStackTrace();  
        \}  
        return false;  
    \}

private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException \{  
        Stat stat = zk.exists(root + "/" + lower,true);  
        //判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听  
        if(stat != null)\{  
            System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower);  
            this.latch = new CountDownLatch(1);  
            this.latch.await(waitTime, TimeUnit.MILLISECONDS);  
            this.latch = null;  
        \}  
        return true;  
    \}

public void unlock() \{  
        try \{  
            System.out.println("unlock " + myZnode);  
            zk.delete(myZnode,-1);  
            myZnode = null;  
            zk.close();  
        \} catch (InterruptedException e) \{  
            e.printStackTrace();  
        \} catch (KeeperException e) \{  
            e.printStackTrace();  
        \}  
    \}

public void lockInterruptibly() throws InterruptedException \{  
        this.lock();  
    \}

public Condition newCondition() \{  
        return null;  
    \}

public class LockException extends RuntimeException \{  
        private static final long serialVersionUID = 1L;  
        public LockException(String e)\{  
            super(e);  
        \}  
        public LockException(Exception e)\{  
            super(e);  
        \}  
    \}  
\}  
⑤ zk锁优缺点

来看下Zookeeper能不能解决前面提到的问题。

锁无法释放？使用Zookeeper可以有效的解决锁无法释放的问题，因为在创建锁的时候，客户端会在ZK中创建一个临时节点，一旦客户端获取到锁之后突然挂掉（Session连接断开），那么这个临时节点就会自动删除掉。其他客户端就可以再次获得锁。

非阻塞锁？使用Zookeeper可以实现非阻塞的锁，客户端可以通过在ZK中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，Zookeeper会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，那么自己就获取到锁，便可以执行业务逻辑了。

不可重入？使用Zookeeper也可以有效的解决不可重入的问题，客户端在创建节点的时候，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获取锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么自己直接获取到锁，如果不一样就再创建一个临时的顺序节点，参与排队。

单点问题？使用Zookeeper可以有效的解决单点问题，ZK是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务。

缺点：

性能上可能并没有缓存服务那么高，因为每次在创建锁和释放锁的过程中，都要动态创建、销毁临时节点来实现锁功能。ZK 中创建和删除节点只能通过 Leader 服务器来执行，然后将数据同步到所有的 Follower 机器上。

【3】基于Redis分布式锁  
Redis分布式锁流程图：

① 基于setnx()、EXPIRE() 方法做分布式锁

setnx 的含义就是 SET if Not Exists，其主要有两个参数 setnx(key, value)。该方法是原子的，如果 key 不存在，则设置当前 key 成功，返回 1；如果当前 key 已经存在，则设置当前 key 失败，返回 0。

expire 设置过期时间，要注意的是 setnx 命令不能设置 key 的超时时间，只能通过 expire() 来对 key 设置。

使用步骤：

setnx(lockkey, 1) 如果返回 0，则说明占位失败；如果返回 1，则说明占位成功  
expire() 命令对 lockkey 设置超时时间，为的是避免死锁问题。  
执行完业务代码后，可以通过 delete 命令删除 key。  
public class DistributedLock \{

private final JedisPool jedisPool;

public DistributedLock(JedisPool jedisPool) \{  
        this.jedisPool = jedisPool;  
    \}

/\*\*  
     \* 加锁  
     \* @param lockName       锁的key  
     \* @param acquireTimeout 获取超时时间  
     \* @param timeout        锁的超时时间  
     \* @return 锁标识  
     \*/  
    public String lockWithTimeout(String lockName, long acquireTimeout, long timeout) \{  
        Jedis conn = null;  
        String retIdentifier = null;  
        try \{  
            // 获取连接  
            conn = jedisPool.getResource();  
            // 随机生成一个value  
            String identifier = UUID.randomUUID().toString();  
            // 锁名，即key值  
            String lockKey = "lock:" + lockName;  
            // 超时时间，上锁后超过此时间则自动释放锁  
            int lockExpire = (int) (timeout / 1000);

// 获取锁的超时时间，超过这个时间则放弃获取锁  
            long end = System.currentTimeMillis() + acquireTimeout;  
            while (System.currentTimeMillis() < end) \{  
                if (conn.setnx(lockKey, identifier) == 1) \{  
                    conn.expire(lockKey, lockExpire);  
                    // 返回value值，用于释放锁时间确认  
                    retIdentifier = identifier;  
                    return retIdentifier;  
                \}  
                // 返回-1代表key没有设置超时时间，为key设置一个超时时间  
                if (conn.ttl(lockKey) == -1) \{  
                    conn.expire(lockKey, lockExpire);  
                \}

try \{  
                    Thread.sleep(10);  
                \} catch (InterruptedException e) \{  
                    Thread.currentThread().interrupt();  
                \}  
            \}  
        \} catch (JedisException e) \{  
            e.printStackTrace();  
        \} finally \{  
            if (conn != null) \{  
                conn.close();  
            \}  
        \}  
        return retIdentifier;  
    \}

/\*\*  
     \* 释放锁  
     \* @param lockName   锁的key  
     \* @param identifier 释放锁的标识  
     \* @return  
     \*/  
    public boolean releaseLock(String lockName, String identifier) \{  
        Jedis conn = null;  
        String lockKey = "lock:" + lockName;  
        boolean retFlag = false;  
        try \{  
            conn = jedisPool.getResource();  
            while (true) \{  
                // 监视lock，准备开始事务  
                conn.watch(lockKey);  
                // 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁  
                if (identifier.equals(conn.get(lockKey))) \{  
                    Transaction transaction = conn.multi();  
                    transaction.del(lockKey);  
                    List<Object> results = transaction.exec();  
                    if (results == null) \{  
                        continue;  
                    \}  
                    retFlag = true;  
                \}  
                conn.unwatch();  
                break;  
            \}  
        \} catch (JedisException e) \{  
            e.printStackTrace();  
        \} finally \{  
            if (conn != null) \{  
                conn.close();  
            \}  
        \}  
        return retFlag;  
    \}  
\}  
这个方案其实是可以解决日常工作中的需求的，但从技术方案的探讨上来说，可能还有一些可以完善的地方。比如，如果在第一步 setnx 执行成功后，在 expire() 命令执行成功前，发生了宕机的现象，那么就依然会出现死锁的问题，所以如果要对其进行完善的话，可以使用 redis 的 setnx()、get() 和 getset() 方法来实现分布式锁。

另外代码中使用了watch监控lockkey，保障事务执行时不能有其他线程对lockkey进行操作。参考博文：Redis - 事务操作与详解

② 基于REDIS 的 SETNX()、GET()、GETSET()方法做分布式锁

使用步骤：

setnx(lockkey, 当前时间+过期超时时间)，如果返回 1，则获取锁成功；如果返回 0 则没有获取到锁，转向 2。  
get(lockkey) 获取值 oldExpireTime ，并将这个 value 值与当前的系统时间进行比较，如果小于当前系统时间，则认为这个锁已经超时，可以允许别的请求重新获取，转向 3。  
计算 newExpireTime = 当前时间+过期超时时间，然后 getset(lockkey, newExpireTime) 会返回当前 lockkey 的值currentExpireTime。  
判断 currentExpireTime 与 oldExpireTime 是否相等，如果相等，说明当前 getset 设置成功，获取到了锁。如果不相等，说明这个锁又被别的请求获取走了，那么当前请求可以直接返回失败，或者继续重试。  
在获取到锁之后，当前线程可以开始自己的业务处理，当处理完毕后，比较自己的处理时间和对于锁设置的超时时间，如果小于锁设置的超时时间，则直接执行 delete 释放锁；如果大于锁设置的超时时间，则不需要再锁进行处理。  
注意，上面两种方式均未考虑持锁人标识！

③ 基于Set命令的分布式锁

如下所示：

set user\_key user\_value NX PX 100  
1  
redis从2.6.12版本开始，SET命令才支持这些参数：

NX：只在在键不存在时，才对键进行设置操作，SET key value NX 效果等同于 SETNX key value

PX millisecond：设置键的过期时间为millisecond毫秒，当超过这个时间后，设置的键会自动失效

上述代码示例是指，当redis中不存在user\_key这个键的时候，才会去设置一个user\_key键，并且给这个键的值设置为 user\_value，且这个键的存活时间为100ms。

为什么这个命令可以帮我们实现锁机制呢？

因为这个命令是只有在某个key不存在的时候，才会执行成功。那么当多个进程同时并发的去设置同一个key的时候，就永远只会有一个进程成功。

当某个进程设置成功之后，就可以去执行业务逻辑了，等业务逻辑执行完毕之后，再去进行解锁。

解锁很简单，只需要删除这个key就可以了，不过删除之前需要判断，这个key对应的value是当初自己设置的那个。

故而为了保证解锁的原子操作可以使用lua脚本，如下示例：

if redis.call("get",KEYS\[1\]==ARGV\[1\] then  
    return redis.call("del",KEYS\[1\])  
)  
else return 0  
end  
1  
2  
3  
4  
5  
Java代码实现如下：

import redis.clients.jedis.Jedis;

import java.util.Arrays;  
import java.util.UUID;  
import java.util.concurrent.TimeUnit;  
import java.util.concurrent.locks.Condition;  
import java.util.concurrent.locks.Lock;

/\*\*  
 \* Created by Janus on 2019/1/28.  
 \*/  
public class RedisDistribute implements Lock \{

private ThreadLocal<String> lockContext = new ThreadLocal<>();

//默认锁的超时时间  
    private long time=100;

private Thread exclusiveOwnerThread;

public RedisDistribute()\{\};

@Override  
    public void lock() \{  
        while (!tryLock())\{  
            try\{  
                Thread.sleep(100);  
            \}catch (InterruptedException e)\{  
                e.printStackTrace();  
            \}  
        \}

@Override  
    public void lockInterruptibly() throws InterruptedException \{

if (Thread.interrupted())\{  
            throw  new InterruptedException();  
        \}  
        while (!tryLock())\{  
            try\{  
                Thread.sleep(100);  
            \}catch (InterruptedException e)\{  
                e.printStackTrace();  
            \}  
        \}

@Override  
    public boolean tryLock() \{  
        try \{  
            return tryLock(time,TimeUnit.MILLISECONDS);  
        \} catch (InterruptedException e) \{  
            e.printStackTrace();  
        \}  
        return false;  
    \}

@Override  
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException \{  
        //为每个锁的持有人分配唯一id  
        String id= UUID.randomUUID().toString();  
        Thread t = Thread.currentThread();  
        Jedis jedis = jedis.getClient();  
        if ("OK".equals(jedis.set("lock",id,"NX","PX",unit.toMillis(time))))\{  
            lockContext.set(id);  
            setExclusiveOwnerThread(t);  
            return true;  
        \}else if(exclusiveOwnerThread==t)\{//当前线程已经获取了锁，可重入  
            return true;  
        \}  
        return false;  
    \}

@Override  
    public void unlock() \{  
        String script = null;  
        try\{  
            Jedis jedis = jedis.getClient();  
            script = inputStream2String(getClass().getResourceAsStream("/redis.script"));  
            if(lockContext.get()==null)\{  
                return;  
            \}  
            //释放锁  
            jedis.eval(script, Arrays.asList("lock"),Arrays.asList(lockContext.get()));  
            lockContext.remove();  
        \}catch (Exception e)\{  
            e.printStackTrace();  
        \}  
          
    \}

@Override  
    public Condition newCondition() \{  
        return null;  
    \}

public void setExclusiveOwnerThread(Thread exclusiveOwnerThread) \{  
        this.exclusiveOwnerThread = exclusiveOwnerThread;  
    \}  
\}  
这种实现方式有3大要点（也是面试概率非常高的地方）：

set命令要用set key value nx px milliseconds；  
value要具有唯一性；  
释放锁时要验证value值，不能误解锁；  
事实上这类琐最大的缺点就是它加锁时只作用在一个Redis节点上，即使Redis通过sentinel保证高可用，如果这个master节点由于某些原因发生了主从切换，那么就会出现锁丢失的情况：

在Redis的master节点上拿到了锁；

但是这个加锁的key还没有同步到slave节点；

master故障，发生故障转移，slave节点升级为master节点；

导致锁丢失。

④ 基于 REDLOCK 做分布式锁

上面redis做分布式锁存在主从的单点失败问题，描述如下：

客户端A从mater获取锁；  
master将锁同步到slave之前宕机；  
slave被晋升为master；  
客户端B取得已被客户端A获取的锁  
RedLock算法

使用多个master节点(单数)；  
客户端记录当前时间，然后向master节点请求锁；  
统计获取锁成功的个数>N/2+1，锁开始使用时间(当前时间-开始获取锁的时间)小于锁释放时间时获取锁成功，否则获取锁失败，向所有节点发送释放锁的命令；  
获取成功后新锁的释放时间=原锁释放时间-获取锁的消耗时间；  
请求锁失败后随机设置重试时间。  
详细解释如下：

在Redis的分布式环境中，我们假设有N个Redis master。这些节点完全互相独立，不存在主从复制或者其他集群协调机制。我们确保将在N个实例上使用与在Redis单实例下相同方法获取和释放锁。现在我们假设有5个Redis master节点，同时我们需要在5台服务器上面运行这些Redis实例，这样保证他们不会同时都宕掉。

为了取到锁，客户端应该执行以下操作:

获取当前Unix时间，以毫秒为单位。

依次尝试从5个实例，使用相同的key和具有唯一性的value（例如UUID）获取锁。当向Redis请求获取锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为10秒，则超时时间应该在5-50毫秒之间。这样可以避免服务器端Redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试去另外一个Redis实例请求获取锁。

客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（N/2+1，这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。

如果取到了锁，key的真正有效时间等于失效时间减去获取锁所使用的时间（步骤3计算的结果）。

如果因为某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁）。

⑤ Redlock源码

redisson已经有对redlock算法封装，接下来对其用法进行简单介绍，并对核心源码进行分析（假设5个redis实例）。

POM依赖

<dependency>  
    <groupId>org.redisson</groupId>  
    <artifactId>redisson</artifactId>  
    <version>3.3.2</version>  
</dependency>  
用法  
首先，我们来看一下redission封装的redlock算法实现的分布式锁用法，非常简单，跟重入锁（ReentrantLock）有点类似：

唯一ID

实现分布式锁的一个非常重要的点就是set的value要具有唯一性，redisson的value是怎样保证value的唯一性呢？答案是UUID+threadId。入口在redissonClient.getLock(“REDLOCK\_KEY”)，源码在Redisson.java和RedissonLock.java中：

protected final UUID id = UUID.randomUUID();  
String getLockName(long threadId)\{  
 return id +":"+ threadId;  
\}  
获取锁

获取锁的代码为 redLock.tryLock()或者 redLock.tryLock(500,10000,TimeUnit.MILLISECONDS)，两者的最终核心源码都是下面这段代码，只不过前者获取锁的默认租约时间leaseTime是 LOCK\_EXPIRATION\_INTERVAL\_SECONDS，即30s：

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) \{  
        internalLockLeaseTime = unit.toMillis(leaseTime);  
// 获取锁时向5个redis实例发送的命令  
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,  
          
// 首先分布式锁的KEY不能存在，如果确实不存在，那么执行hset命令  
//（hset REDLOCK\_KEY uuid+threadId 1），并通过pexpire设置失效时间（也是锁的租约时间）  
                  "if (redis.call('exists', KEYS\[1\]) == 0) then " +  
                      "redis.call('hset', KEYS\[1\], ARGV\[2\], 1); " +  
                      "redis.call('pexpire', KEYS\[1\], ARGV\[1\]); " +  
                      "return nil; " +  
                  "end; " +  
// 如果分布式锁的KEY已经存在，并且value也匹配，表示是当前线程持有的锁，那么重入次数加1，并且设置失效时间  
                  "if (redis.call('hexists', KEYS\[1\], ARGV\[2\]) == 1) then " +  
                      "redis.call('hincrby', KEYS\[1\], ARGV\[2\], 1); " +  
                      "redis.call('pexpire', KEYS\[1\], ARGV\[1\]); " +  
                      "return nil; " +  
                  "end; " +  
// 获取分布式锁的KEY的失效时间毫秒数  
                  "return redis.call('pttl', KEYS\[1\]);",  
                    
// 这三个参数分别对应KEYS\[1\]，ARGV\[1\]和ARGV\[2\]  
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));  
    \}  
获取锁的命令中，

KEYS\[1\]就是 Collections.singletonList(getName())，表示分布式锁的key，即REDLOCK\_KEY；

ARGV\[1\]就是internalLockLeaseTime，即锁的租约时间，默认30s；

ARGV\[2\]就是getLockName(threadId)，是获取锁时set的唯一值，即UUID+threadId：

释放锁

释放锁的代码为redLock.unlock()，核心源码如下：

protected RFuture<Boolean> unlockInnerAsync(long threadId) \{  
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL\_BOOLEAN,  
// 如果分布式锁KEY不存在，那么向channel发布一条消息  
                "if (redis.call('exists', KEYS\[1\]) == 0) then " +  
                    "redis.call('publish', KEYS\[2\], ARGV\[1\]); " +  
                    "return 1; " +  
                "end;" +  
                // 如果分布式锁存在，但是value不匹配，表示锁已经被占用，那么直接返回  
                "if (redis.call('hexists', KEYS\[1\], ARGV\[3\]) == 0) then " +  
                    "return nil;" +  
                "end; " +  
// 如果就是当前线程占有分布式锁，那么将重入次数减1  
                "local counter = redis.call('hincrby', KEYS\[1\], ARGV\[3\], -1); " +  
// 重入次数减1后的值如果大于0，表示分布式锁有重入过，那么只设置失效时间，还不能删除  
                "if (counter > 0) then " +  
                    "redis.call('pexpire', KEYS\[1\], ARGV\[2\]); " +  
                    "return 0; " +  
                "else " +  
// 重入次数减1后的值如果为0，表示分布式锁只获取过1次，那么删除这个KEY，并发布解锁消息  
                    "redis.call('del', KEYS\[1\]); " +  
                    "redis.call('publish', KEYS\[2\], ARGV\[1\]); " +  
                    "return 1; "+  
                "end; " +  
                "return nil;",  
// 这5个参数分别对应KEYS\[1\]，KEYS\[2\]，ARGV\[1\]，ARGV\[2\]和ARGV\[3\]  
                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));