在分布式系统中，实现分布式锁是一项常见的需求。为了追求性能，通常使用Redis使用分布式锁，但是想要实现高性能并且数据安全的分布式锁，并非易事，先看一下分布式锁要满足哪些特性。

分布式锁需要满足以下特性：

互斥，一个线程获取到锁，其他线程只能等待。
高性能，比如MySQL基于磁盘操作实现，性能较差。
加锁操作是原子的
释放锁操作是原子的
避免释放锁失败
避免提前释放锁，比如释放锁操作加在事务里面，就会出现事务提交前，已经释放锁。
加锁操作支持阻塞，避免其他线程不断轮询，浪费CPU。
支持设置锁过期时间，防止释放锁失败，作为兜底策略。
支持断开客户端连接后，自动释放锁（例如zookeeper临时节点）。
释放锁的时候判断是否属于当前线程，避免释放了其他线程的锁。
支持锁可重入，避免当前线程多次加锁的时候，出现死锁。
支持锁自动续期，如果已经给锁设置了过期时间，可能会出现业务还没执行完成，锁已经过期。

使用Redis实现分布式锁有以下10种方式，每种方式都有各自的优缺点，一起分析一下。

使用 setnx 命令

可以使用 Redis 提供的 setnx 命令实现分布式锁，setnx 命令的作用是：

如果键 key 不存在，则将键 key 的值设置为 value，同时返回 1；
如果键 key 已经存在，则不做任何操作，返回 0。

伪代码实现如下：

// 判断加锁是否成功
if (jedis.setnx(lock_key, lock_value) == 1) {
    // 执行业务代码
    doBusiness();
    // 释放锁
    jedis.del(lock_key);
}

上面代码中有个问题，如果执行业务出现异常了，岂不是永远无法释放锁了，所以业务代码要用 try/finally 包裹起来。

防止释放锁失败

// 判断加锁是否成功
if (jedis.setnx(lock_key, lock_value) == 1) {
    try {
        // 执行业务代码
        doBusiness();
    } finally {
        // 释放锁
        jedis.del(lock_key);
    }
}

这样还是有问题的，如果执行业务的时候服务器宕机了怎么办？还是无法释放锁，所以要给锁加上过期时间，到期后可以自动释放锁。

加过期时间

// 判断加锁是否成功
if (jedis.setnx(lock_key, lock_value) == 1) {
    try {
        // 设置过期时间
        jedis.expire(lock_key, timeout);
        // 执行业务代码
        doBusiness();
    } finally {
        // 释放锁
        jedis.del(lock_key);
    }
}

由于加锁和设置过期时间，两个操作不是原子的。可能出现刚加锁成功，还没来得及设置过期时间，服务器就宕机了，造成永远无法释放锁。 Redis 2.6.12 版本提供的 setnx 命令，同时可以设置过期时间，是原子操作。 SET key value [EX seconds] [PX milliseconds] [NX|XX]

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 ：只在键已经存在时，才对键进行设置操作。

原子加锁

// 设置 nx 和 ex 参数
SetParams setParams = new SetParams();
setParams.nx();
setParams.ex(timeout);

// 判断加锁是否成功
if (jedis.set(lock_key, lock_value, setParams) == 1) {
    try {
        // 执行业务代码
        doBusiness();
    } finally {
        // 释放锁
        jedis.del(lock_key);
    }
}

释放锁流程有问题，释放锁的时候，没有判断是否是当前线程的 lock_value，可能会释放了其他线程的锁。例如下面的情况：

A线程加锁成功
A线程执行业务代码耗时过长，锁过期，自动释放锁。
B线程抢占锁
A线程执行结束，释放锁（由于 lock_key 相同，没有判断 lock_value，释放了B线程的锁）

释放了其他线程的锁

所以释放锁的时候，要判断 lock_value 是否属于当前线程。

// 设置 nx 和 ex 参数
SetParams setParams = new SetParams();
setParams.nx();
setParams.ex(timeout);
// 判断加锁是否成功
if (jedis.set(lock_key, lock_value, setParams) == 1) {
    try {
        // 设置过期时间
        jedis.set()
        jedis.expire(lock_key, timeout);
        // 执行业务代码
        doBusiness();
    } finally {
        // 释放锁，判断 lock_value 是否属于当前线程
        if ("lock_value".equals(jedis.get(lock_key))) {
            jedis.del(lock_key);
        }
    }
}

又有个很明显的问题，释放锁流程不是原子操作，可能存在刚判断是否相等，已经被其他线程抢占锁了，导致又释放了其他线程的锁，可以使用 Lua 脚本实现原子操作。

