【分布式系列】分布式锁原理篇-Redisson、Zookeeper、Database

旧城等待， 2023-06-28 08:55 51阅读 0赞

# 热门系列： #

*  ### [【分布式系列】分布式事务：TX-LCN分布式事务框架整合实践][TX-LCN] ###
 *  ###   [程序人生，精彩抢先看][Link 1] ###

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

# **目录** #

1.分布式锁起源

2.分布式锁分类

2.1 Redisson分布式锁

2.2 Zookeeper

2.3 Database实现分布式锁

3. 分布式锁对比

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

# 1.分布式锁起源 #

在分布式出现之前，一般多线程操作，为了防止高并发产生的问题，使用Synchronize和ReentrantLock等加锁方式解决。因为项目服务采用的是单进程的多线程操作。而如今，集群和微服务化兴起的时代，单进程锁已经无法满足我们的业务需求。于是，也就出现了今天我们要讲到的分布式锁的概念。

对于分布式锁的设计，主要需要满足一下几个点：

*  互斥性。在任意时刻，只有一个客户端能持有锁。
 *  不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
 *  具有容错性。只要大部分的Redis节点正常运行，客户端就可以加锁和解锁。
 *  解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了

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

# 2.分布式锁分类 #

分布式锁目前大体可以分为三类：**Redisson、Zookeeper、Database**

## 2.1 Redisson分布式锁 ##

说到Redisson，大家应该都不陌生。Redisson是基于Redis内存数据库封装而来的框架。在集成Redis大部分功能的同时，且实现了类似ReentrantLock的锁功能。

原理：Redisson的锁，其实是通过Redis的setnx和del等方法结合lua脚本语言来实现的。其中获取锁和释放锁的流程如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70 1][]

加锁代码解读：

使用过程中，我们通过getLock（）方法获取到一个RLock对象，其lock（）方法中实际执行方法是lockInterruptibly()

@Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        // 1.尝试获取锁
        Long ttl = tryAcquire(leaseTime, unit);
        // 2.获得锁成功
        if (ttl == null) {
            return;
        }
        // 3.如果没有获得锁，则需等待锁被释放，并通过 Redis 的 channel 订阅锁释放的消息，这里的具体 
        // 实现本文也不深入，只是简单提一下 Redisson 在执行 Redis 命令时提供了同步和异步的两种实 
        // 现，但实际上同步的实现都是基于异步的，具体做法是使用 Netty 中的异步工具 Future 和 
        // FutureListener 结合 JDK 中的 CountDownLatch 一起实现。
        long threadId = Thread.currentThread().getId();
        Future<RedissonLockEntry> future = subscribe(threadId);
        get(future);
    
        try {
        //订阅锁的释放消息成功后，进入一个不断重试获取锁的循环，循环中每次都先试着获取锁，并得到已存 
          在的锁的剩余存活时间。
            while (true) {
                // 4.重试获取锁
                ttl = tryAcquire(leaseTime, unit);
                // 5.成功获得锁
                if (ttl == null) {
                    break;
                }
                // 6.等待锁释放
                //如果锁当前是被占用的，那么等待释放锁的消息，具体实现使用了 JDK 并发的信号量工具 
                //Semaphore 来阻塞线程，当锁释放并发布释放锁的消息后，信号量的 release() 方法会被 
                //调用，此时被信号量阻塞的等待队列中的一个线程就可以继续尝试获取锁了。
                if (ttl >= 0) {
                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    getEntry(threadId).getLatch().acquire();
                }
            }
        } finally {
            // 7.取消订阅
            unsubscribe(future, threadId);
        }
    }

接着看下另一个核心方法tryAcquire()：

