多种方式实现分布式锁的介绍和比较以及代码实现

迈不过友情╰ 2022-10-01 12:53 202阅读 0赞

最近在看分布式系统的相关文章，对于几种分布式锁的实现做了如下整理：

分布式锁主要包含数据库、redis、memcached、zookeeper四种，以下是各个分布式锁的优缺点

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3ODM3Mzgy_size_16_color_FFFFFF_t_70][]

为什么要使用分布式锁：

### **在分布式系统中，各系统同步访问共同的资源是很常见的。因此我们常常需要协调他们的动作。 如果不同的系统或是同一个系统的不同主机之间共享了一个或一组资源，那么访问这些资源的时候，往往需要互斥来防止彼此干扰来保证一致性，在这种情况下，便需要使用到分布式锁。**在同一个jvm进程中时，可以使用JUC提供的一些锁来解决多个线程竞争同一个共享资源时候的线程安全问题，但是当多个不同机器上的不同jvm进程共同竞争同一个共享资源时候，juc包的锁就无能无力了，这时候就需要分布式锁了。 ###

首先来介绍下java的可重入锁和不可重入锁：

1.不可重入锁：

public class Lock{  
        private boolean isLocked = false;  
        public synchronized void lock()  
            throws InterruptedException{  
            while(isLocked){  
                wait();  
            }  
            isLocked = true;  
        }  
     
        public synchronized void unlock(){  
            isLocked = false;  
            notify();  
        }  
    }

2.可重入锁：

public class Lock{
        boolean isLocked = false;
        Thread  lockedBy = null;
        int lockedCount = 0;
        public synchronized void lock() throws InterruptedException{
            Thread callingThread = Thread.currentThread();
            while(isLocked && lockedBy != callingThread){
                wait();
            }
            isLocked = true;
            lockedCount++;
            lockedBy = callingThread;
        }
        public synchronized void unlock(){
            if(Thread.curentThread() == this.lockedBy){
                lockedCount--;
                if(lockedCount == 0){
                    isLocked = false;
                    notify();
                }
            }
        }
    }

从代码实现来看，可重入锁增加了两个状态，锁的计数器和被锁的线程，实现基本上和不可重入的实现一样，如果不同的线程进来，这个锁是没有问题的，但是如果进行递归计算的时候，如果加锁，不可重入锁就会出现死锁的问题。可以参考

*  可重入锁：可以再次进入方法A，就是说在释放锁前此线程可以再次进入方法A（方法A递归）。
 *  不可重入锁（自旋锁）：不可以再次进入方法A，也就是说获得锁进入方法A是此线程在释放锁钱唯一的一次进入方法A。

那这两种锁除了在可能会导致死锁方面的区别外，效率有差别了。具体可以参考

![20190612174832523.png][]包下的两种锁的实现：

//InterProcessMutex lock = new InterProcessMutex(client,ZOOKEEPER_PATH); //可重入锁
                    InterProcessSemaphoreMutex lock = new InterProcessSemaphoreMutex(client,ZOOKEEPER_PATH); //不可重入锁

不可重入锁花费的时间要大于可重入锁花费的时间

第一个是可重入锁的关键实现代码，第二个是不可重入的关键实现代码：

private boolean internalLock(long time, TimeUnit unit) throws Exception {
            Thread currentThread = Thread.currentThread();
            InterProcessMutex.LockData lockData = (InterProcessMutex.LockData)this.threadData.get(currentThread);
            if(lockData != null) {
                lockData.lockCount.incrementAndGet();
                return true;
            } else {
                String lockPath = this.internals.attemptLock(time, unit, this.getLockNodeBytes());
                if(lockPath != null) {
                    InterProcessMutex.LockData newLockData = new InterProcessMutex.LockData(currentThread, lockPath, null);
                    this.threadData.put(currentThread, newLockData);
                    return true;
                } else {
                    return false;
                }
            }
        }