释放锁原子操作

final String releaseLockScript = "if redis.call('get', KEYS[1]) == ARGV[1] then\n" +
        "   return redis.call('del', KEYS[1])\n" +
        "else\n" +
        "   return 0\n" +
        "end";

// 设置 nx 和 ex 参数
SetParams setParams = new SetParams();
setParams.nx();
setParams.ex(timeout);
// 判断加锁是否成功
if (jedis.set(lock_key, lock_value, setParams) == 1) {
    try {
        // 设置过期时间
        jedis.set()
        jedis.expire(lock_key, timeout);
        // 执行业务代码
        doBusiness();
    } finally {
        // 释放锁，原子操作
        jedis.eval(releaseLockScript, Collections.singletonList(lock_key), Collections.singletonList(lock_value))
    }
}

上面的操作流程也不是完美的，刚才分析过，如果执行业务代码耗时过长，锁过期了，就会出现其他线程抢占锁，所以就需要锁自动续期。

锁自动续期

可以使用下面方式实现锁自动续期：

加锁成功后，启动一个定时任务，每个一段时间检测是否执行完成业务代码，依据是锁是否还存在。
这个定时任务每隔三分之一时间检查一次，比如锁过期时间是30秒，就每隔10秒检查一次。
如果锁还存在，就把锁过期时间继续设置成30秒。

Redisson框架已经实现了这个流程，名叫看门狗机制（Watch Dog），底层使用 Lua 脚本实现。

// 获取 Redisson 锁
RLock lock = redissonClient.getLock(lock_key);
try {
    // 加锁，并设置3秒后过期
    lock.lock(3, TimeUnit.SECONDS);
    // 执行业务代码
    doBusiness();
} catch (Exception e) {
    System.out.println("加锁超时");
} finally {
    // 释放锁
    lock.unlock();
}

事务提交前，释放锁

@Transactional
public void updateDB(String lockKey) {
    // 加锁
    boolean lockFlag = redisLock.lock(lockKey);
    // 判断是否加锁成功
    if (lockFlag) {
        // 执行业务代码
        doBusiness();
        // 释放锁
        redisLock.unlock(lockKey);
    }
}

如果使用注解事务，并把加锁和释放锁的逻辑放在事务里面，就可能会出现事务提交前，锁已经释放，导致互斥锁失效。正确的做法是，在开启事务前加锁，在提交事务后释放锁。

public void lock(String lockKey) {
    // 加锁
    boolean lockFlag = redisLock.lock(lockKey);
    // 判断是否加锁成功
    if (lockFlag) {
        // 事务方法单独执行
        updateDB(lockKey);
        // 释放锁
        redisLock.unlock(lockKey);
    }
}

@Transactional
public void updateDB(String lockKey) {
    // 执行业务代码
    doBusiness();
}

注意不要调用本类的事务方法，会导致事务失效，上面为了演示加锁、释放锁要跟事务分开。

可重入锁

可重入锁是允许同一个线程多次获取同一把锁。如果锁不支持可重入的话，在递归调用的场景下，会出现死锁情况。实现可重入锁，也比较简单，只需要增加一个计数器，记录当前线程加锁次数。每加一次锁，计数器加一，释放锁，计数器减一，计数器为零的时候，释放锁。

集群加锁

在对Redis集群加锁的时候，由于故障转移的原因可能会导致互斥锁失效。比如下面的情况：

线程A对集群master节点加锁成功
锁数据还没有来得及同步到slave节点，master节点崩溃了。
集群自动故障转移，slave节点被选举成新的master节点，对外提供服务。
线程B对集群master节点加锁成功，互斥锁失效。

Redlock（Redis Distributed Lock，Redis分布式锁，简称红锁）实现对集群加锁的功能。实现原理如下：

使用多个master节点组成Redis集群（假设有5个master节点）
客户端加锁的时候，按照顺序对5个master节点请求加锁，要求所有请求锁超时时间之和要小于锁过期时间。假设锁过期时间是5秒，则对每个master节点请求锁超时时间最长为一秒，超过一秒还没有加锁成功，就放弃加锁，继续请求下一个master节点。
如果对超过半数的master节点加锁成功，视为获取锁成功。这里是需要对3个master节点加锁成功（N/2+1）。如果没有获取锁成功，则需要释放已经加锁的master节点，避免影响其他客户端获取锁。
锁的实际过期时间等于锁的初始过期时间减去获取锁的时间，比如锁初始过期时间是5秒，获取锁用了2秒，锁实际过期时间是3秒。

四时宝库

程序员的知识宝库

高并发必备，使用Redis分布式锁必须注意的10个细节