private Long tryAcquire(long leaseTime, TimeUnit unit) {
        // 1.将异步执行的结果以同步的形式返回
        //Redisson 实现的执行 Redis 命令都是异步的，但是它在异步的基础上提供了以同步的方式获得执行结果的封装。
        return get(tryAcquireAsync(leaseTime, unit, Thread.currentThread().getId()));
    }
    
    private <T> Future<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, long threadId) {
        if (leaseTime != -1) {
            return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
    
        // 2.用默认的锁超时时间去获取锁
        //分布式锁要确保未来的一段时间内锁一定能够被释放，因此要对锁设置超时释放的时间，在我们没有指 
        //定该时间的情况下，Redisson 默认指定为30秒。
        Future<Long> ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS,
                    TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG);
        ttlRemainingFuture.addListener(new FutureListener<Long>() {
    
            @Override
            public void operationComplete(Future<Long> future) throws Exception {
                if (!future.isSuccess()) {
                    return;
                }
                Long ttlRemaining = future.getNow();
                // 成功获得锁
                if (ttlRemaining == null) {
                    // 3.锁过期时间刷新任务调度
                    //为了避免业务中对共享资源的操作还未完成，锁就被释放掉了，需要定期（锁失效时间的    
                    //三分之一）刷新锁失效的时间，这里 Redisson 使用了 Netty 的 TimerTask、 
                    //Timeout 工具来实现该任务调度。
                    scheduleExpirationRenewal();
                }
            }
    
        });
        return ttlRemainingFuture;
    }
    
    <T> Future<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId,
                    RedisStrictCommand<T> command) {
        internalLockLeaseTime = unit.toMillis(leaseTime);
        // 3.使用 EVAL 命令执行 Lua 脚本获取锁
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                  "if (redis.call('exists', KEYS[1]) == 0) then " +
                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                      "return nil; " +
                  "end; " +
                  "return redis.call('pttl', KEYS[1]);",
                    Collections.<Object>singletonList(getName()), internalLockLeaseTime,
                            getLockName(threadId));
    }

而在加锁关键，又在于lua脚本代码（**使用lua代码执行可保证操作的原子性**）。加锁时的lua脚本说明如下：

加锁需要以下三个参数：

**KEYS\[1\] ：需要加锁的key，这里需要是字符串类型。**

**ARGV\[1\] ：锁的超时时间，防止死锁**

**ARGV\[2\] ：锁的唯一标识，id（UUID.randomUUID()） + “:” + threadId**

> //如果通过 `exists` 命令发现当前 key 不存在，即锁没被占用，则执行 `hset` 写入 Hash 类型数据 **key:全局锁名称**（例如共享资源ID）, **field:锁实例名称**（Redisson客户端ID:线程ID）, **value:1**，并执行 `pexpire` 对该 key 设置失效时间，返回空值 `nil`，至此获取锁成功。
> 
> "if (redis.call('exists', KEYS\[1\]) == 0) then " +  
>     "redis.call('hset', KEYS\[1\], ARGV\[2\], 1); " +  
>     "redis.call('pexpire', KEYS\[1\], ARGV\[1\]); " +  
>     "return nil; " +  
> "end; " +
> 
> // 如果通过 `hexists` 命令发现 Redis 中已经存在当前 key 和 field 的 Hash 数据，说明当前线程之前已经获取到锁，因为这里的锁是**可重入**的，则执行 `hincrby` 对当前 key field 的值**加一**，并重新设置失效时间，返回空值，至此重入获取锁成功。  
> "if (redis.call('hexists', KEYS\[1\], ARGV\[2\]) == 1) then " +  
>     "redis.call('hincrby', KEYS\[1\], ARGV\[2\], 1); " +  
>     //锁重入重新设置超时时间  
>     "redis.call('pexpire', KEYS\[1\], ARGV\[1\]); " +  
>     "return nil; " +  
> "end; " +
> 
>   
> // 最后是锁已被占用的情况，即当前 key 已经存在，但是 Hash 中的 Field 与当前值不同，则执行 `pttl` 获取锁的剩余存活时间并返回，至此获取锁失败。  
> "return redis.call('pttl', KEYS\[1\]);",

解锁实现方法解析：

@Override
    public void unlock() {
        // 1.通过 EVAL 和 Lua 脚本执行 Redis 命令释放锁
        Boolean opStatus = commandExecutor.evalWrite(getName(), LongCodec.INSTANCE,
                        RedisCommands.EVAL_BOOLEAN,
                        "if (redis.call('exists', KEYS[1]) == 0) then " +
                            "redis.call('publish', KEYS[2], ARGV[1]); " +
                            "return 1; " +
                        "end;" +
                        "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                            "return nil;" +
                        "end; " +
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                        "if (counter > 0) then " +
                            "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                            "return 0; " +
                        "else " +
                            "redis.call('del', KEYS[1]); " +
                            "redis.call('publish', KEYS[2], ARGV[1]); " +
                            "return 1; "+
                        "end; " +
                        "return nil;",
                        Arrays.<Object>asList(getName(), getChannelName()), 
                                LockPubSub.unlockMessage, internalLockLeaseTime, 
                                getLockName(Thread.currentThread().getId()));
        // 2.非锁的持有者释放锁时抛出异常
        if (opStatus == null) {
            throw new IllegalMonitorStateException(
                    "attempt to unlock lock, not locked by current thread by node id: "
                    + id + " thread-id: " + Thread.currentThread().getId());
        }
        // 3.释放锁后取消刷新锁失效时间的调度任务
        if (opStatus) {
            cancelExpirationRenewal();
        }
    }

