springboot整合curator实现分布式锁过程_Java教程

springboot curator实现分布式锁

理论篇：

Curator是Netflix开源的一套ZooKeeper客户端框架. Netflix在使用ZooKeeper的过程中发现ZooKeeper自带的客户端太底层, 应用方在使用的时候需要自己处理很多事情, 于是在它的基础上包装了一下, 提供了一套更好用的客户端框架. Netflix在用ZooKeeper的过程中遇到的问题, 我们也遇到了, 所以开始研究一下, 首先从他在github上的源码, wiki文档以及Netflix的技术blog入手.

看完官方的文档之后, 发现Curator主要解决了三类问题：

封装ZooKeeper client与ZooKeeper server之间的连接处理;
提供了一套Fluent风格的操作API;
提供ZooKeeper各种应用场景(recipe, 比如共享锁服务, 集群领导选举机制)的抽象封装.

Curator列举的ZooKeeper使用过程中的几个问题

初始化连接的问题: 在client与server之间握手建立连接的过程中, 如果握手失败, 执行所有的同步方法(比如create, getData等)将抛出异常
自动恢复(failover)的问题: 当client与一台server的连接丢失,并试图去连接另外一台server时, client将回到初始连接模式
session过期的问题: 在极端情况下, 出现ZooKeeper session过期, 客户端需要自己去监听该状态并重新创建ZooKeeper实例 .
对可恢复异常的处理:当在server端创建一个有序ZNode, 而在将节点名返回给客户端时崩溃, 此时client端抛出可恢复的异常, 用户需要自己捕获这些异常并进行重试
使用场景的问题:Zookeeper提供了一些标准的使用场景支持, 但是ZooKeeper对这些功能的使用说明文档很少, 而且很容易用错. 在一些极端场景下如何处理, zk并没有给出详细的文档说明. 比如共享锁服务, 当服务器端创建临时顺序节点成功, 但是在客户端接收到节点名之前挂掉了, 如果不能很好的处理这种情况, 将导致死锁.

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

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

Curator通过以上的处理, 让用户专注于自身的业务本身, 而无需花费更多的精力在zk本身.

实操篇：

CuratorFrameworkFactory类提供了两个方法, 一个工厂方法newClient, 一个构建方法build. 使用工厂方法newClient可以创建一个默认的实例, 而build构建方法可以对实例进行定制. 当CuratorFramework实例构建完成, 紧接着调用start()方法, 在应用结束的时候, 需要调用close()方法. CuratorFramework是线程安全的. 在一个应用中可以共享同一个zk集群的CuratorFramework.

核心对象CuratorFramework的创建如下：

				?

									RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000,3);

									CuratorFramework client = CuratorFrameworkFactory.builder()

									                                      .connectString("")

									                                      .sessionTimeoutMs(5000)

									                                      .connectionTimeoutMs(5000)

									                                      .retryPolicy(retryPolicy)

									                                      .build();

									client.start();

需要使用分布式锁的地方，代码如下：

				?

									String lockOn= "test";

									InterProcessMutex mutex = new InterProcessMutex(curatorFramework,lockOn);

									boolean locked =mutex.acquire(0,TimeUnit.SECONDS);

									//finally部分  

									mutex.release();

分布式锁常用于定时任务，使用自定义注解，使用spring aspect around，在真正的代码执行之前尝试获取锁，获取不到直接退出，获取到锁的，执行具体业务，代码如下：

				?

									@Aspect

									public class DistributedLockAspect{

									    @Pointcut("@annotation(com.**.**.DistributedLock")

									    public void methodAspect(){};  

									    @Around("methodAspect()")

									    public Object execute(ProceedingJoinPoint joinPoint) throws Exception{

									    String lockPath = "/opt/zookeeper/lock";

									    InterProcessMutex mutex = new InterProcessMutex(cruatorFramework,lockPath);

									    try{

									       boolean locked = mutex.acquire(0,TimeUnit.SECONDS);

									       if(!locked){

									          return null;

									      }else{

									        return joinPoint.proceed();

									      }

									   }catch(Exception e){

									       e.printStackTrace();

									   }finally{

									       mutex.release();

									   }

									 }

									}

自定义注解：

				?

									 @Target(ElementType.METHOD)

									 @Retention(RetentionPolicy.RUNTIME)

									 public @interface DistributedLock{

									    String lockPath();  

									 }

注意事项：

1.CuratorFramework对象建议在应用中做单例处理，在具体使用处注入使用, 并在应用结束前销毁，代码如下：

				?

									@Configration

									public class CuratorConfigration{

									    @Bean    

									    public CuratorFramework initCuratorFramework(){

									        //忽略 

									       // 参照前面 CuratorFramework 对象创建部分

									    }    

									}

2.在aspect部分将curatorFramework对象进行关闭

				?

									@PreDestroy

									public void destroy(){

									   CloseableUtils.closeQuietly(curatorFramework);

									}

项目实际应用中分布式锁介绍

锁的介绍

1、悲观锁

顾名思义，很悲观，就是每次拿数据的时候都认为别的线程会修改数据，所以在每次拿的时候都会给数据上锁。上锁之后，当别的线程想要拿数据时，就会阻塞，直到给数据上锁的线程将事务提交或者回滚。传统的关系型数据库里就用到了很多这种锁机制，比如行锁，表锁，共享锁，排他锁等，都是在做操作之前先上锁。

2、行锁

