分布式锁原理及实现

约定不等于承诺〃 2023-01-19 04:56 21阅读 0赞

## **什么是分布式锁？　**　 ##

控制分布式架构中多个模块访问的优先级

要介绍分布式锁，首先要提到与分布式锁相对应的是线程锁、进程锁。

线程锁：主要用来给方法、代码块加锁。当某个方法或代码使用锁，在同一时刻仅有一个线程执行该方法或该代码段。线程锁只在同一JVM中有效果，因为线程锁的实现在根本上是依靠线程之间共享内存实现的，比如synchronized是共享对象头，显示锁Lock是共享某个变量（state）。

进程锁：为了控制同一操作系统中多个进程访问某个共享资源，因为进程具有独立性，各个进程无法访问其他进程的资源，因此无法通过synchronized等线程锁实现进程锁。

分布式锁：当多个进程不在同一个系统中，用分布式锁控制多个进程对资源的访问。

--------------------

**什么情况下需要使用分布式锁**

比如：

高并发下争夺共享资源，比如秒杀对于库存这种共享资源需要用到分布式锁，如果不用分布式锁很可能造成超卖。

# 分布式锁也是锁 #

*  在单体应用的时候，如果多个线程要访问共享资源的时候，我们通常线程间加锁的机制，在某一个时刻，只有一个线程可以对这个资源进行操作，其他线程需要等待锁的释放，Java中也有一些处理锁的机制，比如synchronized。
 *  而到了分布式的环境中，当某个资源可以被多个系统访问使用到的时候，为了保证大家访问这个数据是一致性的，那么就要求再同一个时刻，只能被一个系统使用，这时候线程之间的锁机制就无法起到作用了，因为分布式环境中，系统是会部署到不同的机器上面的，那么就需要【分布式锁】了。

# 什么时候需要使用分布式锁 #

总结来看，当有多个客户端需要访问并操作同一个资源，还需要保持这个资源一致性的时候，就需要使用【分布式锁】，让多客户端互斥的对共享资源进行访问。

举个例子来说明一下：

*  有多个批处理任务，两台机器同时处理，如果不加任何控制的话，很有可能同一个批处理被两台机器分别处理一遍；如果使用分布式锁，在领取任务的时候，一个任务只会被一台机器领到，这样就不会造成任务的重复执行；
 *  再多思考一些，如果A/B两台机器，任务1被A机器领取到进行处理，在处理到一半的时候，A机器挂掉了，那么这个批处理任务也就无法顺利执行了，除非A机器可以恢复。

# 这时候就可以知道分布式锁需要做哪些工作了 #

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

# 使用场景 #

当你的后端服务是以集群形式存在的时候，是一定需要分布式锁的。集群与分布式不同，而这里的分布式与分布式锁也不是同一回事儿。集群可以指多台服务器实现了同样的需求，比如有三台Tomcat，都负责查询模块；而分布式指多台服务器各自不同的功能点，多台功能的整合对外是一个完整的服务，比如一台Tomcat负责查询，一台负责下单。

说回集群，当后端集群要去访问同一个资源的时候，就需要对该资源加锁，保证同一时刻只能有一个对象来修改该资源数据，如果不加锁会导致什么情况呢？

举一个例子：

有两个线程（分别叫T1,T2）做的都是同样的事情，拿到一个叫做A的资源，然后对其进行+1操作。由于线程之间是不会互相通信的，于是就有可能出现下面这种情况：

T1拿到A，读入内存，此时A值为T；

T2拿到A，读入内存，此时A值为T；

T1进行+1操作，此时A实际值为T+1；

T2进行+1操作，此时A的实际值仍然为T+1;

然而，此时A经过两个线程执行+1操作，应该为T+2才对的，所以可以看出，如果没有分布式锁，就会出现数据不一致的问题。如果是上面这种简单的计算还好，如果是你的银行账户，没用分布式锁，此时有两个人给你打钱，结果只有其中一个人的到账了，另一个人的被作为无主钱财被银行充公了，肯定是不行的吧。

所以，保障数据一致性和准确性就是分布式锁的重要性。

--------------------

**分布式解决方案**

针对分布式锁的实现，目前比较常用的有以下几种方案：

1：基于数据库实现分布式锁

2：基于缓存（redis，memcached，tair）实现分布式锁

3：基于Zookeeper实现分布式锁