public Collection<Lease> acquire(int qty, long time, TimeUnit unit) throws Exception
        {
            long startMs = System.currentTimeMillis();
            boolean hasWait = (unit != null);
            long waitMs = hasWait ? TimeUnit.MILLISECONDS.convert(time, unit) : 0;
    
            Preconditions.checkArgument(qty > 0, "qty cannot be 0");
    
            ImmutableList.Builder<Lease> builder = ImmutableList.builder();
            boolean success = false;
            try
            {
                while ( qty-- > 0 )
                {
                    int retryCount = 0;
                    long startMillis = System.currentTimeMillis();
                    boolean isDone = false;
                    while ( !isDone )
                    {
                        switch ( internalAcquire1Lease(builder, startMs, hasWait, waitMs) )
                        {
                            case CONTINUE:
                            {
                                isDone = true;
                                break;
                            }
    
                            case RETURN_NULL:
                            {
                                return null;
                            }
    
                            case RETRY_DUE_TO_MISSING_NODE:
                            {
                                // gets thrown by internalAcquire1Lease when it can't find the lock node
                                // this can happen when the session expires, etc. So, if the retry allows, just try it all again
                                if ( !client.getZookeeperClient().getRetryPolicy().allowRetry(retryCount++, System.currentTimeMillis() - startMillis, RetryLoop.getDefaultRetrySleeper()) )
                                {
                                    throw new KeeperException.NoNodeException("Sequential path not found - possible session loss");
                                }
                                // try again
                                break;
                            }
                        }
                    }
                }
                success = true;
            }
            finally
            {
                if ( !success )
                {
                    returnAll(builder.build());
                }
            }
    
            return builder.build();
        }
    
    
     private InternalAcquireResult internalAcquire1Lease(ImmutableList.Builder<Lease> builder, long startMs, boolean hasWait, long waitMs) throws Exception
        {
            if ( client.getState() != CuratorFrameworkState.STARTED )
            {
                return InternalAcquireResult.RETURN_NULL;
            }
    
            if ( hasWait )
            {
                long thisWaitMs = getThisWaitMs(startMs, waitMs);
                if ( !lock.acquire(thisWaitMs, TimeUnit.MILLISECONDS) )
                {
                    return InternalAcquireResult.RETURN_NULL;
                }
            }
            else
            {
                lock.acquire();
            }
    
            Lease lease = null;
    
            try
            {
                PathAndBytesable<String> createBuilder = client.create().creatingParentContainersIfNeeded().withProtection().withMode(CreateMode.EPHEMERAL_SEQUENTIAL);
                String path = (nodeData != null) ? createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME), nodeData) : createBuilder.forPath(ZKPaths.makePath(leasesPath, LEASE_BASE_NAME));
                String nodeName = ZKPaths.getNodeFromPath(path);
                lease = makeLease(path);
    
                if ( debugAcquireLatch != null )
                {
                    debugAcquireLatch.await();
                }
    
                try
                {
                    synchronized(this)
                    {
                        for(;;)
                        {    
                            List<String> children;
                            try
                            {
                                children = client.getChildren().usingWatcher(watcher).forPath(leasesPath);
                            }
                            catch ( Exception e )
                            {
                                if ( debugFailedGetChildrenLatch != null )
                                {
                                    debugFailedGetChildrenLatch.countDown();
                                }
                                returnLease(lease); // otherwise the just created ZNode will be orphaned causing a dead lock
                                throw e;
                            }
                            if ( !children.contains(nodeName) )
                            {
                                log.error("Sequential path not found: " + path);
                                returnLease(lease);
                                return InternalAcquireResult.RETRY_DUE_TO_MISSING_NODE;
                            }
        
                            if ( children.size() <= maxLeases )
                            {
                                break;
                            }
                            if ( hasWait )
                            {
                                long thisWaitMs = getThisWaitMs(startMs, waitMs);
                                if ( thisWaitMs <= 0 )
                                {
                                    returnLease(lease);
                                    return InternalAcquireResult.RETURN_NULL;
                                }
                                wait(thisWaitMs);
                            }
                            else
                            {
                                wait();
                            }
                        }
                    }
                }
                finally
                {
                    client.removeWatchers();
                }
            }
            finally
            {
                lock.release();
            }
            builder.add(Preconditions.checkNotNull(lease));
            return InternalAcquireResult.CONTINUE;
        }

