原理分析

最近看到好多博主都在推分布式锁，实现方式很多，基于db、redis、zookeeper。zookeeper方式实现起来比较繁琐，这里我们就谈谈基于redis实现分布式锁的正确实现方式。

背景

在很多互联网产品应用中，有些场景需要加锁处理，比如：秒杀，全局递增ID，楼层生成等等。大部分的解决方案是基于DB实现的，Redis为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对Redis的连接并不存在竞争关系。 其次Redis提供一些命令SETNX，GETSET，可以方便实现分布式锁机制。

Redis命令介绍

使用Redis实现分布式锁，有两个重要函数需要介绍。

SETNX命令（SET if Not Exists）

• 语法：

SETNX key value

• 功能：
  当且仅当 key 不存在，将 key 的值设为 value ，并返回1； 若给定的 key 已经存在，则 SETNX 不做任何动作，并返回0。

GETSET命令

• 语法：

GETSET key value

• 功能：
  将给定 key 的值设为 value ，并返回 key 的旧值 (old value)， 当 key 存在但不是字符串类型时，返回一个错误，当key不存在时，返回nil。

GET命令

• 语法：

GET key

• 功能：
  返回 key 所关联的字符串值，如果 key 不存在那么返回特殊值 nil 。

DEL命令

• 语法：

DEL key [KEY …]

• 功能：
  删除给定的一个或多个 key ,不存在的 key 会被忽略。

兵贵精，不在多。分布式锁，我们就依靠这四个命令。但在具体实现，还有很多细节，需要仔细斟酌，因为在分布式并发多进程中，任何一点出现差错，都会导致死锁，hold住所有进程。

加锁实现

SETNX 可以直接加锁操作，比如说对某个关键词foo加锁，客户端可以尝试 SETNX foo.lock <current unix time>。

• 如果返回1，表示客户端已经获取锁，可以往下操作，操作完成后，通过 DEL foo.lock 命令来释放锁。

• 如果返回0，说明foo已经被其他客户端上锁，如果锁是非堵塞的，可以选择返回调用。如果是堵塞调用，就需要进入下一个重试循环，直至成功获得锁或者重试超时。

理想是美好的，现实是残酷的。仅仅使用SETNX加锁带有竞争条件的，在某些特定的情况会造成死锁错误。

处理死锁

在上面的处理方式中，如果获取锁的客户端执行时间过长，进程被kill掉，或者因为其他异常崩溃，导致无法释放锁，就会造成死锁。所以，需要对加锁要做时效性检测。

因此，我们在加锁时，把当前时间戳作为value存入此锁中，通过当前时间戳和redis中的时间戳进行对比，如果超过一定差值，认为锁已经时效，防止锁无限期的锁下去。

但是，在大并发情况，如果同时检测锁失效，并简单粗暴的删除死锁，再通过SETNX上锁，可能会导致竞争条件的产生，即多个客户端同时获取锁。

情景描述如下：

1. C1获取锁，并崩溃。C2和C3调用SETNX上锁返回0后，获得foo.lock的时间戳，通过比对时间戳，发现锁超时。

2. C2 向foo.lock发送DEL命令。

3. C2 向foo.lock发送SETNX获取锁。

4. C3 向foo.lock发送DEL命令，此时C3发送DEL时，其实DEL掉的是C2的锁。

5. C3 向foo.lock发送SETNX获取锁。

此时C2和C3都获取了锁，产生竞争条件，如果在更高并发的情况，可能会有更多客户端获取锁。

所以，DEL锁的操作，不能直接使用在锁超时的情况下，幸好我们有GETSET方法，假设我们现在有另外一个客户端C4，看看如何使用GETSET方式，避免这种情况产生。

1. C1获取锁，并崩溃。C2和C3调用SETNX上锁返回0后，调用GET命令获得foo.lock的时间戳T1，通过比对时间戳，发现锁超时。

