分布式锁

什么场景下会使用分布式锁？

单机应用架构中，秒杀案例使用ReentrantLcok或者synchronized来达到秒杀商品互斥的目的。然而在分布式系统中，会存在多台机器并行去实现同一个功能。也就是说，在多进程中，如果还使用以上JDK提供的进程锁，来并发访问数据库资源就可能会出现商品超卖的情况。因此，需要我们来实现自己的分布式锁。

分布式锁的缺点

比如你用zookeeper实现了分布式锁，他有一个缺点就是并发量上不去，因为他是串行的嘛，如果说可以优化的话(优化就会带来复杂性)，可以借鉴ConcurrentHashMap的分段锁的思想，比如1000个库存，分成20个段，每段50个库存。其实最好的是通过redis原子操作加异步队列

实现一个分布式锁应该具备的特性：

• 高可用、高性能的获取锁与释放锁
• 在分布式系统环境下，一个方法或者变量同一时间只能被一个线程操作
• 具备锁失效机制，网络中断或宕机无法释放锁时，锁必须被删除，防止死锁
• 具备阻塞锁特性，即没有获取到锁，则继续等待获取锁
• 具备非阻塞锁特性，即没有获取到锁，则直接返回获取锁失败
• 具备可重入特性，一个线程中可以多次获取同一把锁，比如一个线程在执行一个带锁的方法，该方法中又调用了另一个需要相同锁的方法，则该线程可以直接执行调用的方法，而无需重新获得锁

关于分布式锁几种实现方式：

• 基于数据库实现分布式锁
• 基于 Redis 实现分布式锁
• 基于 Zookeeper 实现分布式锁

---

Zookeeper实现分布式锁

前两种对于分布式生产环境来说并不是特别推荐，高并发下数据库锁性能太差，Redis在锁时间限制和缓存一致性存在一定问题。这里我们重点介绍一下 Zookeeper 如何实现分布式锁。

实现原理

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务，它内部是一个分层的文件系统目录树结构，规定同一个目录下只能存在唯一文件名。

ZooKeeper数据模型与文件系统目录树(源自网络)

• 数据模型
  PERSISTENT 持久化节点，节点创建后，不会因为会话失效而消失
  EPHEMERAL 临时节点， 客户端session超时此类节点就会被自动删除
  EPHEMERAL_SEQUENTIAL 临时自动编号节点
  PERSISTENT_SEQUENTIAL 顺序自动编号持久化节点，这种节点会根据当前已存在的节点数自动加 1
• 监视器（watcher）
  当创建一个节点时，可以注册一个该节点的监视器，当节点状态发生改变时，watch被触发时，ZooKeeper将会向客户端发送且仅发送一条通知，因为watch只能被触发一次。
  根据zookeeper的这些特性，我们来看看如何利用这些特性来实现分布式锁：

1. 创建一个锁目录lock
2. 线程A获取锁会在lock目录下，创建临时顺序节点
3. 获取锁目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁
4. 线程B创建临时节点并获取所有兄弟节点，判断自己不是最小节点，设置监听(watcher)比自己次小的节点
5. 线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是最小的节点，获得锁

代码分析

尽管ZooKeeper已经封装好复杂易出错的关键服务，将简单易用的接口和性能高效、功能稳定的系统提供给用户。但是如果让一个普通开发者去手撸一个分布式锁还是比较困难的，在秒杀案例中我们直接使用 Apache 开源的curator 开实现 Zookeeper 分布式锁。
这里我们使用以下版本，截止目前最新版4.0.1：

<!-- zookeeper 分布式锁、注意zookeeper版本  这里对应的是3.4.6-->
<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>2.10.0</version>
</dependency>

首先，我们看下InterProcessLock接口中的几个方法：

/**
* 获取锁、阻塞等待、可重入
*/
public void acquire() throws Exception;

/**
* 获取锁、阻塞等待、可重入、超时则获取失败
*/
public boolean acquire(long time, TimeUnit unit) throws Exception;