解锁的lua代码说明：

**KEYS\[1\] ：需要加锁的key，这里需要是字符串类型。**

**KEYS\[2\] ：redis消息的ChannelName,一个分布式锁对应唯一的一个channelName:“redisson\_lock\_\_channel\_\_\{” + getName() + “\}”**

**ARGV\[1\] ：reids消息体，这里只需要一个字节的标记就可以，主要标记redis的key已经解锁，再结合redis的Subscribe，能唤醒其他订阅解锁消息的客户端线程申请锁。**

**ARGV\[2\] ：锁的超时时间，防止死锁**

**ARGV\[3\] ：锁的唯一标识，也就是刚才介绍的 id（UUID.randomUUID()） + “:” + threadId**

>     // key不存在，说明锁已释放，直接执行publish命令发布释放锁消息并返回1。
>     "if (redis.call('exists', KEYS[1]) == 0) then " +
>     "redis.call('publish', KEYS[2], ARGV[1]); " +
>     "return 1; " +
>     "end;" +
>     
>     //key存在，但是field在Hash中不存在，说明自己不是锁持有者，无权释放锁，返回nil。
>     "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
>     "return nil;" +
>     "end; " +
>     
>     //因为锁可重入，所以释放锁时不能把所有已获取的锁全都释放掉，一次只能释放一把锁，因此执行hincrby对锁的值减一。
>     "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
>     
>     // 如果counter>0说明锁在重入，不能删除key，则刷新锁的失效时间并返回0；
>     "if (counter > 0) then " +
>     "redis.call('pexpire', KEYS[1], ARGV[2]); " +
>     "return 0; " +
>     "else " +
>     
>     //否则刚才释放的已经是最后一把锁，则执行del命令删除锁的key，并发布锁释放消息，返回1。
>     "redis.call('del', KEYS[1]); " +
>     "redis.call('publish', KEYS[2], ARGV[1]); " +
>     "return 1; "+
>     "end; " +
>     "return nil;"

Redisson特点：

*  可以使用可重入锁，公平锁，联锁，红锁等多种方式
 *  具有WatchDog作为锁续期监听机制，当锁失效时间小于业务执行所需时间，会自动续期
 *  为防止死锁，可添加锁自动释放时间

Redisson具体使用方式请参考官方文档：[https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95][https_github.com_redisson_redisson_wiki_E7_9B_AE_E5_BD_95]

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

## 2.2 Zookeeper ##

对于zookeeper，之前很多业务是结合Dubbo作为分布式服务调用框架来设计的。在此过程中，zookeeper作为注册中心的角色负责各个服务接口的注册与调度。但其实zookeeper还有很多其他用途，比如配置管理，集群管理，队列管理以及我们今天要说到的分布式锁。（zookeeper其他用途了解参考：[https://blog.csdn.net/lingbo229/article/details/81052078][https_blog.csdn.net_lingbo229_article_details_81052078]）

zookeeper实现分布式锁的原理：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70 2][]

原理就是多个节点同时在一个指定的节点下面创建临时会话顺序节点，谁创建的节点序号最小，谁就获得了锁，并且其他节点就会监听序号比自己小的节点，一旦序号比自己小的节点被删除了，其他节点就会得到相应的事件，然后查看自己是否为序号最小的节点，如果是，则获取锁。（**zk有重试机制，一旦zk集群检测不到客户端的心跳，就会重试，Curator客户端支持多种重试策略。多次重试之后还不行的话才会删除临时节点。**）

java有集成zookeeper作为锁的实现类，可以直接使用zookeeper第三方库[Curator][]客户端，这个客户端中封装了一个可重入的锁服务。

public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
        try {
            return interProcessMutex.acquire(timeout, unit);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return true;
    }
    
    public boolean unlock() {
        try {
            interProcessMutex.release();
        } catch (Throwable e) {
            log.error(e.getMessage(), e);
        } finally {
            executorService.schedule(new Cleaner(client, path), delayTimeForClean, TimeUnit.MILLISECONDS);
        }
        return true;
    }

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

