文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Curator实现分布式锁(可重入 不可重入 读写 联锁 信号量 栅栏 计数器)

2023-08-19 11:59

关注

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

Curator主要解决了三类问题:

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

curator 4.3.0支持zookeeper 3.4.x3.5,但是需要注意curator传递进来的依赖,需要和实际服务器端使用的版本相符,以使用zookeeper 3.4.14为例。

    org.apache.curator    curator-framework    4.3.0                        org.apache.zookeeper            zookeeper                org.apache.curator    curator-recipes    4.3.0                        org.apache.zookeeper            zookeeper                org.apache.zookeeper    zookeeper    3.4.14

1. 配置

添加curator客户端配置:

@Configurationpublic class CuratorConfig {    @Bean    public CuratorFramework curatorFramework(){        // 重试策略,这里使用的是指数补偿重试策略,重试3次,初始重试间隔1000ms,每次重试之后重试间隔递增。        RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);        // 初始化Curator客户端:指定链接信息 及 重试策略        CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.1.111:2181", retry);        client.start(); // 开始链接,如果不调用该方法,很多方法无法工作        return client;    }}

2. 可重入锁InterProcessMutex

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

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

测试方法:

@Autowiredprivate CuratorFramework curatorFramework;public void checkAndLock() {     InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "/curator/lock");    try {        // 加锁        mutex.acquire();        // 处理业务        // 例如查询库存 扣减库存                // this.testSub(mutex); 如想重入,则需要使用同一个InterProcessMutex对象        // 释放锁        mutex.release();    } catch (Exception e) {        e.printStackTrace();    }}public void testSub(InterProcessMutex mutex) {    try {        mutex.acquire();    System.out.println("测试可重入锁。。。。");        mutex.release();    } catch (Exception e) {        e.printStackTrace();    }}

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

3. 不可重入锁InterProcessSemaphoreMutex

具体实现:InterProcessSemaphoreMutexInterProcessMutex调用方法类似,区别在于该锁是不可重入的,在同一个线程中不可重入。

public InterProcessSemaphoreMutex(CuratorFramework client, String path);public void acquire();public boolean acquire(long time, TimeUnit unit);public void release();

案例:

@Autowiredprivate CuratorFramework curatorFramework;public void deduct() {    InterProcessSemaphoreMutex mutex = new InterProcessSemaphoreMutex(curatorFramework, "/curator/lock");    try {        mutex.acquire();        // 处理业务        // 例如查询库存 扣减库存    } catch (Exception e) {        e.printStackTrace();    } finally {        try {            mutex.release();        } catch (Exception e) {            e.printStackTrace();        }    }}

4. 可重入读写锁InterProcessReadWriteLock

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

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

注意:写锁在释放之前会一直阻塞请求线程,而读锁不会

public void testZkReadLock() {    try {        InterProcessReadWriteLock rwlock = new InterProcessReadWriteLock(curatorFramework, "/curator/rwlock");        rwlock.readLock().acquire(10, TimeUnit.SECONDS);        // TODO:一顿读的操作。。。。        //rwlock.readLock().unlock();    } catch (Exception e) {        e.printStackTrace();    }}public void testZkWriteLock() {    try {        InterProcessReadWriteLock rwlock = new InterProcessReadWriteLock(curatorFramework, "/curator/rwlock");        rwlock.writeLock().acquire(10, TimeUnit.SECONDS);        // TODO:一顿写的操作。。。。        //rwlock.writeLock().unlock();    } catch (Exception e) {        e.printStackTrace();    }}

5. 联锁InterProcessMultiLock

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

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

6. 信号量InterProcessSemaphoreV2

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

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

案例代码:

StockController中添加方法:

@GetMapping("test/semaphore")public String testSemaphore(){    this.stockService.testSemaphore();    return "hello Semaphore";}

StockService中添加方法:

public void testSemaphore() {    // 设置资源量 限流的线程数    InterProcessSemaphoreV2 semaphoreV2 = new InterProcessSemaphoreV2(curatorFramework, "/locks/semaphore", 5);    try {        Lease acquire = semaphoreV2.acquire();// 获取资源,获取资源成功的线程可以继续处理业务操作。否则会被阻塞住        this.redisTemplate.opsForList().rightPush("log", "10010获取了资源,开始处理业务逻辑。" + Thread.currentThread().getName());        TimeUnit.SECONDS.sleep(10 + new Random().nextInt(10));        this.redisTemplate.opsForList().rightPush("log", "10010处理完业务逻辑,释放资源=====================" + Thread.currentThread().getName());        semaphoreV2.returnLease(acquire); // 手动释放资源,后续请求线程就可以获取该资源    } catch (Exception e) {        e.printStackTrace();    }}

7. 栅栏barrier

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

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

DistributedBarrier类的主要方法如下:

setBarrier() - 设置栅栏waitOnBarrier() - 等待栅栏移除removeBarrier() - 移除栅栏
  1. DistributedDoubleBarrier双栅栏,允许客户端在计算的开始和结束时同步。当足够的进程加入到双栅栏时,进程开始计算,当计算完成时,离开栅栏。DistributedDoubleBarrier实现了双栅栏的功能。构造函数如下:

    // client - the client// barrierPath - path to use// memberQty - the number of members in the barrierpublic 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()方法停止阻塞,继续运行。

8. 共享计数器

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

8.1. SharedCount

共享计数器SharedCount相关方法如下:

// 构造方法public SharedCount(CuratorFramework client, String path, int seedValue);// 获取共享计数的值public int getCount();// 设置共享计数的值public void setCount(int newCount) throws Exception;// 当版本号没有变化时,才会更新共享变量的值public boolean  trySetCount(VersionedValue previous, int newCount);// 通过监听器监听共享计数的变化public void addListener(SharedCountListener listener);public void addListener(final SharedCountListener listener, Executor executor);// 共享计数在使用之前必须开启public void start() throws Exception;// 关闭共享计数public void close() throws IOException;

使用案例:

StockController:

@GetMapping("test/zk/share/count")public String testZkShareCount(){    this.stockService.testZkShareCount();    return "hello shareData";}

StockService:

public void testZkShareCount() {    try {        // 第三个参数是共享计数的初始值        SharedCount sharedCount = new SharedCount(curatorFramework, "/curator/count", 0);        // 启动共享计数器        sharedCount.start();        // 获取共享计数的值        int count = sharedCount.getCount();        // 修改共享计数的值        int random = new Random().nextInt(1000);        sharedCount.setCount(random);        System.out.println("我获取了共享计数的初始值:" + count + ",并把计数器的值改为:" + random);        sharedCount.close();    } catch (Exception e) {        e.printStackTrace();    }}

8.2. DistributedAtomicNumber

DistributedAtomicNumber接口是分布式原子数值类型的抽象,定义了分布式原子数值类型需要提供的方法。

DistributedAtomicNumber接口有两个实现:DistributedAtomicLongDistributedAtomicInteger

在这里插入图片描述

这两个实现将各种原子操作的执行委托给了DistributedAtomicValue,所以这两种实现是类似的,只不过表示的数值类型不同而已。这里以DistributedAtomicLong 为例进行演示

DistributedAtomicLong除了计数的范围比SharedCount大了之外,比SharedCount更简单易用。它首先尝试使用乐观锁的方式设置计数器, 如果不成功(比如期间计数器已经被其它client更新了), 它使用InterProcessMutex方式来更新计数值。此计数器有一系列的操作:

最后必须检查返回结果的succeeded(), 代表此操作是否成功。如果操作成功, preValue()代表操作前的值, postValue()代表操作后的值。

来源地址:https://blog.csdn.net/weixin_43847283/article/details/128603255

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-服务器
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