常用分布式锁实现方式总结

大多数互联网系统都是分布式部署的，分布式部署确实能带来性能和效率上的提升，但为此，我们就需要多解决一个分布式环境下，数据一致性的问题。

当某个资源在多系统之间，具有共享性的时候，为了保证大家访问这个资源数据是一致的，那么就必须要求在同一时刻只能被一个客户端处理，不能并发的执行，否者就会出现同一时刻有人写有人读，大家访问到的数据就不一致了。

一、我们为什么需要分布式锁？

在单机时代，虽然不需要分布式锁，但也面临过类似的问题，只不过在单机的情况下，如果有多个线程要同时访问某个共享资源的时候，我们可以采用线程间加锁的机制，即当某个线程获取到这个资源后，就立即对这个资源进行加锁，当使用完资源之后，再解锁，其它线程就可以接着使用了。例如，在JAVA中，甚至专门提供了一些处理锁机制的一些API（synchronize/Lock等）。

但是到了分布式系统的时代，这种线程之间的锁机制，就没作用了，系统可能会有多份并且部署在不同的机器上，这些资源已经不是在线程之间共享了，而是属于进程之间共享的资源。

因此，为了解决这个问题，我们就必须引入「分布式锁」。

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

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

• 排他性：在同一时间只会有一个客户端能获取到锁，其它客户端无法同时获取
• 避免死锁：这把锁在一段有限的时间之后，一定会被释放（正常释放或异常释放）
• 高可用：获取或释放锁的机制必须高可用且性能佳

讲完了背景和理论，那我们接下来再看一下分布式锁的具体分类和实际运用。

二、分布式锁的实现方式有哪些？

目前主流的有三种，从实现的复杂度上来看，从上往下难度依次增加：

• 基于数据库实现
• 基于Redis实现
• 基于ZooKeeper实现

无论哪种方式，其实都不完美，依旧要根据咱们业务的实际场景来选择。

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

   • 基于数据库的乐观锁
   • 基于数据库的悲观锁

我们先来看一下如何基于「乐观锁」来实现：

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

下面找图举例，

（图片来源网络）

如图，假设同一个账户，用户A和用户B都要去进行取款操作，账户的原始余额是2000，用户A要去取1500，用户B要去取1000，如果没有锁机制的话，在并发的情况下，可能会出现余额同时被扣1500和1000，导致最终余额的不正确甚至是负数。但如果这里用到乐观锁机制，当两个用户去数据库中读取余额的时候，除了读取到2000余额以外，还读取了当前的版本号version=1，等用户A或用户B去修改数据库余额的时候，无论谁先操作，都会将版本号加1，即version=2，那么另外一个用户去更新的时候就发现版本号不对，已经变成2了，不是当初读出来时候的1，那么本次更新失败，就得重新去读取最新的数据库余额。

通过上面这个例子可以看出来，使用「乐观锁」机制，必须得满足：
（1）锁服务要有递增的版本号version
（2）每次更新数据的时候都必须先判断版本号对不对，然后再写入新的版本号

伪代码：

UPDATE
    table_name
SET
    otherColumn=#{otherColumn_value},
    version = version+1
WHERE
    otherColumn= #{otherColumn_value} and version = #{version}

我们再来看一下如何基于「悲观锁」来实现：

悲观锁也叫作排它锁，在Mysql中是基于 for update 来实现加锁的，例如：

//锁定的方法-伪代码
 public boolean lock() {
        connection.setAutoCommit(false)
        while (true) {
            {
                result = select * from methodLock where method_name = xxx for update;
                if (result == null) {
                    //结果不为空，
                    //则说明获取到了锁
                    return true;
                }
            } catch (Exception e) {
            }
            //没有获取到锁，继续获取
            sleep(1000);
        }
        return false;
    }

//释放锁-伪代码
public void unlock(){
    connection.commit();
}

上面的示例中，user表中，id是主键，通过 for update 操作，数据库在查询的时候就会给这条记录加上排它锁。
（需要注意的是，在InnoDB中只有字段加了索引的，才会是行级锁，否者是表级锁，所以这个id字段要加索引）

当这条记录加上排它锁之后，其它线程是无法操作这条记录的。