使用Zookeeper实现分布式锁的优点

*  有效的解决单点问题，不可重入问题，非阻塞问题以及锁无法释放的问题。实现起来较为简单。

使用Zookeeper实现分布式锁的缺点

*  性能上不如使用缓存实现分布式锁。 需要对ZK的原理有所了解。

具体实现可参考：[https://my.oschina.net/yangjianzhou/blog/1930493][https_my.oschina.net_yangjianzhou_blog_1930493]

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

## 2.3 Database实现分布式锁 ##

实现原理：采用数据库表的唯一键的形式。如果同一个时刻，多个线程同时向一个表中插入同样的记录，由于唯一键的原因，只能有一个线程插入成功。待业务逻辑代码执行完后，删除这条记录，来释放锁。

> CREATE TABLE \`methodLock\` (  
>   \`id\` int(11) NOT NULL AUTO\_INCREMENT COMMENT '主键',  
>   \`method\_name\` varchar(64) NOT NULL DEFAULT '' COMMENT '锁定的方法名',  
>   \`desc\` 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='锁定中的方法';

当我们想要锁住某个方法时，执行以下SQL：

> insert into methodLock(method\_name,desc) values (‘method\_name’,‘desc’)

因为我们对`method_name`做了唯一性约束，这里如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。

当方法执行完毕之后，想要释放锁的话，需要执行以下Sql:

> delete from methodLock where method\_name ='method\_name'

数据库方式实现分布式锁有如下几个问题点：

> 1、这把锁强依赖数据库的可用性，数据库是一个单点，一旦数据库挂掉，会导致业务系统不可用。
> 
> 2、这把锁没有失效时间，一旦解锁操作失败，就会导致锁记录一直在数据库中，其他线程无法再获得到锁。
> 
> 3、这把锁只能是非阻塞的，因为数据的insert操作，一旦插入失败就会直接报错。没有获得锁的线程并不会进入排队队列，要想再次获得锁就要再次触发获得锁操作。
> 
> 4、这把锁是非重入的，同一个线程在没有释放锁之前无法再次获得该锁。因为数据中数据已经存在了。

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

# 3. 分布式锁对比 #

*  zookeeper分布式锁实现简单，集群自己来保证数据一致性，但是会存在建立无用节点且多节点之间需要同步数据的问题，因此一般适合于并发量小的场景使用，例如定时任务的运行等。
 *  redis分布式锁（非redlock）由于redis自己的高性能原因，会有很好的性能，但是极端情况下会存在两个客户端获取锁（可以通过监控leader故障和运维措施来缓解和解决该问题），因此适用于高并发的场景。
 *  database分布式锁由于数据库本身的限制：性能不高且不满足高可用（即是存在备份，也会导致数据不一致），因此，工作中很难见到真正使用数据库来作为分布式锁的解决方案，这里使用数据库实现主要是为了理解分布式锁的实现原理。
 *  **从性能角度（从高到低）**
    
     *  缓存 > Zookeeper >= 数据库
 *  **从可靠性角度（从高到低）**
    
     *  Zookeeper > 缓存 > 数据库

> ### 本博客皆为学习、分享、探讨为本，欢迎各位朋友评论、点赞、收藏、关注，一起加油！ ###

![20201028144508453.gif][]

[TX-LCN]: https://blog.csdn.net/yy339452689/article/details/103871242
[Link 1]: https://blog.csdn.net/yy339452689/article/details/109314083
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200109092935677.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200109093004556.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5,size_16,color_FFFFFF,t_70
[https_github.com_redisson_redisson_wiki_E7_9B_AE_E5_BD_95]: https://github.com/redisson/redisson/wiki/%E7%9B%AE%E5%BD%95
[https_blog.csdn.net_lingbo229_article_details_81052078]: https://blog.csdn.net/lingbo229/article/details/81052078
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200108153259994.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l5MzM5NDUyNjg5,size_16,color_FFFFFF,t_70
[Curator]: https://curator.apache.org/
[https_my.oschina.net_yangjianzhou_blog_1930493]: https://my.oschina.net/yangjianzhou/blog/1930493
[20201028144508453.gif]: https://img-blog.csdnimg.cn/20201028144508453.gif