在分析这几种实现方案之前我们先来想一下，我们需要的分布式锁应该是怎么样的？（这里以方法锁为例，资源锁同理）

*  可以保证在分布式部署的应用集群中，同一个方法在同一时间只能被一台机器上的一个线程执行。
 *  这把锁要是一把可重入锁（避免死锁）
 *  这把锁最好是一把阻塞锁（根据业务需求考虑要不要这条）
 *  有高可用的获取锁和释放锁功能
 *  获取锁和释放锁的性能要好

--------------------

**基于数据库实现分布式锁**

**基于数据库表**

要实现分布式锁，最简单的方式可能就是直接创建一张锁表，然后通过操作该表中的数据来实现了。

当我们要锁住某个方法或资源时，我们就在该表中增加一条记录，想要释放锁的时候就删除这条记录。

创建这样一张数据库表：

CREATE TABLE `database_lock` (
        `id` BIGINT NOT NULL AUTO_INCREMENT,
        `resource` int NOT NULL COMMENT '锁定的资源',
        `description` varchar(1024) NOT NULL DEFAULT "" COMMENT '描述',
        PRIMARY KEY (`id`),
        UNIQUE KEY `uiq_idx_resource` (`resource`) 
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表';

当我们想要获得锁时，可以插入一条数据：

INSERT INTO database_lock(resource, description) VALUES (1, 'lock');

注意：在表database\_lock中，resource字段做了唯一性约束，这样如果有多个请求同时提交到数据库的话，数据库可以保证只有一个操作可以成功（其它的会报错：ERROR 1062 (23000): Duplicate entry ‘1’ for key ‘uiq\_idx\_resource’），那么那么我们就可以认为操作成功的那个请求获得了锁。

当需要释放锁的时，可以删除这条数据：

DELETE FROM database_lock WHERE resource=1;

这种实现方式非常的简单，但是需要注意以下几点：

（1）这种锁没有失效时间，一旦释放锁的操作失败就会导致锁记录一直在数据库中，其它线程无法获得锁。这个缺陷也很好解决，比如可以做一个定时任务去定时清理。  
（2）这种锁的可靠性依赖于数据库。建议设置备库，避免单点，进一步提高可靠性。  
（3）这种锁是非阻塞的，因为插入数据失败之后会直接报错，想要获得锁就需要再次操作。如果需要阻塞式的，可以弄个for循环、while循环之类的，直至INSERT成功再返回。  
（4）这种锁也是非可重入的，因为同一个线程在没有释放锁之前无法再次获得锁，因为数据库中已经存在同一份记录了。想要实现可重入锁，可以在数据库中添加一些字段，比如获得锁的主机信息、线程信息（5）等，那么在再次获得锁的时候可以先查询数据，如果当前的主机信息和线程信息等能被查到的话，可以直接把锁分配给它。

**乐观锁**

顾名思义，系统认为数据的更新在大多数情况下是不会产生冲突的，只在数据库更新操作提交的时候才对数据作冲突检测。如果检测的结果出现了与预期数据不一致的情况，则返回失败信息。

乐观锁大多数是基于数据版本(version)的记录机制实现的。何谓数据版本号？即为数据增加一个版本标识，在基于数据库表的版本解决方案中，一般是通过为数据库表添加一个 “version”字段来实现读取出数据时，将此版本号一同读出，之后更新时，对此版本号加1。在更新过程中，会对版本号进行比较，如果是一致的，没有发生改变，则会成功执行本次操作；如果版本号不一致，则会更新失败。

为了更好的理解数据库乐观锁在实际项目中的使用，这里就列举一个典型的电商库存的例子。一个电商平台都会存在商品的库存，当用户进行购买的时候就会对库存进行操作（库存减1代表已经卖出了一件）。我们将这个库存模型用下面的一张表optimistic\_lock来表述，参考如下：

CREATE TABLE `optimistic_lock` (
        `id` BIGINT NOT NULL AUTO_INCREMENT,
        `resource` int NOT NULL COMMENT '锁定的资源',
        `version` int NOT NULL COMMENT '版本信息',
        `created_at` datetime COMMENT '创建时间',
        `updated_at` datetime COMMENT '更新时间',
        `deleted_at` datetime COMMENT '删除时间', 
        PRIMARY KEY (`id`),
        UNIQUE KEY `uiq_idx_resource` (`resource`) 
    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='数据库分布式锁表';

其中：id表示主键；resource表示具体操作的资源，在这里也就是特指库存；version表示版本号。

在使用乐观锁之前要确保表中有相应的数据，比如：

INSERT INTO optimistic_lock(resource, version, created_at, updated_at) VALUES(20, 1, CURTIME(), CURTIME());

如果只是一个线程进行操作，数据库本身就能保证操作的正确性。主要步骤如下：

STEP1 - 获取资源：SELECT resource FROM optimistic_lock WHERE id = 1
    　　STEP2 - 执行业务逻辑
    　　STEP3 - 更新资源：UPDATE optimistic_lock SET resource = resource -1 WHERE id = 1

然而在并发的情况下就会产生一些意想不到的问题：比如两个线程同时购买一件商品，在数据库层面实际操作应该是库存（resource）减2，但是由于是高并发的情况，第一个线程执行之后（执行了STEP1、STEP2但是还没有完成STEP3），第二个线程在购买相同的商品（执行STEP1），此时查询出的库存并没有完成减1的动作，那么最终会导致2个线程购买的商品却出现库存只减1的情况。

在引入了version字段之后，那么具体的操作就会演变成下面的内容：

STEP1 - 获取资源： SELECT resource, version FROM optimistic_lock WHERE id = 1
    　　STEP2 - 执行业务逻辑
    　　STEP3 - 更新资源：UPDATE optimistic_lock SET resource = resource -1, version = version + 1 WHERE id = 1 AND version = oldVersion

其实，借助更新时间戳（updated\_at）也可以实现乐观锁，和采用version字段的方式相似：更新操作执行前线获取记录当前的更新时间，在提交更新时，检测当前更新时间是否与更新开始时获取的更新时间戳相等。

乐观锁的优点比较明显，由于在检测数据冲突时并不依赖数据库本身的锁机制，不会影响请求的性能，当产生并发且并发量较小的时候只有少部分请求会失败。缺点是需要对表的设计增加额外的字段，增加了数据库的冗余，另外，当应用并发量高的时候，version值在频繁变化，则会导致大量请求失败，影响系统的可用性。我们通过上述sql语句还可以看到，数据库锁都是作用于同一行数据记录上，这就导致一个明显的缺点，在一些特殊场景，如大促、秒杀等活动开展的时候，大量的请求同时请求同一条记录的行锁，会对数据库产生很大的写压力。所以综合数据库乐观锁的优缺点，乐观锁比较适合并发量不高，并且写操作不频繁的场景。

**悲观锁**

除了可以通过增删操作数据库表中的记录以外，我们还可以借助数据库中自带的锁来实现分布式锁。在查询语句后面增加FOR UPDATE，数据库会在查询过程中给数据库表增加悲观锁，也称排他锁。当某条记录被加上悲观锁之后，其它线程也就无法再改行上增加悲观锁。

悲观锁，与乐观锁相反，总是假设最坏的情况，它认为数据的更新在大多数情况下是会产生冲突的。

在使用悲观锁的同时，我们需要注意一下锁的级别。MySQL InnoDB引起在加锁的时候，只有明确地指定主键(或索引)的才会执行行锁 (只锁住被选取的数据)，否则MySQL 将会执行表锁(将整个数据表单给锁住)。

在使用悲观锁时，我们必须关闭MySQL数据库的自动提交属性（参考下面的示例），因为MySQL默认使用autocommit模式，也就是说，当你执行一个更新操作后，MySQL会立刻将结果进行提交。

mysql> SET AUTOCOMMIT = 0;
    Query OK, 0 rows affected (0.00 sec)

这样在使用FOR UPDATE获得锁之后可以执行相应的业务逻辑，执行完之后再使用COMMIT来释放锁。

我们不妨沿用前面的database\_lock表来具体表述一下用法。假设有一线程A需要获得锁并执行相应的操作，那么它的具体步骤如下：

STEP1 - 获取锁：SELECT \* FROM database\_lock WHERE id = 1 FOR UPDATE;。  
　　　　STEP2 - 执行业务逻辑。  
　　　　STEP3 - 释放锁：COMMIT。

如果另一个线程B在线程A释放锁之前执行STEP1，那么它会被阻塞，直至线程A释放锁之后才能继续。注意，如果线程A长时间未释放锁，那么线程B会报错，参考如下（lock wait time可以通过innodb\_lock\_wait\_timeout来进行配置）：

ERROR 1205 (HY000): Lock wait timeout exceeded; try restarting transaction

上面的示例中演示了指定主键并且能查询到数据的过程（触发行锁），如果查不到数据那么也就无从“锁”起了。

如果未指定主键（或者索引）且能查询到数据，那么就会触发表锁，比如STEP1改为执行（这里的version只是当做一个普通的字段来使用，与上面的乐观锁无关）：

SELECT * FROM database_lock WHERE description='lock' FOR UPDATE;

或者主键不明确也会触发表锁，又比如STEP1改为执行：

SELECT * FROM database_lock WHERE id>0 FOR UPDATE;

注意，虽然我们可以显示使用行级锁（指定可查询的主键或索引），但是MySQL会对查询进行优化，即便在条件中使用了索引字段，但是否真的使用索引来检索数据是由MySQL通过判断不同执行计划的代价来决定的，如果MySQL认为全表扫描效率更高，比如对一些很小的表，它有可能不会使用索引，在这种情况下InnoDB将使用表锁，而不是行锁。

在悲观锁中，每一次行数据的访问都是独占的，只有当正在访问该行数据的请求事务提交以后，其他请求才能依次访问该数据，否则将阻塞等待锁的获取。悲观锁可以严格保证数据访问的安全。但是缺点也明显，即每次请求都会额外产生加锁的开销且未获取到锁的请求将会阻塞等待锁的获取，在高并发环境下，容易造成大量请求阻塞，影响系统可用性。另外，悲观锁使用不当还可能产生死锁的情况。

还有一个问题，就是我们要使用排他锁来进行分布式锁的lock，那么一个排他锁长时间不提交，就会占用数据库连接。一旦类似的连接变得多了，就可能把数据库连接池撑爆

总结

总结一下使用数据库来实现分布式锁的方式，这两种方式都是依赖数据库的一张表，一种是通过表中的记录的存在情况确定当前是否有锁存在，另外一种是通过数据库的排他锁来实现分布式锁。

数据库实现分布式锁的优点

直接借助数据库，容易理解。

数据库实现分布式锁的缺点

会有各种各样的问题，在解决问题的过程中会使整个方案变得越来越复杂。

操作数据库需要一定的开销，性能问题需要考虑。

使用数据库的行级锁并不一定靠谱，尤其是当我们的锁表并不大的时候。

--------------------

**二：基于Zookeeper实现分布式锁**

ZooKeeper是一个为分布式应用提供一致性服务的开源组件，它内部是一个分层的文件系统目录树结构，规定同一个目录下只能有一个唯一文件名。基于ZooKeeper实现分布式锁的步骤如下：

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

可以直接使用zookeeper第三方库Curator客户端，这个客户端中封装了一个可重入的锁服务。

![23c62b4250582329bc56935d3e5ddcf9.png][]

Curator提供的InterProcessMutex是分布式锁的实现。acquire方法用户获取锁，release方法用于释放锁。

使用ZK实现的分布式锁好像完全符合了本文开头我们对一个分布式锁的所有期望。但是，其实并不是，Zookeeper实现的分布式锁其实存在一个缺点，那就是性能上可能并没有缓存服务那么高。因为每次在创建锁和释放锁的过程中，都要动态创建、销毁瞬时节点来实现锁功能。ZK中创建和删除节点只能通过Leader服务器来执行，然后将数据同不到所有的Follower机器上。

其实，使用Zookeeper也有可能带来并发问题，只是并不常见而已。考虑这样的情况，由于网络抖动，客户端可ZK集群的session连接断了，那么zk以为客户端挂了，就会删除临时节点，这时候其他客户端就可以获取到分布式锁了。就可能产生并发问题。这个问题不常见是因为zk有重试机制，一旦zk集群检测不到客户端的心跳，就会重试，Curator客户端支持多种重试策略。多次重试之后还不行的话才会删除临时节点。（所以，选择一个合适的重试策略也比较重要，要在锁的粒度和并发之间找一个平衡。）

也可以使用创建临时节点的方式实现分布式锁，例子：

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 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 ZooKeeperLock implements Watcher {
    
        private ZooKeeper zk = null;
        private String rootLockNode;            // 锁的根节点  locks
        private String lockName;                // 竞争资源，用来生成子节点名称  test1
        private String currentLock;             // 当前锁
        private String waitLock;                // 等待的锁（前一个锁）
        private CountDownLatch countDownLatch;  // 计数器（用来在加锁失败时阻塞加锁线程）
        private int sessionTimeout = 30000;     // 超时时间
        
        // 1. 构造器中创建ZK链接，创建锁的根节点
        public ZooKeeperLock(String zkAddress, String rootLockNode, String lockName) {
            this.rootLockNode = rootLockNode;
            this.lockName = lockName;
            try {
                // 创建连接，zkAddress格式为：IP:PORT
                zk = new ZooKeeper(zkAddress,this.sessionTimeout,this);
                // 检测锁的根节点是否存在，不存在则创建
                Stat stat = zk.exists(rootLockNode,false);
                if (null == stat) {
                    zk.create(rootLockNode, 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();
            }
        }
        
        // 2. 加锁方法，先尝试加锁，不能加锁则等待上一个锁的释放
        public boolean lock() {
            if (this.tryLock()) {
                System.out.println("线程【" + Thread.currentThread().getName() + "】加锁（" + this.currentLock + "）成功！");
                return true;
            } else {
                return waitOtherLock(this.waitLock, this.sessionTimeout);
            }
        }
        
        public boolean tryLock() {
            // 分隔符
            String split = "_lock_";
            if (this.lockName.contains("_lock_")) {
                throw new RuntimeException("lockName can't contains '_lock_' ");
            }
            try {
                // 创建锁节点（临时有序节点）
                this.currentLock = zk.create(this.rootLockNode + "/" + this.lockName + split, new byte[0],
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
                System.out.println("线程【" + Thread.currentThread().getName() 
                            + "】创建锁节点（" + this.currentLock + "）成功，开始竞争...");
                // 取所有子节点
                List<String> nodes = zk.getChildren(this.rootLockNode, false);
                // 取所有竞争lockName的锁
                List<String> lockNodes = new ArrayList<String>();
                for (String nodeName : nodes) {
                    if (nodeName.split(split)[0].equals(this.lockName)) {
                        lockNodes.add(nodeName);
                    }
                }
                Collections.sort(lockNodes);
                // 取最小节点与当前锁节点比对加锁
                String currentLockPath = this.rootLockNode + "/" + lockNodes.get(0);
                if (this.currentLock.equals(currentLockPath)) {
                    return true;
                }
                // 加锁失败，设置前一节点为等待锁节点
                String currentLockNode = this.currentLock.substring(this.currentLock.lastIndexOf("/") + 1);
                int preNodeIndex = Collections.binarySearch(lockNodes, currentLockNode) - 1;
                this.waitLock = lockNodes.get(preNodeIndex);
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return false;
        }
    
        private boolean waitOtherLock(String waitLock, int sessionTimeout) {
            boolean islock = false;
            try {
                // 监听等待锁节点
                String waitLockNode = this.rootLockNode + "/" + waitLock;
                Stat stat = zk.exists(waitLockNode,true); // watcher等待的节点
                if (null != stat) {
                    System.out.println("线程【" + Thread.currentThread().getName() 
                                + "】锁（" + this.currentLock + "）加锁失败，等待锁（" + waitLockNode + "）释放...");
                    // 设置计数器，使用计数器阻塞线程
                    this.countDownLatch = new CountDownLatch(1);
                    islock = this.countDownLatch.await(sessionTimeout,TimeUnit.MILLISECONDS);
                    this.countDownLatch = null;
                    if (islock) {
                        System.out.println("线程【" + Thread.currentThread().getName() + "】锁（" 
                                    + this.currentLock + "）加锁成功，锁（" + waitLockNode + "）已经释放");
                    } else {
                        System.out.println("线程【" + Thread.currentThread().getName() + "】锁（" 
                                    + this.currentLock + "）加锁失败...");
                    }
                } else {
                    islock = true;
                }
            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return islock;
        }
        
        // 3. 释放锁
        public void unlock() throws InterruptedException {
            try {
                Stat stat = zk.exists(this.currentLock,false);
                if (null != stat) {
                    System.out.println("线程【" + Thread.currentThread().getName() + "】释放锁 " + this.currentLock);
                    zk.delete(this.currentLock, -1);
                    this.currentLock = null;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (KeeperException e) {
                e.printStackTrace();
            } finally {
                zk.close();
            }
        }
        
        // 4. 监听器回调
        @Override
        public void process(WatchedEvent watchedEvent) {
            if (null != this.countDownLatch && watchedEvent.getType() == Event.EventType.NodeDeleted) {
                // 计数器减一，恢复线程操作
                this.countDownLatch.countDown();
            }
        }
    
    }

public class Test {
        public static void doSomething() {
            System.out.println("线程【" + Thread.currentThread().getName() + "】正在运行...");
        }
    
        public static void main(String[] args) {
            Runnable runnable = new Runnable() {
                public void run() {
                    ZooKeeperLock lock = null;
                    lock = new ZooKeeperLock("127.0.0.1:2181","/locks", "test1");
                    if (lock.lock()) {    // 获取zookeeper的节点锁
                        doSomething();
                        try {
                            Thread.sleep(1000);
                            lock.unlock();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            };
    
            for (int i = 0; i < 10; i++) {
                Thread t = new Thread(runnable);
                t.start();
            }
        }
    }

console结果：

![c41748f87c09378faa8b63ba9ffa2e0d.png][]

![4fcbca8e1c584bb2d9d9cb60b195937c.png][]

--------------------

**三：基于缓存实现分布式锁**

相比较于基于数据库实现分布式锁的方案来说，基于缓存来实现在性能方面会表现的更好一点。而且很多缓存是可以集群部署的，可以解决单点问题。

目前有很多成熟的缓存产品，包括Redis，memcached以及我们公司内部的Tair。

这里以Tair为例来分析下使用缓存实现分布式锁的方案。关于Redis和memcached在网络上有很多相关的文章，并且也有一些成熟的框架及算法可以直接使用。

基于Tair的实现分布式锁其实和Redis类似，其中主要的实现方式是使用`TairManager.put`方法来实现。

![1974f36bf0f30df9339e4a9b69334124.png][]

以上实现方式同样存在几个问题：

> 1、这把锁没有失效时间，一旦解锁操作失败，就会导致锁记录一直在tair中，其他线程无法再获得到锁。
> 
> 2、这把锁只能是非阻塞的，无论成功还是失败都直接返回。
> 
> 3、这把锁是非重入的，一个线程获得锁之后，在释放锁之前，无法再次获得该锁，因为使用到的key在tair中已经存在。无法再执行put操作。

当然，同样有方式可以解决。

*  没有失效时间？tair的put方法支持传入失效时间，到达时间之后数据会自动删除。
 *  非阻塞？while重复执行。
 *  非可重入？在一个线程获取到锁之后，把当前主机信息和线程信息保存起来，下次再获取之前先检查自己是不是当前锁的拥有者。

但是，失效时间我设置多长时间为好？如何设置的失效时间太短，方法没等执行完，锁就自动释放了，那么就会产生并发问题。如果设置的时间太长，其他获取锁的线程就可能要平白的多等一段时间。这个问题使用数据库实现分布式锁同样存在

总结

可以使用缓存来代替数据库来实现分布式锁，这个可以提供更好的性能，同时，很多缓存服务都是集群部署的，可以避免单点问题。并且很多缓存服务都提供了可以用来实现分布式锁的方法，比如Tair的put方法，redis的setnx方法等。并且，这些缓存服务也都提供了对数据的过期自动删除的支持，可以直接设置超时时间来控制锁的释放。

**使用缓存实现分布式锁的优点**

性能好，实现起来较为方便。

**使用缓存实现分布式锁的缺点**

通过超时时间来控制锁的失效时间并不是十分的靠谱。

--------------------

**三种方案的比较**

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

从理解的难易程度角度（从低到高）

数据库 > 缓存 > Zookeeper

从实现的复杂性角度（从低到高）

Zookeeper >= 缓存 > 数据库

从性能角度（从高到低）

缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）

Zookeeper > 缓存 > 数据库

![20210508103852838.jpeg][]

[23c62b4250582329bc56935d3e5ddcf9.png]: /images/20221021/f2a53601dd014f5d9f3a8ca248d5943c.png
[c41748f87c09378faa8b63ba9ffa2e0d.png]: /images/20221021/e35e3b4f4bbf48669616b1cb3ff4eec2.png
[4fcbca8e1c584bb2d9d9cb60b195937c.png]: /images/20221021/8d72d06cfb1143daaf1873ca057e1ee2.png
[1974f36bf0f30df9339e4a9b69334124.png]: /images/20221021/133d64eace474fa6bc500672c662f414.png
[20210508103852838.jpeg]: /images/20221021/ce81f02bb4cf45599d316789e036a0bf.png