2. C4(调用SETNX上锁返回0后，调用GET命令获得foo.lock的时间戳T1，通过比对时间戳，发现锁超时)向foo.lock发送GESET命令，GETSET foo.lock 并得到foo.lock中老的时间戳T2。

• 如果T1=T2，说明C4获得锁。

• 如果T1!=T2，说明C4之前有另外一个客户端C5通过调用GETSET方式获取并更改了时间戳，C4未获得锁。只能进入下次循环中。

时间戳问题

我们看到foo.lock的value值为时间戳，所以要在多客户端情况下，保证锁有效，一定要同步各服务器的时间。如果各服务器间，时间有差异，时间不一致的客户端，在判断锁超时，就会出现偏差，从而产生竞争条件。锁的超时与否，严格依赖时间戳。

锁覆盖问题

现在唯一的问题是，C4设置foo.lock的新时间戳，是否会对C5获取得锁产生影响？

其实我们可以看到C4和C5只有在调用GET命令获得foo.lock的时间戳，通过比对时间戳，发现锁超时后，几乎同时调用GETSET方式获取锁，执行的时间差值极小，并且写入foo.lock中的都是有效时间戳，所以对锁并没有影响。

为了让这个锁更加强壮，获取锁的客户端，应该在调用关键业务时，再次调用GET方法获取T1，和写入的T0时间戳进行对比，以免锁因其他情况被执行DEL意外解开而不知。****但是如果遇到上面描述得问题，则T0则会与T1不一致，当然差别一般会很小。这就是锁覆盖问题****。

锁覆盖会导致什么问题呢？

当客户端的锁过期时间被覆盖，会造成锁不具有标识性，会造成客户端无法释放锁（客户端只能释放明确自己持有的锁）。

nil 问题

GET返回nil时应该走哪种逻辑？

一、第一种走循环走setnx逻辑

1. C1客户端获取锁，并且处理完后，DEL掉锁。

2. 在DEL锁之前，C2通过SETNX向foo.lock设置时间戳T0失败，发现有客户端获取锁，进入GET操作。C2 向foo.lock发送GET命令，获取返回值T1(nil)(因为此时C1执行DEL删除锁)。

3. C2 循环，进入下一次SETNX逻辑。

二、第二种走超时逻辑

1. C1客户端获取锁，并且处理完后，DEL掉锁。

2. 在DEL锁之前，C2通过SETNX向foo.lock设置时间戳T0发现有客户端获取锁，进入GET操作。C2 向foo.lock发送GET命令，获取返回值T1(nil)(因为此时C1执行DEL删除锁)。

3. C2 通过 T0 > T1 + expire 对比，进入GETSET流程。

4. C2调用GETSET向foo.lock发送T0时间戳，返回foo.lock的原值T2，C2判断如果T2=T1相等，获得锁，如果T2!=T1，未获得锁。

分析

两种逻辑貌似都是OK，但是从逻辑处理上来说，当GET返回nil，表示锁是被删除的，而不是超时，应该走SETNX逻辑加锁。

对于"第二种走超时逻辑"是否会造成死锁，尚不清楚，不过推荐采用第一种方式。

GETSET返回nil时应该怎么处理？

前提：假设C4客户端获取锁后由于异常退出等原因未正常释放锁，导致锁超时。此时，C1、C2和C3客户端同时请求获取锁。C1、C2和C3客户端调用GET接口，C1返回T1，此时C3网络情况更好，快速进入获取锁，并执行DEL删除锁，C2返回T2(nil)。C1进入超时处理逻辑。C2面临上面提到「GET返回nil时应该走哪种逻辑？」的两种选择：1. 也进入超时处理逻辑；2. 继续循环走setnx逻辑（推荐）；

1. C1向foo.lock发送GETSET命令，获取返回值T11(nil)。C1比对C1和C11发现两者不同，处理逻辑认为未获取锁，然后继续循环走setnx逻辑。