那么，这样的话，我们就可以认为获得了排它锁的这个线程是拥有了分布式锁，然后就可以执行我们想要做的业务逻辑，当逻辑完成之后，再调用上述释放锁的语句即可。

乐观锁是在应用层加锁，而悲观锁是在数据库层加锁(for update)

乐观锁顾名思义就是在操作时很乐观，这数据只有我在用，我先尽管用，最后发现不行时就回滚。

悲观锁在操作时很悲观，生怕数据被其他人更新掉，我就先将其先锁住，让别人用不了，我操作完成后再释放掉。

悲观锁需要数据库级别上的的实现，程序中是做不到的，如果在长事务环境中，如果并发量大，数据库连接都被锁占用，其它读写业务都崩溃了，数据会一直被锁住，导致并发性能大大地降低。

一般来说如果并发量很高的话，建议使用悲观锁，否则的话就使用乐观锁。

如果并发量很高时使用乐观锁的话，会导致很多的并发事务回滚、操作失败。

总之，冲突几率大用悲观，小就用乐观。

1. 基于Redis实现

基于Redis实现的锁机制，主要是依赖redis自身的原子操作，例如：

SET user_key user_value NX PX 100

尽量不要去使用setnx，该命令没有设置过期参数，只有设置key和value，还需要去判断加锁是否成功，之后去设置过期时间，不能保证原子性，如果去设置过期时间的时候发生错误，这时候key就是-1，永不失效，这将会发生死锁。

Long result = jedis.setnx(lockKey, identifier);
if (result == 1) {
    // 若在这里程序突然崩溃，则无法设置过期时间，将发生死锁
    jedis.expire(lockKey, lockExpire);
}

redis从2.6.12版本开始，SET命令才支持这些参数：
NX：只有在键不存在时，才对键进行设置操作，SET key value NX 效果等同于 SETNX key value ，XX仅当键已经存在时才设置它
PX millisecond：设置键的过期时间为millisecond毫秒，当超过这个时间后，设置的键会自动失效，EX单位为秒

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

为什么这个命令可以帮我们实现锁机制呢？
因为这个命令是只有在某个key不存在的时候，才会执行成功。那么当多个进程同时并发的去设置同一个key的时候，就永远只会有一个进程成功。
当某个进程设置成功之后，就可以去执行业务逻辑了，等业务逻辑执行完毕之后，再去进行解锁。

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

另外，针对redis集群模式的分布式锁，可以采用redis的Redlock机制。

单Redis实例实现分布式锁：

依赖：

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.9.0</version>
</dependency>

代码：

package com.redis;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Transaction;
import redis.clients.jedis.exceptions.JedisException;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.UUID;

public class RedisUtils {
    //定义连接池独享
    private static JedisPool jedisPool;
    private static String HOST = "192.168.184.133";
    private static int PORT = 6379;
    //连接超时时间单位毫秒数
    private static int TIMEOUT = 6*1000;
    private static String PASSWORD = "123456";
    //连接耗尽时是否阻塞，false报异常，true阻塞直到超时（默认）
    private static boolean isBlock = true;

    static {
        // 使用连接池
        // 【2】创建JedisPool所需的连接池配置
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        //最大连接数，默认8
        //最大的Jedis实例数（连接池是Jedis实例，默认是8）
        poolConfig.setMaxTotal(1024);
        // 最大空闲数,默认8
        //最大的空闲实例数，设置这个可以保留足够的连接，快速获取到Jedis对象
        poolConfig.setMaxIdle(200);
        // 最小空闲连接数，默认0
        poolConfig.setMinIdle(0);
        // 对象最小空闲时间，默认1800000毫秒(30分钟)
        poolConfig.setMinEvictableIdleTimeMillis(1800000);
        // 获取连接的最大等待毫秒数。如果设为小于0，则永远等待
        poolConfig.setMaxWaitMillis(-1);
        // 在创建对象时检测对象是否有效，true是，默认值是false
        poolConfig.setTestOnCreate(true);
        //从对象池获取对象时检测对象是否有效，默认false
        //提前检查Jedis对象，为true获取的Jedis一定是可用的
        poolConfig.setTestOnBorrow(true);
        //在向对象池中归还对象时是否检测对象有效，true是，默认值是false
        poolConfig.setTestOnReturn(false);
        //在检测空闲对象线程检测到对象不需要移除时，是否检测对象的有效性。true是，默认值是false
        poolConfig.setTestWhileIdle(false);
        //检测空闲对象线程每次检测的空闲对象的数量。默认值是3；如果这个值小于0,则每次检测的空闲对象数量等于当前空闲对象数量除以这个值的绝对值，并对结果向上取整
        poolConfig.setNumTestsPerEvictionRun(3);
        //是否启用后进先出, 默认true
        poolConfig.setLifo(true);
        //多长时候执行一次空闲对象检测。单位是毫秒数。如果小于等于0，则不执行检测线程。默认值是-1
        poolConfig.setTimeBetweenEvictionRunsMillis(-1);
        //当对象池没有空闲对象时，新的获取对象的请求是否阻塞。true阻塞。默认值是true;
        poolConfig.setBlockWhenExhausted(isBlock);
        //是否启用pool的jmx管理功能, 默认true
        poolConfig.setJmxEnabled(true);
        //【1】创建JedisPool连接池
        jedisPool = new JedisPool(poolConfig, HOST, PORT, TIMEOUT, PASSWORD);
    }