首先可重入锁的关键代码逻辑非常简单，而且使用了Atomic原子操作，效率非常高，但是不可重入锁代码量非常大，为了实现一个类似于Semaphore的工具，进行很多的判断，效率非常低，有兴趣的可以升入研究下这两种锁。

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

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

一个好的分布式锁需要具备以下特性：

*  可重入
 *  同一时间点,只有一个线程持有锁
 *  容错性, 当锁节点宕机时, 能及时释放锁
 *  高性能
 *  无单点问题

一、数据库锁

1.数据库锁的实现流程：

![16b45758268286d7_imageslim][]![åå¸å¼éæµç¨å¾-åºäºæ°æ®åºåå¸å¼é.jpg][-_.jpg]

基于数据库的分布式锁：

1.常用的一种方式是使用表的唯一约束特性。当往数据库中成功插入一条数据时, 代表只获取到锁。将这条数据从数据库中删除，则释放送。

**因此需要创建一张锁表**

CREATE TABLE `methodLock` (
      `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
      `method_name` varchar(64) NOT NULL DEFAULT '' COMMENT '锁定的方法名',
      `cust_id` varchar(1024) NOT NULL DEFAULT '客户端唯一编码',
      `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '保存数据时间，自动生成',
      PRIMARY KEY (`id`),
      UNIQUE KEY `uidx_method_name` (`method_name `) USING BTREE
    )
     ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';
    复制代码

**添加锁**

insert into methodLock(method_name,cust_id) values (‘method_name’,‘cust_id’)
    复制代码

这里cust\_id 可以是机器的mac地址+线程编号, 确保一个线程只有唯一的一个编号。***通过这个编号， 可以有效的判断是否为锁的创建者，从而进行锁的释放以及重入锁判断***

**释放锁**

delete from methodLock where method_name ='method_name' and cust_id = 'cust_id'
    复制代码

**重入锁判断**

select 1 from methodLock where method_name ='method_name' and cust_id = 'cust_id'
    复制代码

**加锁以及释放锁的代码示例**

/**
    * 获取锁
    */
    public boolean lock(String methodName){
        boolean success = false;
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        try{
            //添加锁
           success = insertLock(methodName, custId);
        } catch(Exception e) {
            //如添加失败
        }
        return success;
    }
    
    /**
    * 释放锁
    */
    public boolean unlock(String methodName) {
        boolean success = false;
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        try{
            //添加锁
           success = deleteLock(methodName, custId);
        } catch(Exception e) {
            //如添加失败
        }
        return success;
    }
    复制代码

**完整流程**

public void test() {
        String methodName = "methodName";
        //判断是否重入锁
        if (!checkReentrantLock(methodName)) {
            //非重入锁
            while (!lock(methodName)) {
                //获取锁失败, 则阻塞至获取锁
                try{
                    Thread.sleep(100)
                } catch(Exception e) {
                }
            }
        }
        //TODO 业务处理
        
        //释放锁
        unlock(methodName);
    }
    复制代码

以上代码还存在一些问题:

*  没有失效时间。 解决方案:设置一个定时处理, 定期清理过期锁
 *  单点问题。 解决方案: 弄几个备份数据库，数据库之前双向同步，一旦挂掉快速切换到备库上

2.基于数据库的乐观锁

数据库乐观锁也能保证线程安全，通常在代码层面我们都会这样做：