2. C2有两种选择：

• 进入超时处理逻辑；
  C2 向foo.lock发送GETSET命令，获取返回值T22(C1写入的时间戳)。C2比对T2和T22发现两者不同，处理逻辑认为未获取锁，然后继续循环走setnx逻辑。

• 继续循环走setnx逻辑；

4. 很明显，C1和C2最终都会继续循环走setnx逻辑，然后通过SETNX向foo.lock设置时间戳T0会失败，这其实是因为在步骤1中C1执行GETSET命令导致的。此时C1和C2都认为未获取锁，其实C1是已经获取锁了，但是他的处理逻辑没有考虑GETSET返回nil的情况，只是单纯的用GET和GETSET值进行对比。

分析

至于为什么会出现这种情况？就如上面设想的场景那样，多客户端时，每个客户端连接redis后，发出的命令并不是连续的，导致从单客户端看到的好像连续的命令，到redis server后，这两条命令之间可能已经插入大量的其他客户端发出的命令，比如DEL,SETNX等。

正确的处理方式就是GETSET返回nil时，获取锁成功。

总结

1. 必要的超时机制：获取锁的客户端一旦崩溃，一定要有过期机制，否则其他客户端都降无法获取锁，造成死锁问题。

2. 分布式锁，多客户端的时间戳不能保证严格意义的一致性，所以在某些特定因素下，有可能存在问题。要适度的机制，可以承受小概率的事件产生。

3. 只对关键处理节点加锁，良好的习惯是，把相关的资源准备好，比如连接数据库后，调用加锁机制获取锁，直接进行操作，然后释放，尽量减少持有锁的时间。

4. 在持有锁期间要不要CHECK锁，如果需要严格依赖锁的状态，最好在关键步骤中做锁的CHECK检查机制，但是根据我们的测试发现，在大并发时，每一次CHECK锁操作，都要消耗掉几个毫秒，而我们的整个持锁处理逻辑才不到10毫秒，玩客没有选择做锁的检查。

5. sleep学问，为了减少对redis的压力，获取锁尝试时，循环之间一定要做sleep操作。但是sleep时间是多少是门学问。需要根据自己的redis的QPS，加上持锁处理时间等进行合理计算。如果redis的QPS足够高，也可以考虑循环之间不sleep，循环一定次数/时间执行yeild，提高响应速度。

6. 至于为什么不使用Redis的muti，expire，watch等机制，可以查下参考资料，找下原因。

代码实现

代码库

https://github.com/HuTu92/distributed-lock

源码

package com.github.hutu92.concurrent.locks;

import com.alibaba.fastjson.JSON;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

/**
 * Created by liuchunlong on 2018/8/31.
 * <p>
 * 基于redis的分布式锁 v1
 *
 * 需要客户端时间同步
 */
public class DistributedLock {

    private static final long RETRY_BARRIER = 3 * 1000; // 请求锁重试屏障，单位毫秒

    private final JedisPool jedisPool; // redis连接池
    private final String lockKey; // lock Key
    private final long lockExpiryInNanos; // 锁的过期时长，单位纳秒

    private static final ThreadLocal<Lock> lockThreadLocal = new ThreadLocal<Lock>();

    /**
     * 构造方法
     *
     * @param jedisPool          redis连接池
     * @param lockKey            锁的Key
     * @param lockExpiryInMillis 锁的过期时长，单位毫秒
     */
    public DistributedLock(JedisPool jedisPool, String lockKey, long lockExpiryInMillis) {
        this.jedisPool = jedisPool;
        this.lockKey = lockKey;
        this.lockExpiryInNanos = lockExpiryInMillis * 1000;
    }

    /**
     * 构造方法
     * <p>
     * 使用锁默认的过期时长Integer.MAX_VALUE，即锁永远不会过期
     *
     * @param jedisPool redis连接池
     * @param lockKey   锁的Key
     */
    public DistributedLock(JedisPool jedisPool, String lockKey) {
        this(jedisPool, lockKey, Integer.MAX_VALUE);
    }