通过select for update语句给sid = 1的数据行上了锁

3、表锁

				?

									select * from student for update;

4、页锁

行锁锁指定行，表锁锁整张表，页锁是折中实现，即一次锁定相邻的一组记录。

5、共享锁

共享锁又称为读锁，一个线程给数据加上共享锁后，其他线程只能读数据，不能修改。

6、排他锁

排他锁又称为写锁，和共享锁的区别在于，其他线程既不能读也不能修改。

7、乐观锁

乐观锁其实不会上锁。顾名思义，很乐观，它默认别的线程不会修改数据，所以不会上锁。只是在更新前去判断别的线程在此期间有没有修改数据，如果修改了，会交给业务层去处理。

目前几乎很多大型网站及应用都是分布式部署的，分布式场景中的数据一致性问题一直是一个比较重要的话题。分布式的CAP理论告诉我们“任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。”所以，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证“最终一致性”，只要这个最终时间是在用户可以接受的范围内即可。
在很多场景中，我们为了保证数据的最终一致性，需要很多的技术方案来支持，比如分布式事务、分布式锁等。有的时候，我们需要保证一个方法在同一时间内只能被同一个线程执行。在单机环境中，Java中其实提供了很多并发处理相关的API，但是这些API在分布式场景中就无能为力了。也就是说单纯的Java Api并不能提供分布式锁的能力。所以针对分布式锁的实现目前有多种方案：

1、基于数据库实现分布式锁

2、基于缓存（redis，memcached）实现分布式锁

3、基于Zookeeper实现分布式锁

4、在分析这几种实现方案之前我们先来想一下，我们需要的分布式锁应该是怎么样的？（这里以方法锁为例，资源锁同理）

可以保证在分布式部署的应用集群中，同一个方法在同一时间只能被一台机器上的一个线程执行。

这把锁要是一把可重入锁（避免死锁）
这把锁最好是一把阻塞锁（根据业务需求考虑要不要这条）
有高可用的获取锁和释放锁功能
获取锁和释放锁的性能要好

悲观锁-数据库锁

借助数据中自带的锁来实现分布式的锁

				?

									public boolean lock(){

									    connection.setAutoCommit(false)

									    while(true){

									        try{

									            result = select * from methodLock where method_name=xxx for update;

									            if(result==null){

									                return true;

									            }

									        }catch(Exception e){

									        }

									        sleep(1000);

									    }

									    return false;

									}

在查询语句后面增加for update，数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。

我们可以认为获得排它锁的线程即可获得分布式锁，当获取到锁之后，可以执行方法的业务逻辑，执行完方法之后，再通过以下方法解锁：

				?

									public void unlock(){

									    connection.commit();

									}

通过connection.commit()操作来释放锁。

这种方法可以有效的解决上面提到的无法释放锁和阻塞锁的问题。

阻塞锁，for update语句会在执行成功后立即返回，在执行失败时一直处于阻塞状态，直到成功。

锁定之后服务宕机，无法释放，使用这种方式，服务宕机之后数据库会自己把锁释放掉。

但是还是无法直接解决数据库单点和可重入问题。

悲观锁-缓存锁

相比较于基于数据库实现分布式锁的方案来说，基于缓存来实现在性能方面会表现的更好一点。而且很多缓存是可以集群部署的，可以解决单点问题。

redis2.6之后，SET命令支持超时和key存在检查，这是一个原子操作

springboot整合curator实现分布式锁过程

缓存锁优势是性能出色，劣势就是由于数据在内存中，一旦缓存服务宕机，锁数据就丢失了。像redis自带复制功能，可以对数据可靠性有一定的保证，但是由于复制也是异步完成的，因此依然可能出现master节点写入锁数据而未同步到slave节点的时候宕机，锁数据丢失问题。

分布式锁—zookeeper

基于zookeeper临时有序节点可以实现的分布式锁。大致思想即为：每个客户端对某个方法加锁时，在zookeeper上的与该方法对应的指定节点的目录下，生成一个唯一的瞬时有序节点。判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

来看下Zookeeper能不能解决前面提到的问题。

锁无法释放:使用Zookeeper可以有效的解决锁无法释放的问题，因为在创建锁的时候，客户端会在ZK中创建一个临时节点，一旦客户端获取到锁之后突然挂掉（Session连接断开），那么这个临时节点就会自动删除掉。其他客户端就可以再次获得锁。
非阻塞锁:使用Zookeeper可以实现阻塞的锁，客户端可以通过在ZK中创建顺序节点，并且在节点上绑定监听器，一旦节点有变化，Zookeeper会通知客户端，客户端可以检查自己创建的节点是不是当前所有节点中序号最小的，如果是，那么自己就获取到锁，便可以执行业务逻辑了。
不可重入:使用Zookeeper也可以有效的解决不可重入的问题，客户端在创建节点的时候，把当前客户端的主机信息和线程信息直接写入到节点中，下次想要获取锁的时候和当前最小的节点中的数据比对一下就可以了。如果和自己的信息一样，那么自己直接获取到锁，如果不一样就再创建一个临时的顺序节点，参与排队。
单点问题:使用Zookeeper可以有效的解决单点问题，ZK是集群部署的，只要集群中有半数以上的机器存活，就可以对外提供服务。

以上为个人经验，希望能给大家一个参考，也希望大家多多支持服务器之家。

原文链接：https://www.cnblogs.com/xifenglou/p/10455553.html