select goods\_num from goods where goods\_name = "小本子";  
update goods set goods\_num = goods\_num -1 where goods\_name = "小本子";  
上面的SQL是一组的，通常先查询出当前的goods\_num，然后再goods\_num上进行减1的操作修改库存，当并发的情况下，这条语句可能导致原本库存为3的一个商品经过两个人购买还剩下2库存的情况就会导致商品的多卖。那么数据库乐观锁是如何实现的呢？  
首先定义一个version字段用来当作一个版本号，每次的操作就会变成这样：

select goods\_num,version from goods where goods\_name = "小本子";  
update goods set goods\_num = goods\_num -1,version =查询的version值自增 where goods\_name ="小本子" and version=查询出来的version；  
为什么加个version字段就满足了呢，因为数据库本身特性的帮忙，update语句执行的时候，如果更新的时候update语句不走索引就会将表锁住保证了一个时刻只有一个线程能进入更新，等这次更新释放锁以后才会执行下一次的update操作，如果提交的数据版本号大于数据库表当前版本号，则予以更新，否则认为是过期数据，这样就能保证了程序的安全性。

优点  
  乐观锁机制避免了长事务中的数据库加锁开销，大大提升了大并发量下的系统整体性能表现。

缺点  
  过分的依赖数据的逻辑，无法控制系统外部对数据库的操作。  
3.基于数据库的悲观锁

我们还用上面创建的数据库表，可以通过数据库的排它锁来实现分布式锁。基于MySQL的InnoDB引擎，可以使用以下方法来实现加锁操作：

public boolean lock()\{  
         Connection.setAutoCommit(false);  
         while (true) \{  
               try \{  
                     result = select \* from MethodLock where methodName = 'xxxx' for update;  
                     if (result == null) \{  
                           return false;  
                     \}  
               \} catch (Exception e) \{  
   
               \}  
               sleep(1000);  
         \}  
         returnType false;  
   \}  
在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁（这里再多提一句，InnoDB引擎在加锁的时候，只有通过索引进行检索的时候才会使用行级锁，否则会使用表级锁。这里我们希望使用行级锁，就要给method\_name添加索引，值得注意的是，这个索引一定要创建成唯一索引，否则会出现多个重载方法之间无法同时被访问的问题。重载方法的话建议把参数类型也加上）。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。

我们可以认为获得排它锁的线程即可获得分布式锁，当获取到锁之后，可以执行方法的业务逻辑，执行完方法之后，再通过以下方法解锁：

public void unlock()\{  
    connection.commit();  
\}  
通过connection.commit()操作来释放锁。

这里还可能存在另外一个问题，虽然我们对method\_name 使用了唯一索引，并且显示使用for update来使用行级锁。但是，MySql会对查询进行优化，即便在条件中使用了索引字段，但是否使用索引来检索数据是由 MySQL 通过判断不同执行计划的代价来决定的，如果 MySQL 认为全表扫效率更高，比如对一些很小的表，它就不会使用索引，这种情况下 InnoDB 将使用表锁，而不是行锁。如果发生这种情况就悲剧了。

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

这种方法可以有效的解决上面提到的无法释放锁和阻塞锁的问题：

1、阻塞锁？for update语句会在执行成功后立即返回，在执行失败时一直处于阻塞状态，直到成功。   
2、锁定之后服务宕机，无法释放？使用这种方式，服务宕机之后数据库会自己把锁释放掉。

但是还是无法解决数据库单点和可重入的问题。

优点：直接借助数据库容易理解；

缺点：   
1、会有各种各样的问题，在解决问题的过程中会使整个方案变的越来越复杂。   
2、操作数据库需要一定的开销，性能问题需要考虑。

二、基于Redis的分布式锁

![åå¸å¼éæµç¨å¾-åºäºredisåå¸å¼é.jpg][-_redis_.jpg]

使用redis 的set(String key, String value, String nxxx, String expx, int time)命令