    /**
     * 获取锁在redis中的Key标记
     *
     * @return locks key
     */
    public String getLockKey() {
        return this.lockKey;
    }

    /**
     * 锁的过期时长
     *
     * @return
     */
    public long getLockExpiryInNanos() {
        return lockExpiryInNanos;
    }

    /**
     * 请求分布式锁，不会阻塞，直接返回
     *
     * @param jedis redis 连接
     * @return 成功获取锁返回true, 否则返回false
     */
    private boolean tryAcquire(Jedis jedis) {

        final Lock newLock = new Lock(System.nanoTime() + this.lockExpiryInNanos);

        /**
         * 将新锁(newLock)写入redis中。如果成功写入，redis中不存在锁，获取锁成功；否则，redis中已存在锁，获取锁失败；
         */
        if (jedis.setnx(this.lockKey, newLock.toString()) == 1) {
            lockThreadLocal.set(newLock);
            return true;
        }

        /**
         * 至此，说明redis中已存在锁，获取锁失败，则需要进行如下操作:
         * 1. 判断redis中已存在的锁是否过期，如果过期则直接获取锁；
         * 2. 否则，获取锁失败；
         */

        final String currentLockValue = jedis.get(lockKey);
        // 特别的，当jedis.get()获取已存在的锁currentLockValue为空时，应该重新SETNX
        if (currentLockValue == null || currentLockValue.length() == 0) {
            tryAcquire(jedis);
        }
        final Lock currentLock = Lock.fromJson(currentLockValue); // redis中已存在的锁

        // 如果redis中已存在的锁已超时，则重新获取锁
        if (isExpired(currentLock)) {
            String originLockValue = jedis.getSet(lockKey, newLock.toString());

            /**
             * 这里还有个前置条件:
             *      会对已存在的锁进行校验，jedis.get()和jedis.getSet()获取的锁必须是同一锁，重新获取锁才成功
             */

            // 特别的，当jedis.getSet()获取已存在的锁originLockValue为空时，则认定获取锁成功
            if (originLockValue == null || originLockValue.length() == 0) {
                lockThreadLocal.set(newLock);
                return true;
            }

            if (originLockValue.equals(currentLockValue)) {
                lockThreadLocal.set(newLock);
                return true;
            }
        }

        return false;
    }

    /**
     * 请求分布式锁，不会阻塞，直接返回
     *
     * @return 成功获取锁返回true, 否则返回false
     */
    public boolean tryAcquire() {

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            return tryAcquire(jedis);
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }

    /**
     * 超时请求分布式锁，会阻塞
     *
     * 采用"自旋获取锁"的方式，直至获取锁成功或者请求锁超时
     *
     * @param acquireTimeoutInMillis 锁的请求超时时长
     * @return
     */
    public boolean acquire(long acquireTimeoutInMillis) {

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();

            long acquireTime = System.currentTimeMillis();

            // 锁的请求到期时间
            long expiryTime = System.currentTimeMillis() + acquireTimeoutInMillis;

            while (expiryTime >= System.currentTimeMillis()) {
                boolean result = tryAcquire(jedis);
                if (result) { // 获取锁成功直接返回，否则循环重试
                    return true;
                }

                if ((System.currentTimeMillis() - acquireTime) > RETRY_BARRIER) {
                    Thread.yield();
                }
            }

        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
        return false;
    }

    /**
     * 释放锁
     */
    public void release() {

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            release(jedis);
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }

    /**
     * 释放锁
     *
     * @param jedis
     */
    private void release(Jedis jedis) {
        Lock currlock = lockThreadLocal.get();
        if (currlock != null) {
            final String currentLockValue = jedis.get(lockKey);
            if (currentLockValue != null && currentLockValue.length() != 0) {
                final Lock currentLock = Lock.fromJson(currentLockValue); // redis中已存在的锁
                if (currlock.equals(currentLock)) {
                    lockThreadLocal.remove();
                    jedis.del(lockKey);
                }
            }
        }
    }