/**
* 释放锁
*/
public void release() throws Exception;

/**
* Returns true if the mutex is acquired by a thread in this JVM
*/
boolean isAcquiredInThisProcess();
获取锁：
//获取锁
public void acquire() throws Exception
    {
        if ( !internalLock(-1, null) )
        {
            throw new IOException("Lost connection while trying to acquire lock: " + basePath);
        }
    }
具体实现：
private boolean internalLock(long time, TimeUnit unit) throws Exception
    {
        /*
         实现同一个线程可重入性，如果当前线程已经获得锁，
         则增加锁数据中lockCount的数量(重入次数)，直接返回成功
        */
        //获取当前线程
        Thread currentThread = Thread.currentThread();
        //获取当前线程重入锁相关数据
        LockData lockData = threadData.get(currentThread);
        if ( lockData != null )
        {
            //原子递增一个当前值，记录重入次数，后面锁释放会用到
            lockData.lockCount.incrementAndGet();
            return true;
        }
        //尝试连接zookeeper获取锁
        String lockPath = internals.attemptLock(time, unit, getLockNodeBytes());
        if ( lockPath != null )
        {
            //创建可重入锁数据，用于记录当前线程重入次数
            LockData newLockData = new LockData(currentThread, lockPath);
            threadData.put(currentThread, newLockData);
            return true;
        }
        //获取锁超时或者zk通信异常返回失败
        return false;
    }