*  第一个为key，我们使用key来当锁，因为key是唯一的。
 *  第二个为value，我们传的是custId，这里cust\_id 可以是机器的mac地址+线程编号, 确保一个线程只有唯一的一个编号。***通过这个编号， 可以有效的判断是否为锁的创建者，从而进行锁的释放以及重入锁判断***
 *  第三个为nxxx，这个参数我们填的是NX，意思是SET IF NOT EXIST，即当key不存在时，我们进行set操作；若key已经存在，则不做任何操作
 *  第四个为expx，这个参数我们传的是PX，意思是我们要给这个key加一个过期的设置，具体时间由第五个参数决定。
 *  第五个为time，与第四个参数相呼应，代表key的过期时间。

**代码示例**

private static final String LOCK_SUCCESS = "OK";
    private static final String SET_IF_NOT_EXIST = "NX";
    private static final String SET_WITH_EXPIRE_TIME = "PX";
    private static final Long RELEASE_SUCCESS = 1L;
    
    // Redis客户端
    private Jedis jedis;
    
    /**
     * 尝试获取分布式锁
     * @param lockKey 锁
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public boolean lock(String lockKey, int expireTime) {
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        String result = jedis.set(lockKey, custId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
    
        if (LOCK_SUCCESS.equals(result)) {
            return true;
        }
        
        return false;
    }
    
    /**
     * 释放分布式锁
     * @param lockKey 锁
     * @param requestId 请求标识
     * @return 是否释放成功
     */
    public boolean unlock(String lockKey,) {
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        //script为Lua脚本，该脚本是原子性的，来避免释放掉的是其他客户端的redis锁的情况
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
        Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(custId));
    
        if (RELEASE_SUCCESS.equals(result)) {
            return true;
        }
        return false;
    }
    
    /**
     * 获取锁信息
     * @param lockKey 锁
     * @return 是否重入锁
     */
    public boolean checkReentrantLock(String lockKey){
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        
        //获取当前锁的客户唯一表示码
        String currentCustId = redis.get(lockKey);
        if (custId.equals(currentCustId)) {
            return true;
        }
        return false;
    }

**完整流程**

public void test() {
        String lockKey = "lockKey";
        //判断是否重入锁
        if (!checkReentrantLock(lockKey)) {
            //非重入锁
            while (!lock(lockKey)) {
                //获取锁失败, 则阻塞至获取锁
                try{
                    Thread.sleep(100)
                } catch(Exception e) {
                }
            }
        }
        //TODO 业务处理
        
        //释放锁
        unlock(lockKey);
    }

三、基于memcached的分布式锁

![åå¸å¼éæµç¨å¾-åºäºmemcachedåå¸å¼é.jpg][-_memcached_.jpg]

memcached的实现方式和redis类似, 使用的是命令add(key, value, expireDate),注:仅当缓存中不存在键时，才会添加成功

*  第一个为key，我们使用key来当锁，因为key是唯一的。
 *  第二个为value，我们传的是custId，这里cust\_id
 *  第三个为expireDate, 设置一个过期时间,比如： new Date(1000\*10)，则表示十秒之后从Memcached内存缓存中删除）。

**代码示例**

// Redis客户端
    private MemCachedClient memCachedClient;
    
    /**
     * 尝试获取分布式锁
     * @param lockKey 锁
     * @param expireTime 超期时间
     * @return 是否获取成功
     */
    public boolean lock(String lockKey, Date expireDate) {
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        Boolean result = false;
        try {
            result = memCachedClient.add(lockKey, custId,expireDate);
        } catch(Excetion e) {
        }
        return result;
    }
    
    /**
     * 释放分布式锁
     * @param lockKey 锁
     * @param requestId 请求标识
     * @return 是否释放成功
     */
    public boolean unlock(String lockKey,) {
        //获取客户唯一识别码,例如:mac+线程信息
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        Boolean result = false;
        try {
            String currentCustId = memCachedClient.get(lockKey);
            if (custId.equals(currentCustId)) {
                result = memCachedClient.delete(lockKey, custId,expireDate);
            }
        } catch(Excetion e) {
        }
        return result;
    }
    
    /**
     * 获取锁信息
     * @param lockKey 锁
     * @return 是否重入锁
     */
    public boolean checkReentrantLock(String lockKey){
        //获取客户唯一识别码,例如:mac+线程信息
        String custId = getCustId();
        //获取当前锁的客户唯一表示码
        try {
             String currentCustId = memCachedClient.get(lockKey);
            if (custId.equals(currentCustId)) {
                return true;
            }
        } catch(Excetion e) {
        }
       
        return false;
    }