    /**
     * 判断当前线程是否持有锁
     *
     * 未持有锁或者锁超时，返回false
     *
     * @return
     */
    public boolean isLocked() {
        Lock currlock = lockThreadLocal.get();
        // 如果当前线程保存的lock不为null，并且未超时，则当前线程必然持有锁，锁未被意外释放
        return currlock != null && !currlock.isExpired();
    }

    /**
     * 判断指定的lock是否是当前线程持有的锁
     *
     * @return
     */
    boolean isMine(final Lock lock) {
        Lock currlock = lockThreadLocal.get();
        return currlock != null && currlock.equals(lock);
    }

    /**
     * 判断锁是否超时
     *
     * @param lock
     * @return
     */
    boolean isExpired(final Lock lock) {
        return lock.isExpired();
    }

    /**
     * 锁
     */
    protected static class Lock {

        private long expiryTime; // 锁的过期时间，注意，不是过期时长，单位纳秒

        Lock(long expiryTime) {
            this.expiryTime = expiryTime;
        }

        /**
         * 解析字符串，根据解析出的过期时间构造Lock
         *
         * @param json
         * @return
         */
        static Lock fromJson(String json) {
            return JSON.parseObject(json, Lock.class);
        }

        @Override
        public String toString() {
            return JSON.toJSONString(this, false);
        }

        public long getExpiryTime() {
            return expiryTime;
        }

        /**
         * 判断锁是否超时，如果锁的过期时间小于当前系统时间，则判定锁超时
         *
         * @return
         */
        boolean isExpired() {
            return this.expiryTime < System.nanoTime();
        }

        @Override
        public boolean equals(Object obj) {
            return obj != null
                    && obj instanceof Lock
                    && this.expiryTime == ((Lock) obj).getExpiryTime();
        }
    }
}

优化

上面存在的锁覆盖问题是不可避免的，还有就是要求客户端时间同步。下面我们进一步优化这一问题。

Redis命令介绍

SET

• 语法：

SET key value [EX seconds] [PX milliseconds] [NX|XX]

• 功能：
  将字符串值 value 关联到 key 。
  如果 key 已经持有其他值， SET 就覆写旧值，无视类型。
  对于某个原本带有生存时间（TTL）的键来说， 当 SET 命令成功在这个键上执行时，这个键原有的 TTL 将被清除。

• 可选参数
  从 Redis 2.6.12 版本开始，SET 命令的行为可以通过一系列参数来修改：

• EX second ：设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value 。

• PX millisecond ：设置键的过期时间为 millisecond 毫秒。 SET key value PX millisecond 效果等同于 PSETEX key millisecond value。

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

• XX ：只在键已经存在时，才对键进行设置操作。

因为 SET 命令可以通过参数来实现和 SETNX 、 SETEX 和 PSETEX 三个命令的效果，所以将来的 Redis 版本可能会废弃并最终移除 SETNX 、 SETEX 和 PSETEX 这三个命令。

• 返回值：
  在 Redis 2.6.12 版本以前， SET 命令总是返回 OK 。
  从 Redis 2.6.12 版本开始， SET 在设置操作成功完成时，才返回 OK 。
  如果设置了 NX 或者 XX ，但因为条件没达到而造成设置操作未执行，那么命令返回空批量回复（NULL Bulk Reply）。

使用模式

命令 SET resource-name anystring NX EX max-lock-time 是一种在 Redis 中实现锁的简单方法。

客户端执行以上的命令：

• 如果服务器返回 OK ，那么这个客户端获得锁。

• 如果服务器返回 NIL ，那么客户端获取锁失败，可以在稍后再重试。

设置的过期时间到达之后，锁将自动释放。

可以通过以下修改，让这个锁实现更健壮：