Zookeeper获取锁实现：
 String attemptLock(long time, TimeUnit unit, byte[] lockNodeBytes) throws Exception
    {    
        //获取当前时间戳
        final long      startMillis = System.currentTimeMillis();
        //如果unit不为空(非阻塞锁)，把当前传入time转为毫秒
        final Long      millisToWait = (unit != null) ? unit.toMillis(time) : null;
        //子节点标识
        final byte[]    localLockNodeBytes = (revocable.get() != null) ? new byte[0] : lockNodeBytes;
        //尝试次数
        int             retryCount = 0;

        String          ourPath = null;
        boolean         hasTheLock = false;
        boolean         isDone = false;
        //自旋锁，循环获取锁
        while ( !isDone )
        {
            isDone = true;

            try
            {
                //在锁节点下创建临时且有序的子节点，例如:_c_008c1b07-d577-4e5f-8699-8f0f98a013b4-lock-000000001
                ourPath = driver.createsTheLock(client, path, localLockNodeBytes);
                //如果当前子节点序号最小，获得锁则直接返回，否则阻塞等待前一个子节点删除通知(release释放锁)
                hasTheLock = internalLockLoop(startMillis, millisToWait, ourPath);
            }
            catch ( KeeperException.NoNodeException e )
            {
                //异常处理，如果找不到节点，这可能发生在session过期等时，因此，如果重试允许，只需重试一次即可
                if ( client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                {
                    isDone = false;
                }
                else
                {
                    throw e;
                }
            }
        }
        //如果获取锁则返回当前锁子节点路径
        if ( hasTheLock )
        {
            return ourPath;
        }

        return null;
    }
判断是否为最小节点：
  private boolean internalLockLoop(long startMillis, Long millisToWait, String ourPath) throws Exception
    {
        boolean     haveTheLock = false;
        boolean     doDelete = false;
        try
        {
            if ( revocable.get() != null )
            {
                client.getData().usingWatcher(revocableWatcher).forPath(ourPath);
            }
            //自旋获取锁
            while ( (client.getState() == CuratorFrameworkState.STARTED) && !haveTheLock )
            {
                //获取所有子节点集合
                List<String>        children = getSortedChildren();
                //判断当前子节点是否为最小子节点
                String              sequenceNodeName = ourPath.substring(basePath.length() + 1); // +1 to include the slash

                PredicateResults    predicateResults = driver.getsTheLock(client, children, sequenceNodeName, maxLeases);
                //如果是最小节点则获取锁
                if ( predicateResults.getsTheLock() )
                {
                    haveTheLock = true;
                }
                else
                {
                    //获取前一个节点，用于监听
                    String  previousSequencePath = basePath + "/" + predicateResults.getPathToWatch();

                    synchronized(this)
                    {
                        try 
                        {
                            //这里使用getData()接口而不是checkExists()是因为，如果前一个子节点已经被删除了那么会抛出异常而且不会设置事件监听器，而checkExists虽然也可以获取到节点是否存在的信息但是同时设置了监听器，这个监听器其实永远不会触发，对于Zookeeper来说属于资源泄露
                            client.getData().usingWatcher(watcher).forPath(previousSequencePath);
                            if ( millisToWait != null )
                            {
                                millisToWait -= (System.currentTimeMillis() - startMillis);
                                startMillis = System.currentTimeMillis();
                                //如果设置了获取锁等待时间
                                if ( millisToWait <= 0 )
                                {
                                    doDelete = true;    // 超时则删除子节点
                                    break;
                                }
                                //等待超时时间
                                wait(millisToWait);
                            }
                            else
                            {
                                wait();//一直等待
                            }
                        }
                        catch ( KeeperException.NoNodeException e ) 
                        {
                            // it has been deleted (i.e. lock released). Try to acquire again
                            //如果前一个子节点已经被删除则deException，只需要自旋获取一次即可
                        }
                    }
                }
            }
        }
        catch ( Exception e )
        {
            ThreadUtils.checkInterrupted(e);
            doDelete = true;
            throw e;
        }
        finally
        {
            if ( doDelete )
            {
                deleteOurPath(ourPath);//获取锁超时则删除节点
            }
        }
        return haveTheLock;
    }
释放锁：
 public void release() throws Exception
    {

        Thread currentThread = Thread.currentThread();
        LockData lockData = threadData.get(currentThread);
        //没有获取锁，你释放个球球，如果为空抛出异常
        if ( lockData == null )
        {
            throw new IllegalMonitorStateException("You do not own the lock: " + basePath);
        }
        //获取重入数量
        int newLockCount = lockData.lockCount.decrementAndGet();
        //如果重入锁次数大于0，直接返回
        if ( newLockCount > 0 )
        {
            return;
        }
        //如果重入锁次数小于0，抛出异常
        if ( newLockCount < 0 )
        {
            throw new IllegalMonitorStateException("Lock count has gone negative for lock: " + basePath);
        }
        try
        {
            //释放锁
            internals.releaseLock(lockData.lockPath);
        }
        finally
        {
            //移除当前线程锁数据
            threadData.remove(currentThread);
        }
    }

测试案例
为了更好的理解其原理和代码分析中获取锁的过程，这里我们实现一个简单的Demo：

/**
 * 基于curator的zookeeper分布式锁
 */
public class CuratorUtil {
    private static String address = "192.168.1.180:2181";

    public static void main(String[] args) {
        //1、重试策略：初试时间为1s 重试3次
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3); 
        //2、通过工厂创建连接
        CuratorFramework client = CuratorFrameworkFactory.newClient(address, retryPolicy);
        //3、开启连接
        client.start();
        //4 分布式锁
        final InterProcessMutex mutex = new InterProcessMutex(client, "/curator/lock"); 
        //读写锁
        //InterProcessReadWriteLock readWriteLock = new InterProcessReadWriteLock(client, "/readwriter");

        ExecutorService fixedThreadPool = Executors.newFixedThreadPool(5);

        for (int i = 0; i < 5; i++) {
            fixedThreadPool.submit(new Runnable() {
                @Override
                public void run() {
                    boolean flag = false;
                    try {
                        //尝试获取锁，最多等待5秒
                        flag = mutex.acquire(5, TimeUnit.SECONDS);
                        Thread currentThread = Thread.currentThread();
                        if(flag){
                            System.out.println("线程"+currentThread.getId()+"获取锁成功");
                        }else{
                            System.out.println("线程"+currentThread.getId()+"获取锁失败");
                        }
                        //模拟业务逻辑，延时4秒
                        Thread.sleep(4000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    } finally{
                        if(flag){
                            try {
                                mutex.release();
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                    }
                }
            });
        }
    }
}

这里我们开启5个线程，每个线程获取锁的最大等待时间为5秒，为了模拟具体业务场景，方法中设置4秒等待时间。开始执行main方法，通过ZooInspector监控/curator/lock下的节点如下图：

对，没错，设置4秒的业务处理时长就是为了观察生成了几个顺序节点。果然如案例中所述，每个线程都会生成一个节点并且还是有序的。
观察控制台，我们会发现只有两个线程获取锁成功，另外三个线程超时获取锁失败会自动删除节点。线程执行完毕我们刷新一下/curator/lock节点，发现刚才创建的五个子节点已经不存在了。

---

基于数据库的分布式锁

实现方式

数据库的乐观锁(通过版本号)或者悲观锁(通过for update)
还可以通过mysql的unique key来实现分布式锁，在mysql中插入一条记录，表明获取锁。删除一条记录，表明释放锁。 且在mysql表中设置一个unique key字段， 当有一台机器获得锁后， 其他机器无法获取。
这个有几个问题：

1.  如果一台机器获得锁，在释放锁之前进程挂了， 那么其他机器无法获取到锁。 可以引入锁有效时间的概念，超时后，删除记录，释放锁。

2.  万一获取锁的操作失败了，就直接做错误处理， 也不太好。 可以引入循环重试的方式来解决，控制重试次数。

参考https://blog.csdn.net/stpeace/article/details/84679324

基于数据库分布式锁注意的地方

• 基于数据库表的方式需要集群中多个节点的服务器时钟同步。
• 基于mysql数据库时需要在JDBC连接地址中增加时区配置，serverTimezone=GMT%2b8。
  关于脑裂问题：
• 由于网络、进程假死、中间件可能导致的脑裂问题在一定程度上不可避免，所以我们需要做一种权衡，即脑裂发生时是否可以避免问题进一步扩大。基于JDBC的分布式锁住要避免多个节点同时对事务表进行扫描，两者都是进行数据库操作，因为它们都基于数据库所以能够在一定程度上避免脑裂导致的集群处理问题，但是基于zookeeper,redis的方式可能会导致真实的脑裂发生

使用MYSQL的GET_LOCK函数

能够锁定用户定义的字符串。当然你不应该使用它来锁定行（MySQL已经做得很好）但可能锁定外部资源。

https://www.xaprb.com/blog/2006/07/26/how-to-coordinate-distributed-work-with-mysqls-get_lock/

基于 Redis 的分布式锁（可以单机也可以集群）

首先说明一下setnx()命令，setnx的含义就是SET if Not Exists，其主要有两个参数 setnx(key, value)。该方法是原子的，如果key不存在，则设置当前key成功，返回1；如果当前key已经存在，则设置当前key失败，返回0。value是System.currentTimeMillis() (获取锁的时间)+锁持有的时间。但是要注意的是setnx命令不能设置key的超时时间，只能通过expire()来对key设置。

具体的使用步骤如下:

1. setnx(lockkey, 1) 如果返回0，则说明占位失败；如果返回1，则说明占位成功
2. expire()命令对lockkey设置超时时间，为的是避免死锁问题。
3. 执行完业务代码后，可以通过delete命令删除key。
   为了保证在某个Redis节点不可用的时候算法能够继续运行，这个获取锁的操作还有一个超时时间(timeOut)，它要远小于锁的有效时间（几十毫秒量级）

   上边的解决方案可能存在的问题

   这个方案其实是可以解决日常工作中的需求的，但从技术方案的探讨上来说，可能还有一些可以完善的地方。比如，如果在第一步setnx执行成功后，在expire()命令执行成功前，发生了宕机的现象，那么就依然会出现死锁的问题，所以如果要对其进行完善的话，可以使用redis的setnx()、get()和getset()方法来实现分布式锁。

getSet(key,value)的命令会返回key对应的value，然后再把key原来的值更新为value。也就是说getSet()返回的是已过期的时间戳。如果这个已过期的时间戳等于currentValue，说明获取锁成功。

假设客户端A一开始持有锁，保存在redis中的value(时间戳)等于T1。 这时候客户端A的锁已经过期，那么C，D客户端就可以开始争抢锁了。currentValue是T1，C客户端的value是T2，D客户端的value是T3。首先C客户端进入到String oldValue = jedis.getSet(realKey, value);这行代码，获得的oldValue是T1，同时也会把realKey对应的value更新为T2。再执行后续的代码，oldValue等于currentValue，那么客户端C获取锁成功。接着D客户端也执行到了String oldValue = jedis.getSet(realKey, value);这行代码，获取的oldValue是T2，同时也会把realKey对应的value更新为T3。由于oldValue不等于currentValue，那么客户端D获取锁失败。

public boolean lock(KeyPrefix prefix, String key, String value) {
    Jedis jedis = null;
    Long lockWaitTimeOut = 200L;
    Long deadTimeLine = System.currentTimeMillis() + lockWaitTimeOut;

    try {
        jedis = jedisPool.getResource();
        String realKey = prefix.getPrefix() + key;

        for (;;) {
            if (jedis.setnx(realKey, value) == 1) {
                return true;
            }

            String currentValue = jedis.get(realKey);

            // if lock is expired
            if (!StringUtils.isEmpty(currentValue) &&
                    Long.valueOf(currentValue) < System.currentTimeMillis()) {
                // gets last lock time
                String oldValue = jedis.getSet(realKey, value);

                if (!StringUtils.isEmpty(oldValue) && oldValue.equals(currentValue)) {
                    return true;
                }
            }

            lockWaitTimeOut = deadTimeLine - System.currentTimeMillis();

            if (lockWaitTimeOut <= 0L) {
                return false;
            }
        }
    } finally {
        returnToPool(jedis);
    }
}

单机Redis分布式锁

不同的客户端通过生成随机字符串对制定的Key进行set if not exists + ttl 操作来进行抢占,通过key的TTL时间来决定持有锁的时间. 然后通过LUA脚本执行事务操作进行Compare and delete进行释放锁.其中的TTL和本机时钟有关.

• 释放锁示例
  lua脚本内容如下

  if redis.call("get",KEYS[1]) == ARGV[1] then
      return redis.call("del",KEYS[1])
  else
      return 0
  end

这段Lua脚本在执行的时候要把的lockValue作为ARGV[1]的值传进去，把lockKey作为KEYS[1]的值传进去。现在来看看解锁的java代码

public void unlock() {
    // 使用lua脚本进行原子删除操作
    String checkAndDelScript = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
                                "return redis.call('del', KEYS[1]) " +
                                "else " +
                                "return 0 " +
                                "end";
    jedis.eval(checkAndDelScript, 1, lockKey, lockValue);
}

分布式锁丢失的问题

官方其实并不推荐这种方式，因为它在集群模式下会产生锁丢失的问题 —— 在主从发生切换的时候。官方推荐的分布式锁叫 RedLock，作者认为这个算法较为安全，推荐我们使用。不过掌阅这边一直还使用上面最简单的分布式锁，为什么我们不去使用 RedLock 呢，因为它的运维成本会高一些，需要 3 台以上独立的 Redis 实例，用起来要繁琐一些。另外呢 Redis 集群发生主从切换的概率也并不高，即使发生了主从切换出现锁丢失的概率也很低，因为主从切换往往都有一个过程，这个过程的时间通常会超过锁的过期时间，也就不会发生锁的异常丢失。还有呢就是分布式锁遇到锁冲突的机会也不多，这正如一个公司里明星程序员也比较有限一样，总是遇到锁排队那说明结构上需要优化。
https://juejin.im/post/5d0f3c2be51d45595319e355

集群Redis分布式锁

Redis 的作者提供了RedLock 的算法来实现一个分布式锁。

• 加锁
  RedLock算法加锁步骤如下:

1. 获取当前Unix时间，以毫秒为单位。
2. 依次尝试从N个实例，使用相同的key和随机值获取锁。在步骤2，当向Redis设置锁时,客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。例如你的锁自动失效时间为10秒，则超时时间应该在5-50毫秒之间。这样可以避免服务器端Redis已经挂掉的情况下，客户端还在死死地等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试另外一个Redis实例。
3. 客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。
4. 如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）。
5. 如果因为某些原因，获取锁失败（没有在至少N/2+1个Redis实例取到锁或者取锁时间已经超过了有效时间），客户端应该在所有的Redis实例上进行解锁（即便某些Redis实例根本就没有加锁成功）。

• 解锁
  向所有的Redis实例发送释放锁命令即可，不用关心之前有没有从Redis实例成功获取到锁.

RedLock的设计是糟糕的！！！！

https://martin.kleppmann.com/2016/02/08/how-to-do-distributed-locking.html
这篇文章主要阐述了RedLock的缺点，作者说RedLock不适合用作分布式锁.从效率上说,RedLock不好,从安全性上讲,也得不到保证.比如有两个进程竞争锁,第一个进程首先抢到了锁,但是它因各种原因不工作了,过了租期,那么这个时候,另外一个进程就可以抢到锁,这个时候第一个进程又好了,他想向Storage中写入记录.我们可以通过版本号来防止这种情况产生.比如可以使用 ZooKeeper 作为锁服务，您可以使用 zxid 或 znode 版本号。但是在RedLock中并没有这种机制,

However, Redlock is not like this. Its safety depends on a lot of timing assumptions: it assumes that all Redis nodes hold keys for approximately the right length of time before expiring; that the network delay is small compared to the expiry duration; and that process pauses are much shorter than the expiry duration.

结论就是,如果不需要保证正确性,使用 Redis 的简单的单节点锁定算法(条件 set-If-not-exists 用于获得锁，原子 delete-If-value-matches 用于释放锁).

如果需要锁以保证正确性，请不要使用 Redlock。 相反，请使用一个适当的共识系统，比如 ZooKeeper

实际秒杀场景中的应用

实际上抢购，好像不用分布式锁，？？？？而是直接将库存放入到redis，是否结束标记放入到内存中，通过内存标记和redis中的decr()预减库存，然后将秒杀消息入队到消息队列中，最后消费消息并落地到DB中

分布式锁的重入该如何实现？

不推荐使用可重入锁，它加重了客户端的复杂性，在编写业务方法时注意在逻辑结构上进行调整完全可以不使用可重入锁。 –老钱

其实就是基于ThreadLocal和引用计数。

https://juejin.im/post/5bbb0d8df265da0abd3533a5#comment

如何确保过期时间大于业务执行时间

为了避免这个问题，Redis 分布式锁不要用于较长时间的任务。如果真的偶尔出现了，数据出现的小波错乱可能需要人工介入解决。 –老钱

可以加一个定时线程增加过期时间,比如在RedLock会起一个后台watch dog线程，检测到程序还没执行完的话，会继续延长过期时间。

客户端在处理请求时加锁没加成功怎么办？

• 直接抛出异常，通知用户稍后重试；
• sleep 一会再重试；
• 将请求转移至延时队列，过一会再试；

codis(好牛逼)

Codis 在设计上相比 Redis Cluster 官方集群方案要简单很多，因为它将分布式的问题交
给了第三方 zk/etcd 去负责，自己就省去了复杂的分布式一致性代码的编写维护工作。而
Redis Cluster 的内部实现非常复杂，它为了实现去中心化，混合使用了复杂的 Raft 和
Gossip 协议，还有大量的需要调优的配置参数，当集群出现故障时，维护人员往往不知道从
何处着手。
https://github.com/CodisLabs/codis

参考

https://juejin.im/post/5b737b9b518825613d3894f4
https://blog.csdn.net/u010963948/article/details/79006572 非常好的一篇文章，值得多看