curator实现分布式锁

叁歲伎倆 2022-12-24 13:52 157阅读 0赞

## 基于zookeeper的Curator ##

Apache Curator是一个比较完善的ZooKeeper客户端框架，通过封装的一套高级API 简化了ZooKeeper的操作。通过查看官方文档，可以发现Curator主要解决了三类问题：

*  封装ZooKeeper client与ZooKeeper server之间的连接处理
 *  提供了一套Fluent风格的操作API
 *  提供ZooKeeper各种应用场景(recipe， 比如：分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等)的抽象封装。

Curator主要从一下几个方面降低了zk使用的复杂性：

*  重试机制:提供可插拔的重试机制, 它将给捕获所有可恢复的异常配置一个重试策略，并且内部也提供了几种标准的重试策略(比如指数补偿)
 *  连接状态监控: Curator初始化之后会一直对zk连接进行监听，一旦发现连接状态发生变化将会作出相应的处理
 *  zk客户端实例管理:Curator会对zk客户端到server集群的连接进行管理，并在需要的时候重建zk实例，保证与zk集群连接的可靠性
 *  各种使用场景支持:Curator实现了zk支持的大部分使用场景（甚至包括zk自身不支持的场景），这些实现都遵循了zk的最佳实践，并考虑了各种极端情况。

## Zookeeper实现分布式锁的机制 ##

1.  使用zk的临时节点和有序节点，每个线程获取锁就是在zk创建一个**临时有序的节点**，比如在/lock/目录下。
2.  创建节点成功后，获取/lock目录下的所有临时节点，再判断当前线程创建的节点是否是所有的节点的序号最小的节点。
3.  如果当前线程创建的节点是所有节点序号最小的节点，则认为获取锁成功。
4.  比如当前线程获取到的节点序号为/lock/003,然后所有的节点列表为\[/lock/001,/lock/002,/lock/003\],则对/lock/002这个节点添加一个事件监听器。
5.  如果锁释放了，会唤醒下一个序号的节点，然后重新执行第3步，判断是否自己的节点序号是最小。  
    比如/lock/001释放了，/lock/002监听到时间，此时节点集合为\[/lock/002,/lock/003\],则/lock/002为最小序号节点，获取到锁。

> 临时顺序节点与临时节点不同的是产生的节点是有序的，我们可以利用这一特点，只让当前线程监听上一序号的线程，每次获取锁的时候判断自己的序号是否为最小，最小即获取到锁，执行完毕就删除当前节点继续判断谁为最小序号的节点。  
> 当线程数庞大时会发生“惊群”现象，zookeeper节点可能会运行缓慢甚至宕机。这是因为其他线程没获取到锁时都会监听/lockPath节点，当A线程释放完毕，海量的线程都同时停止阻塞，去争抢锁，这种操作十分耗费资源，且性能大打折扣。

## Curator实现分布式锁代码 ##

application.yml配置zk连接参数：

# zk配置
    curator:
      retryCount: 5      #重试次数
      baseSleepTimeMs: 1000   #重试间隔时间
      connectString: 192.168.8.23:2181,192.168.8.24:2181,192.168.8.25:2181    # zk地址
      sessionTimeoutMs: 60000          #session超时时间
      connectionTimeoutMs: 5000         #连接超时时间

获取配置文件参数的实体类：

@Data
    @Component
    @ConfigurationProperties(prefix = "curator")
    public class WrapperZk {
    
        private int retryCount;
    
        private int elapsedTimeMs;
    
        private String connectString;
    
        private int sessionTimeoutMs;
    
        private int connectionTimeoutMs;
    
        private int baseSleepTimeMs;
    }

Curator配置类：

@Configuration
    @Slf4j
    public class ZkConfiguration {
        @Autowired
        WrapperZk wrapperZk;
    
        @Bean(initMethod = "start")
        public CuratorFramework curatorFramework(){
            RetryPolicy retrYPolicy = new ExponentialBackoffRetry(wrapperZk.getBaseSleepTimeMs(), wrapperZk.getRetryCount());
            CuratorFramework client = CuratorFrameworkFactory.newClient(wrapperZk.getConnectString(),retrYPolicy);
            log.info("zk curator初始化完成...");
            return client;
        }
    }

> 其中RetryPolicy为重试策略，第一个参数为baseSleepTimeMs初始的sleep时间，用于计算之后的每次重试的sleep时间。第二个参数为maxRetries，最大重试次数。

服务端代码：

@Controller
    @RequestMapping("/testzk")
    @Slf4j
    public class MicroController {
        
        private final static String PATH = "/rootlock02";
        
        @Autowired
        ZkConfiguration zkConfiguration;
    
        @RequestMapping("/lock1")
        @ResponseBody
        public String getLock1() throws Exception {
            InterProcessMutex lock = new InterProcessMutex(zkConfiguration.curatorFramework(), PATH);
            for (int i = 0; i < 3; i++) {
                Thread thread = new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            log.info(Thread.currentThread().getName() + "尝试获取锁....");
                            lock.acquire();
                            log.info(Thread.currentThread().getName() + "获取锁成功....");
                            log.info(Thread.currentThread().getName() + "开始执行业务逻辑....");
                            Thread.sleep(10000);
                            lock.release();
                            log.info(Thread.currentThread().getName() + "释放锁成功....");
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                });
                thread.start();
            }
            return "";
        }
    }

> 这里我用的是最常用的可重入排他锁，也是公平锁（InterProcessMutex）  
> InterProcessMutex：分布式可重入排它锁  
> InterProcessSemaphoreMutex：分布式排它锁  
> InterProcessReadWriteLock：分布式读写锁  
> InterProcessMultiLock：将多个锁作为单个实体管理的容器