• 不使用固定的字符串作为键的值，而是设置一个不可猜测（non-guessable）的长随机字符串，作为口令串（token）。

• 不使用 DEL 命令来释放锁，而是发送一个 Lua 脚本，这个脚本只在客户端传入的值和键的口令串相匹配时，才对键进行删除。

这两个改动可以防止持有过期锁的客户端误删现有锁的情况出现。

以下是一个简单的解锁脚本示例：

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

源码

package com.github.hutu92;

import com.alibaba.fastjson.JSON;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Created by liuchunlong on 2018/9/4.
 * <p>
 * 基于redis的分布式锁 v2
 * <p>
 * 不需要客户端时间同步
 */
public class DistributedLock {

    private static final long RETRY_BARRIER = 600; // 重试屏障，单位毫秒
    private static final long INTERVAL_TIMES = 200; // 下一次重试等待，单位毫秒

    private final JedisPool jedisPool; // redis连接池
    private final String lockKey; // lock Key
    private final long lockExpiryInMillis; // 锁的过期时长，单位纳秒

    private final ThreadLocal<Lock> lockThreadLocal = new ThreadLocal<Lock>();

    /**
     * 构造方法
     *
     * @param jedisPool          redis连接池
     * @param lockKey            锁的Key
     * @param lockExpiryInMillis 锁的过期时长，单位毫秒
     */
    public DistributedLock(JedisPool jedisPool, String lockKey, long lockExpiryInMillis) {
        this.jedisPool = jedisPool;
        this.lockKey = lockKey;
        this.lockExpiryInMillis = lockExpiryInMillis;
    }

    /**
     * 构造方法
     * <p>
     * 使用锁默认的过期时长Integer.MAX_VALUE，即锁永远不会过期
     *
     * @param jedisPool redis连接池
     * @param lockKey   锁的Key
     */
    public DistributedLock(JedisPool jedisPool, String lockKey) {
        this(jedisPool, lockKey, Integer.MAX_VALUE);
    }

    /**
     * 获取锁在redis中的Key标记
     *
     * @return locks key
     */
    public String getLockKey() {
        return this.lockKey;
    }

    /**
     * 锁的过期时长
     *
     * @return
     */
    public long getLockExpiryInMillis() {
        return lockExpiryInMillis;
    }

    /**
     * can override
     *
     * @param jedis
     * @return
     */
    private String nextUid(Jedis jedis) {
        // 可以考虑雪花算法..
        return UUID.randomUUID().toString();
    }

    private synchronized Jedis getClient() {
        return jedisPool.getResource();
    }

    private synchronized void closeClient(Jedis jedis) {
        jedis.close();
    }

    /**
     * 请求分布式锁，不会阻塞，直接返回
     *
     * @param jedis redis 连接
     * @return 成功获取锁返回true, 否则返回false
     */
    private boolean tryAcquire(Jedis jedis) {

        final Lock nLock = new Lock(nextUid(jedis));
        String result = jedis.set(this.lockKey, nLock.toString(), "NX", "PX", this.lockExpiryInMillis);
        if ("OK".equals(result)) {
            lockThreadLocal.set(nLock);
            return true;
        }
        return false;
    }

    /**
     * 请求分布式锁，不会阻塞，直接返回
     *
     * @return 成功获取锁返回true, 否则返回false
     */
    public boolean tryAcquire() {

        Jedis jedis = null;
        try {
            jedis = getClient();
            return tryAcquire(jedis);
        } finally {
            if (jedis != null) {
                closeClient(jedis);
            }
        }
    }