    /**
     * 获取jedis连接
     */
    public static Jedis getJedis() {
        if (jedisPool != null) {
            //获取Jedis对象
            Jedis jedis = jedisPool.getResource();
            return jedis;
        }
        return null;
    }

    /**
     * 释放jedis资源
     */
    public static void releaseResource(Jedis jedis) {
        if (jedis != null) {
            jedisPool.returnResource(jedis);
        }
    }

    /**
     * 加锁
     * 因为 SetNX 不具备设置过期时间的功能，所以我们需要借助 Expire 来设置，
     * 同时我们需要把两者用 Multi/Exec 包裹起来以确保请求的原子性，以免 SetNX 成功了 Expire 却失败了。
     * 可惜还有问题：当多个请求到达时，虽然只有一个请求的 SetNX 可以成功，但是任何一个请求的 Expire 却都可以成功，
     * 如此就意味着即便获取不到锁，也可以刷新过期时间，如果请求比较密集的话，那么过期时间会一直被刷新，导致锁一直有效。
     * 于是乎我们需要在保证原子性的同时，有条件的执行 Expire,从 2.6.12 起，SET 涵盖了 SETEX 的功能，
     * 并且 SET 本身已经包含了设置过期时间的功能，也就是说，我们前面需要的功能只用 SET 就可以实现。代码入下：
     * @param lockName  锁的key
     * @param acquireTimeout  获取超时时间
     * @param timeout   锁的超时时间（毫秒）
     * @return 锁标识
     */
    public static String lockWithTimeout(String lockName,long acquireTimeout,long timeout) {
        Jedis jedis = null;
        String retIdentifier = null;
        try {
            // 获取连接
            jedis = getJedis();
            // 随机生成一个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) {
                //NX|XX, NX -- Only set the key if it does not already exist. XX -- Only set the key if it already exist.
                //EX|PX, expire time units: EX = seconds; PX = milliseconds
                String result=jedis.set(lockKey, identifier,"NX", "EX",lockExpire);
                if (jedis !=null &&  result != null && result.equalsIgnoreCase("OK")) {
                    // 返回value值，用于释放锁时确认
                    retIdentifier = identifier;
                    return retIdentifier;
                }
                //休眠一下继续获取锁
                try {
                    Thread.sleep(10);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            releaseResource(jedis);
        }
        return retIdentifier;
    }

    /**
     * 用Lua 脚本来实现释放锁
     * @param lockName 锁的key
     * @param identifier    释放锁的标识
     * @return
     */
    public static boolean releaseLock1(String lockName, String identifier) {
        Jedis jedis = null;
        String lockKey = "lock:" + lockName;
        boolean retFlag = false;
        try {
            jedis = getJedis();
            String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
            Object obj=jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(identifier));
            int result = Integer.parseInt(obj==null?"":obj.toString());
            if (Objects.equals(1, result)) {
                retFlag = true;
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
            releaseResource(jedis);
        }
        return retFlag;
    }

    /**
     * 用事务来实现释放锁
     * @param lockName 锁的key
     * @param identifier    释放锁的标识
     * @return
     */
    public static boolean releaseLock(String lockName, String identifier) {
        Jedis jedis = null;
        String lockKey = "lock:" + lockName;
        boolean retFlag = false;
        try {
            jedis = getJedis();
            while (true) {
                // 监视lock，准备开始事务，如果key被修改则放弃删除
                jedis.watch(lockKey);
                // 通过前面返回的value值判断是不是该锁，若是该锁，则删除，释放锁
                //identifier.equals(jedis.get(lockKey)),如果拿锁超时，这里会报空指针异常，因为上面拿锁的时候超时，返回identifier为null,所以改为下面这样
                if (jedis.get(lockKey).equals(identifier)) {
                    Transaction transaction = jedis.multi();
                    transaction.del(lockKey);
                    List<Object> results = transaction.exec();
                    if (results.size()==0) {
                        continue;
                    }
                    retFlag = true;
                }
                jedis.unwatch();
                break;
            }
        } catch (JedisException e) {
            e.printStackTrace();
        } finally {
          releaseResource(jedis);
        }
        return retFlag;
    }

}

从表面上看，似乎效果还不错，但是这里有一个问题：这个架构中存在一个严重的单点失败问题。如果Redis挂了怎么办？你可能会说，可以通过增加一个slave节点解决这个问题。但这通常是行不通的。这样做，我们不能实现资源的独享,因为Redis的主从同步通常是异步的。

在这种场景（主从结构）中存在明显的竞态:

1. 客户端A从master获取到锁
2. 在master将锁同步到slave之前，master宕掉了。
3. slave节点被晋级为master节点
4. 客户端B取得了同一个资源被客户端A已经获取到的另外一个锁。安全失效！

主从异步复制几乎可以看做是同步的，从操作主节点到返回响应，异步复制就完成了，从节点也有锁了，但是有时候程序就是这么巧，比如说正好一个节点挂掉的时候，多个客户端同时取到了锁。如果你可以接受这种小概率错误，那用这个基于复制的方案就完全没有问题。

1. 基于ZooKeeper实现

其实基于ZooKeeper，就是使用它的临时有序节点来实现的分布式锁。

原理就是：当某客户端要进行逻辑的加锁时，就在zookeeper上的某个指定节点的目录下，去生成一个唯一的临时有序节点， 然后判断自己是否是这些有序节点中序号最小的一个，如果是，则算是获取了锁。如果不是，则说明没有获取到锁，那么就需要在序列中找到比自己小的那个节点，并对其调用exist()方法，对其注册事件监听，当监听到这个节点被删除了，那就再去判断一次自己当初创建的节点是否变成了序列中最小的。如果是，则获取锁，如果不是，则重复上述步骤。

当释放锁的时候，只需将这个临时节点删除即可。

（图片来自网络）

如图，locker是一个持久节点，node_1/node_2/…/node_n 就是上面说的临时节点，由客户端client去创建的。
client_1/client_2/…/clien_n 都是想去获取锁的客户端。以client_1为例，它想去获取分布式锁，则需要跑到locker下面去创建临时节点（假如是node_1）创建完毕后，看一下自己的节点序号是否是locker下面最小的，如果是，则获取了锁。如果不是，则去找到比自己小的那个节点（假如是node_2），找到后，就监听node_2，直到node_2被删除，那么就开始再次判断自己的node_1是不是序列中最小的，如果是，则获取锁，如果还不是，则继续找一下一个节点。

依赖：

<dependency>
    <groupId>com.101tec</groupId>
    <artifactId>zkclient</artifactId>
    <version>0.10</version>
</dependency>

实现：

package com.redis.zklock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

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;

public class DistributedLock implements Lock, Watcher {
    private ZooKeeper zk = null;
    // 根节点
    private String ROOT_LOCK = "/locks";
    // 竞争的资源
    private String lockName;
    // 等待的前一个锁
    private String WAIT_LOCK;
    // 当前锁
    private String CURRENT_LOCK;
    // 计数器
    private CountDownLatch countDownLatch;
    private int sessionTimeout = 30000;
    private List<Exception> exceptionList = new ArrayList<Exception>();

