分布式锁3: zk实现分布式锁5 使用中间件curator-interprocessmutex可重入锁

news2025/1/13 3:09:13

一 curator的说明

1.1 curator的说明

curator是netflix公司开源的一个zk客户端。对Zookeeper提供的原生客户端进行封装，简化了Zookeeper客户端的开发量。Curator解决了很多zookeeper客户端非常底层的细节开发工作，包括连接重连、反复注册wathcer和NodeExistsException异常等。

1.2 curator的特点

1.封装ZooKeeper client与ZooKeeper server之间的连接处理

2.提供了一套Fluent风格的操作API

3.提供ZooKeeper各种应用场景(recipe,比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装，这些实现都遵循了zk的最佳实践，并考虑了各种极端情况。

1.3 curator其他功能锁（了解）

1.InterProcessSemaphoreMutex 不可重入锁，在同一个线程中不可重入。

2.InterProcessReadWriteLock 读写锁，一个拥有写锁的线程可重入读锁，但是读锁却不能进入写锁。这也意味着写锁可以降级成读锁。从读锁升级成写锁是不成的。

3.Multi Shared Lock是一个锁的容器。当调用acquire，所有的锁都会被acquire，如果请求失败，所有的锁都会被release。

4.InterProcessSemaphoreV2：一个计数的信号量

5.DistributedBarrier构造函数中barrierPath参数用来确定一个栅栏，只要barrierPath参数相同(路径相同)就是同一个栅栏。

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

二 curator的InterProcessMutex可重入锁

2.1 原理

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

2.2 常用API

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

2.3 代码实现

2.3.1.pom文件sdk引入

  <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>5.3.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>5.3.0</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.zookeeper</groupId>
                    <artifactId>zookeeper</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

2.3.2 初始化客户端

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

2.3.3 分布式锁代码

 @Autowired
    private CuratorFramework curatorFramework;
    public   void checkAndLocksByCurator() throws KeeperException, InterruptedException {
        InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "/curator/lock");
        try {
            // 加锁
            mutex.acquire();
            // 先查询库存是否充足
            Stock stock = this.stockMapper.selectById(1L);
            // 再减库存
            if (stock != null && stock.getCount() > 0){
                stock.setCount(stock.getCount() - 1);
                this.stockMapper.updateById(stock);
            }
             this.testSub(mutex);
            // 释放锁
            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();
        }
    }