    /**
     * 超时请求分布式锁，会阻塞
     *
     * 采用"自旋获取锁"的方式，直至获取锁成功或者请求锁超时
     *
     * @param acquireTimeoutInMillis 锁的请求超时时长
     * @return
     */
    public boolean acquire(long acquireTimeoutInMillis) throws InterruptedException {

        Jedis jedis = null;
        try {

            jedis = getClient();

            long acquireTime = System.currentTimeMillis();
            long expiryTime = System.currentTimeMillis() + acquireTimeoutInMillis; // 锁的请求到期时间

            while (expiryTime >= System.currentTimeMillis()) {
                boolean result = tryAcquire(jedis);
                if (result) { // 获取锁成功直接返回，否则循环重试
                    return true;
                }

                Thread.sleep(INTERVAL_TIMES);
            }

        } finally {
            if (jedis != null) {
                closeClient(jedis);
            }
        }
        return false;
    }

    /**
     * 释放锁
     *
     * @return
     */
    public boolean release() throws InterruptedException {
        return release(Integer.MAX_VALUE);
    }

    /**
     * 释放锁
     *
     * @return
     */
    public boolean release(long releaseTimeoutInMillis) throws InterruptedException {

        Jedis jedis = null;
        try {
            jedis = getClient();
            return release(jedis, releaseTimeoutInMillis);
        } finally {
            if (jedis != null) {
                closeClient(jedis);
            }
        }
    }

    /**
     * 释放锁
     *
     * @param jedis
     * @param releaseTimeoutInMillis
     * @return
     */
    private boolean release(Jedis jedis, long releaseTimeoutInMillis) throws InterruptedException {
        Lock cLock = lockThreadLocal.get();
        if (cLock == null) {
            System.out.println("lock is null!");
        }
        if (cLock != null) {
            String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";

            long releaseTime = System.currentTimeMillis();
            long expiryTime = System.currentTimeMillis() + releaseTimeoutInMillis; // 锁的释放到期时间

            while (expiryTime >= System.currentTimeMillis()) {
                Object result = jedis.eval(luaScript, Collections.singletonList(this.lockKey),
                        Collections.singletonList(cLock.toString()));
                if (((Long) result) == 1L) {
                    lockThreadLocal.remove();
                    return true;
                }

                Thread.sleep(INTERVAL_TIMES);
            }
        }
        return false;
    }


    /**
     * 锁
     */
    protected static class Lock {

        private String uid; // lock 唯一标识

        Lock(String uid) {
            this.uid = uid;
        }

        public String getUid() {
            return uid;
        }

        @Override
        public String toString() {
            return JSON.toJSONString(this, false);
        }
    }

}

性能调优

这里我们使用ab性能测试工具来模拟测试。

由于没有使用队列，对高并发请求进行削峰，所以所有的压力都会被打到redis上。为了测试方便我这里只是本地启动了单机redis，没有做其它的调优配置。

我们并发测试场景是1000个并发请求，总共2000个请求。

ab -n 2000 -c 1000 "localhost:8080/lock/v2/seckill"

上述的地址是一个接口，接口代码如下：

@RestController
@RequestMapping("/lock")
public class LockController {

    private static LongAdder longAdder = new LongAdder();
    private static Long ACQUIRE_TIMEOUT_IN_MILLIS = (long) Integer.MAX_VALUE;
    private static Long stock = 100000L;
    private static DistributedLock lock;

    static {
        longAdder.add(stock);
    }

    private final JedisPool jedisPool;

    @Autowired
    public LockController(JedisPool jedisPool) {
        this.jedisPool = jedisPool;
        lock = new DistributedLock(jedisPool, "seckillV2_" + UUID.randomUUID().toString());
    }

    @GetMapping("/v2/seckill")
    public String seckillV2() throws InterruptedException {

        boolean acquireResult = false;
        try {
            acquireResult = lock.acquire(ACQUIRE_TIMEOUT_IN_MILLIS);

            if (!acquireResult) {
                return "人太多了，换个姿势操作一下!";
            }

            if (longAdder.longValue() == 0L) {
                return "已抢光!";
            }

            doSomeThing(jedisPool);

            longAdder.decrement();

            System.out.println("已抢: " + (stock - longAdder.longValue()) + ", 还剩下: " + longAdder.longValue());

        } finally {
            if (acquireResult) {
                boolean releaseResult = lock.release();
                if (!releaseResult) {
                    System.out.println("释放锁失败！");
                }
            }
        }

        return "OK";
    }

