Curator中的分布式锁解读

ゞ浴缸里的玫瑰 2023-10-09 21:03 3阅读 0赞

**目录**

基本介绍

基本配置

可重入锁InterProcessMutex

不可重入锁InterProcessSemaphoreMutex

可重入读写锁InterProcessReadWriteLock

联锁InterProcessMultiLock

信号量InterProcessSemaphoreV2

栅栏barrier

倒计数器

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

#### 基本介绍 ####

Curator是netflix公司开源的一套zookeeper客户端，目前是Apache的顶级项目。与Zookeeper提供的原生客户端相比，Curator的抽象层次更高，简化了Zookeeper客户端的开发量。Curator解决了很多zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册wathcer和  
NodeExistsException 异常等。

通过查看官方文档，可以发现Curator主要解决了三类问题：

*  封装ZooKeeper client与ZooKeeper server之间的连接处理
 *  提供了一套Fluent风格的操作API
 *  提供ZooKeeper各种应用场景(recipe， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装，这些实现都遵循了zk的最佳实践，并考虑了各种极端情况

Curator由一系列的模块构成，对于一般开发者而言，常用的是curator-framework和curator-  
recipes：

*  curator-framework：提供了常见的zk相关的底层操作
 *  curator-recipes：提供了一些zk的典型使用场景的参考。本节重点关注的分布式锁就是该包提供的

#### 基本配置 ####

引入依赖：  
最新版本的curator 4.3.0支持zookeeper 3.4.x和3.5，但是需要注意curator传递进来的依赖，需要和实际服务器端使用的版本相符，以我们目前使用的zookeeper 3.4.14为例。

<dependency>
                <groupId>org.apache.zookeeper</groupId>
                <artifactId>zookeeper</artifactId>
                <version>3.4.14</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.arakhne.afc.slf4j</groupId>
                        <artifactId>slf4j-log4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>4.3.0</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.apache.zookeeper</groupId>
                        <artifactId>zookeeper</artifactId>
                    </exclusion>
                </exclusions>
          </dependency>

添加curator客户端配置：

@Configuration
    public class zkCuratorConfig {
        @Bean
        public CuratorFramework curatorFramework(){
            //后台重试，每个1000ms重试一次，重试3次
            RetryPolicy retryPolicy=new ExponentialBackoffRetry(1000,3);
            //初始化CuratorFramework客户端，如果有多哥zk地址，以逗号分割
            CuratorFramework client = CuratorFrameworkFactory.
                                    newClient("192.168.107.135", retryPolicy);
            //启动链接
            client.start();
            return client
        }
    }

#### 可重入锁InterProcessMutex ####

Reentrant和JDK的ReentrantLock类似， 意味着同一个客户端在拥有锁的同时，可以多次获取，不会被 阻塞。它是由类InterProcessMutex来实现。

// 常用构造方法
    public InterProcessMutex(CuratorFramework client, String path)
    // 获取锁
    public void acquire();
    // 带超时时间的可重入锁
    public boolean acquire(long time, TimeUnit unit);
    // 释放锁
    public void release();

改造service测试方法：

@Autowired
    
        private CuratorFramework curatorFramework;
    
        public void checkAndLock() {
            try {
                // 加锁，获取锁失败重试
                InterProcessMutex mutex = new InterProcessMutex(curatorFramework,"/curator/lock");
                mutex.acquire();
                // 先查询库存是否充足
                Stock stock = this.stockMapper.selectById(1L);
                // 再减库存
                if (stock != null && stock.getCount() > 0) {
                    stock.setCount(stock.getCount() - 1);
                    this.stockMapper.updateById(stock);
                }
                // 释放锁
                mutex.release();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

注意：如想重入，则需要使用同一个InterProcessMutex对象。

压力测试结果：

![bfa6ec23f9f84812889b7cf5950336bb.png][]

#### 不可重入锁InterProcessSemaphoreMutex ####

具体实现：InterProcessSemaphoreMutex。与InterProcessMutex调用方法类似，区别在于该锁是不 可重入的，在同一个线程中不可重入。

// 构造函数需要包含的锁的集合，或者一组ZooKeeper的path
    public InterProcessMultiLock(List<InterProcessLock> locks);
    public InterProcessMultiLock(CuratorFramework client, List<String> paths);
    // 获取锁
    public void acquire();
    public boolean acquire(long time, TimeUnit unit);
    // 释放锁
    public synchronized void release();

#### 可重入读写锁InterProcessReadWriteLock ####

类似JDK的ReentrantReadWriteLock。一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁。这也意味着写锁可以降级成读锁。从读锁升级成写锁是不成的。主要实现类  
InterProcessReadWriteLock：

// 构造方法
    public InterProcessReadWriteLock(CuratorFramework client, String basePath);
    // 获取读锁对象
    InterProcessMutex readLock();
    // 获取写锁对象
    InterProcessMutex writeLock();

#### 联锁InterProcessMultiLock ####

Multi Shared Lock是一个锁的容器。当调用acquire， 所有的锁都会被acquire，如果请求失败，所有的锁都会被release。同样调用release时所有的锁都被release(失败被忽略)。基本上，它就是组锁的代表，在它上面的请求释放操作都会传递给它包含的所有的锁。实现类InterProcessMultiLock：

#### 信号量InterProcessSemaphoreV2 ####

一个计数的信号量类似JDK的Semaphore。JDK中Semaphore维护的一组许可(permits)，而Cubator中称之为租约(Lease)。注意，所有的实例必须使用相同的numberOfLeases值。调用acquire会返回一个租约对象。客户端必须在finally中close这些租约对象，否则这些租约会丢失掉。但是，如果客户端session由于某种原因比如crash丢掉， 那么这些客户端持有的租约会自动close， 这样其它客户端可以继续使用这些租约。主要实现类InterProcessSemaphoreV2：

// 构造方法
    public InterProcessSemaphoreV2(CuratorFramework client, String path, int
    maxLeases);
    // 注意一次你可以请求多个租约，如果Semaphore当前的租约不够，则请求线程会被阻塞。
    // 同时还提供了超时的重载方法
    public Lease acquire();
    public Collection<Lease> acquire(int qty);
    public Lease acquire(long time, TimeUnit unit);
    public Collection<Lease> acquire(int qty, long time, TimeUnit unit)
    // 租约还可以通过下面的方式返还
    public void returnAll(Collection<Lease> leases);
    public void returnLease(Lease lease);

#### 栅栏barrier ####

1. DistributedBarrier构造函数中barrierPath参数用来确定一个栅栏，只要barrierPath参数相同  
(路径相同)就是同一个栅栏。通常情况下栅栏的使用如下：

*  1. 主client设置一个栅栏
 *  2. 其他客户端就会调用waitOnBarrier()等待栅栏移除，程序处理线程阻塞
 *  3. 主client移除栅栏，其他客户端的处理程序就会同时继续运行。

DistributedBarrier类的主要方法如下：

setBarrier() - 设置栅栏
    waitOnBarrier() - 等待栅栏移除
    removeBarrier() - 移除栅栏

2. DistributedDoubleBarrier双栅栏，允许客户端在计算的开始和结束时同步。当足够的进程加入到  
双栅栏时，进程开始计算，当计算完成时，离开栅栏。DistributedDoubleBarrier实现了双栅栏的  
功能。构造函数如下：

// client - the client
    // barrierPath - path to use
    // memberQty - the number of members in the barrier
    public DistributedDoubleBarrier(CuratorFramework client, String
    barrierPath, int memberQty);
    enter()、enter(long maxWait, TimeUnit unit) - 等待同时进入栅栏
    leave()、leave(long maxWait, TimeUnit unit) - 等待同时离开栅栏

memberQty是成员数量，当enter方法被调用时，成员被阻塞，直到所有的成员都调用了enter。  
当leave方法被调用时，它也阻塞调用线程，直到所有的成员都调用了leave。  
注意：参数memberQty的值只是一个阈值，而不是一个限制值。当等待栅栏的数量大于或等于这  
个值栅栏就会打开！  
与栅栏(DistributedBarrier)一样,双栅栏的barrierPath参数也是用来确定是否是同一个栅栏的，双  
栅栏的使用情况如下：

*  1. 从多个客户端在同一个路径上创建双栅栏(DistributedDoubleBarrier),然后调用enter()方
 *  法，等待栅栏数量达到memberQty时就可以进入栅栏。
 *  2. 栅栏数量达到memberQty，多个客户端同时停止阻塞继续运行，直到执行leave()方法，等待memberQty个数量的栅栏同时阻塞到leave()方法中。
 *  3. memberQty个数量的栅栏同时阻塞到leave()方法中，多个客户端的leave()方法停止阻塞，继续运行。

#### 倒计数器 ####

利用ZooKeeper可以实现一个集群共享的计数器。只要使用相同的path就可以得到最新的计数器值，这是由ZooKeeper的一致性保证的。Curator有两个计数器， 一个是用int来计数，一个用long来计数。SharedCount这个类使用int类型来计数。主要涉及三个类。

* SharedCount
    * SharedCountReader
    * SharedCountListener

SharedCount代表计数器， 可以为它增加一个SharedCountListener，当计数器改变时此Listener可以监听到改变的事件，而SharedCountReader可以读取到最新的值， 包括字面值和带版本信息的值VersionedValue。

DistributedAtomicLong  
除了计数的范围比SharedCount大了之外， 它首先尝试使用乐观锁的方式设置计数器， 如果不成功(比如期间计数器已经被其它client更新了)， 它使用InterProcessMutex方式来更新计数值。此计数器有一系列的操作：  
get(): 获取当前值  
increment()：加一  
decrement(): 减一  
add()：增加特定的值  
subtract(): 减去特定的值  
trySet(): 尝试设置计数值  
forceSet(): 强制设置计数值  
你必须检查返回结果的succeeded()， 它代表此操作是否成功。如果操作成功， preValue()代表操作前的值， postValue()代表操作后的值。

[bfa6ec23f9f84812889b7cf5950336bb.png]: https://img-blog.csdnimg.cn/bfa6ec23f9f84812889b7cf5950336bb.png