**完整流程**

总结

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

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

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

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

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

## 四. 基于zookeeper的分布式锁 ##

![åå¸å¼éæµç¨å¾-åºäºzookeeperåå¸å¼é.jpg][-_zookeeper_.jpg]

基于zookeeper临时有序节点可以实现的分布式锁。

大致思想即为：每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。 判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。 当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

来看下Zookeeper能不能解决前面提到的问题。

*  锁无法释放？使用Zookeeper可以有效的解决锁无法释放的问题，因为在创建锁的时候，客户端会在ZK中创建一个临时节点，一旦客户端获取到锁之后突然挂掉（Session连接断开），那么这个临时节点就会自动删除掉。其他客户端就可以再次获得锁。
 *  非阻塞锁？使用Zookeeper可以实现阻塞的锁，客户端可以通过在ZK中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，Zookeeper会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，那么自己就获取到锁，便可以执行业务逻辑了。
 *  不可重入？使用Zookeeper也可以有效的解决不可重入的问题，客户端在创建节点的时候，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获取锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么自己直接获取到锁，如果不一样就再创建一个临时的顺序节点，参与排队。
 *  单点问题？使用Zookeeper可以有效的解决单点问题，ZK是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务。

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

**完整流程**

public void test() {
        //Curator提供的InterProcessMutex是分布式锁的实现。通过acquire获得锁，并提供超时机制，release方法用于释放锁。
        InterProcessMutex lock = new InterProcessMutex(client, ZK_LOCK_PATH);
        try {
            //获取锁
            if (lock.acquire(10 * 1000, TimeUnit.SECONDS)) {
                //TODO 业务处理
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                //释放锁
                lock.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
    }

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

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

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

Redis和Memcache的分布式锁可以视作为同一种缓存锁。下面对以上的三种分布式锁做一个比较：

### 三种方案的比较 ###

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

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

数据库 > 缓存 > Zookeeper

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

Zookeeper >= 缓存 > 数据库

从性能角度（从高到低）

缓存 > Zookeeper >= 数据库

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

Zookeeper > 缓存 > 数据库

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L20wXzM3ODM3Mzgy_size_16_color_FFFFFF_t_70]: /images/20220118/22f1998497244381a2c26928ae3b191d.png
[20190612174832523.png]: /images/20220118/8ad81ea607a7403e8578fe065fcf8401.png
[16b45758268286d7_imageslim]: https://user-gold-cdn.xitu.io/2019/6/11/16b45758268286d7?imageslim
[-_.jpg]: https://user-gold-cdn.xitu.io/2019/6/11/16b45758268286d7?imageView2/0/w/1280/h/960/format/webp/ignore-error/1
[-_redis_.jpg]: https://user-gold-cdn.xitu.io/2019/6/11/16b4575cdd932256?imageView2/0/w/1280/h/960/format/webp/ignore-error/1
[-_memcached_.jpg]: https://user-gold-cdn.xitu.io/2019/6/11/16b45763efd74355?imageView2/0/w/1280/h/960/format/webp/ignore-error/1
[-_zookeeper_.jpg]: https://user-gold-cdn.xitu.io/2019/6/11/16b45767afe12c7f?imageView2/0/w/1280/h/960/format/webp/ignore-error/1