    private void doSomeThing(JedisPool jedisPool) {
        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();

            jedis.incr("already_bought");
        } finally {
            if (jedis != null) {
                jedis.close();
            }
        }
    }
}

那么我们这里说的性能调优指的是什么呢？

仔细分析上面的源码你会发现，获取锁的逻辑是循环获取的，再每次循环之间，应该怎么去处理？如果不做任何处理，直接继续下一个循环，表面上看能够及时的获取锁，但这会给redis更大的压力，如果redis扛不住，到最后只会适得其反；而如果sleep等待，那么等待多久呢？等待久了，锁的获取和释放就会不及时；使用yield如何？等等

No1

if ((System.currentTimeMillis() - acquireTime) > RETRY_BARRIER) {
    Thread.yield();
}

请求获取锁的前600毫秒内直接循环重试，如果超过600毫秒还未获取到锁则每次循环都将线程推迟到下一个时间片执行。

image

主要参数说明：

• Failed requests：失败的请求

• Time per request：每个请求的平均耗时

No2

if ((System.currentTimeMillis() - acquireTime) > RETRY_BARRIER) {
    Thread.sleep(INTERVAL_TIMES);
} else {
    Thread.yield();
}

请求获取锁的前600毫秒内每次循环重试都先将线程推迟到下一个时间片，如果超过600毫秒还未获取到锁则每次循环都将线程休眠200毫秒。

image

<figcaption style="margin: 10px 0px 0px; padding: 0px; max-width: 100%; box-sizing: border-box !important; word-wrap: break-word !important; line-height: inherit; text-align: center; color: rgb(153, 153, 153); font-size: 0.7em;">carbon -1-</figcaption>

</figure>

很明显，出错率降低了很多，每个请求的耗时也减少了一半，这是因为，No1中在600毫秒内的直接循环重试，会产生很多意义的请求，给redis造成了巨大的压力，无法响应请求。

No3

Thread.sleep(INTERVAL_TIMES);

请求获取锁的每次循环重试都将线程休眠200毫秒。

image

<figcaption style="margin: 10px 0px 0px; padding: 0px; max-width: 100%; box-sizing: border-box !important; word-wrap: break-word !important; line-height: inherit; text-align: center; color: rgb(153, 153, 153); font-size: 0.7em;">carbon -2-</figcaption>

</figure>

No4

Thread.sleep(INTERVAL_TIMES * 10);

请求获取锁的每次循环重试都将线程休眠2秒。

image

<figcaption style="margin: 10px 0px 0px; padding: 0px; max-width: 100%; box-sizing: border-box !important; word-wrap: break-word !important; line-height: inherit; text-align: center; color: rgb(153, 153, 153); font-size: 0.7em;">carbon -3-</figcaption>

</figure>

很明显，休眠时间过长，会使部分线程请求锁的时间变长，不能够及时获取到锁。

No5

Thread.yield();

请求获取锁的每次循环重试都将线程推迟到下一个时间片执行。

image

<figcaption style="margin: 10px 0px 0px; padding: 0px; max-width: 100%; box-sizing: border-box !important; word-wrap: break-word !important; line-height: inherit; text-align: center; color: rgb(153, 153, 153); font-size: 0.7em;">carbon -4-</figcaption>

</figure>

总结

总的来说，No2与No3表现的都还可以。但是No2使用了Thread.yield();也会给redis造成压力，我可以对比下两者的 Percentage of the requests served within a certain time (ms) 数据。可以看到No3的90%以下请求的用户平均时间要明显低于No2的。所以最终我们选择No3策略。

当然你也可以根据你的redis的QPS自行调整策略。