    /**
     * 配置分布式锁
     * @param config 连接的url
     * @param lockName 竞争资源
     */
    public DistributedLock(String config, String lockName) {
        this.lockName = lockName;
        try {
            // 连接zookeeper
            zk = new ZooKeeper(config, sessionTimeout, this);
            Stat stat = zk.exists(ROOT_LOCK, false);
            if (stat == null) {
                // 如果根节点不存在，则创建根节点
                zk.create(ROOT_LOCK, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    // 节点监视器
    public void process(WatchedEvent event) {
        if (this.countDownLatch != null) {
            this.countDownLatch.countDown();
        }
    }

    public void lock() {
        if (exceptionList.size() > 0) {
            throw new LockException(exceptionList.get(0));
        }
        try {
            if (this.tryLock()) {
                System.out.println(Thread.currentThread().getName() + " " + lockName + "获得了锁");
                return;
            } else {
                // 等待锁
                waitForLock(WAIT_LOCK, sessionTimeout);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    public boolean tryLock() {
        try {
            String splitStr = "_lock_";
            if (lockName.contains(splitStr)) {
                throw new LockException("锁名有误");
            }
            // 创建临时有序节点
            CURRENT_LOCK = zk.create(ROOT_LOCK + "/" + lockName + splitStr, new byte[0],
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println(CURRENT_LOCK + " 已经创建");
            // 取所有子节点
            List<String> subNodes = zk.getChildren(ROOT_LOCK, false);
            // 取出所有lockName的锁
            List<String> lockObjects = new ArrayList<String>();
            for (String node : subNodes) {
                String _node = node.split(splitStr)[0];
                if (_node.equals(lockName)) {
                    lockObjects.add(node);
                }
            }
            Collections.sort(lockObjects);
            System.out.println(Thread.currentThread().getName() + " 的锁是 " + CURRENT_LOCK);
            // 若当前节点为最小节点，则获取锁成功
            if (CURRENT_LOCK.equals(ROOT_LOCK + "/" + lockObjects.get(0))) {
                return true;
            }

            // 若不是最小节点，则找到自己的前一个节点
            String prevNode = CURRENT_LOCK.substring(CURRENT_LOCK.lastIndexOf("/") + 1);
            WAIT_LOCK = lockObjects.get(Collections.binarySearch(lockObjects, prevNode) - 1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
        return false;
    }

    public boolean tryLock(long timeout, TimeUnit unit) {
        try {
            if (this.tryLock()) {
                return true;
            }
            return waitForLock(WAIT_LOCK, timeout);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    // 等待锁
    private boolean waitForLock(String prev, long waitTime) throws KeeperException, InterruptedException {
        Stat stat = zk.exists(ROOT_LOCK + "/" + prev, true);

        if (stat != null) {
            System.out.println(Thread.currentThread().getName() + "等待锁 " + ROOT_LOCK + "/" + prev);
            this.countDownLatch = new CountDownLatch(1);
            // 计数等待，若等到前一个节点消失，则precess中进行countDown，停止等待，获取锁
            this.countDownLatch.await(waitTime, TimeUnit.MILLISECONDS);
            this.countDownLatch = null;
            System.out.println(Thread.currentThread().getName() + " 等到了锁");
        }
        return true;
    }

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

    public Condition newCondition() {
        return null;
    }

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

    public class LockException extends RuntimeException {
        private static final long serialVersionUID = 1L;
        public LockException(String e){
            super(e);
        }
        public LockException(Exception e){
            super(e);
        }
    }
}

测试：

package com.redis.zklock;

public class Test {
    static int n = 500;

    public static void secskill() {
        System.out.println(--n);
    }

    public static void main(String[] args) {

        Runnable runnable = new Runnable() {
            public void run() {
                DistributedLock lock = null;
                try {
                    lock = new DistributedLock("192.168.184.133:2181,192.168.184.133:2182,192.168.184.133:2183", "test1");
                    lock.lock();
                    secskill();
                    System.out.println(Thread.currentThread().getName() + "正在运行");
                } finally {
                    if (lock != null) {
                        lock.unlock();
                    }
                }
            }
        };

        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(runnable);
            t.start();
        }
    }
}

上面几种方式，哪种方式都无法做到完美。就像CAP一样，在复杂性、可靠性、性能等方面无法同时满足，所以，根据不同的应用场景选择最适合自己的才是王道。

以上，就讲完了为什么我们需要分布式锁这个技术，以及分布式锁中常见的三种机制，欢迎大家一起交流。