【zookeeper】Apache curator的使用及zk分布式锁实现

た入场券 2021-09-29 14:36 449阅读 0赞

上篇,本篇主要讲Apache开源的curator的使用,有了curator,利用Java对zookeeper的操作变得极度便捷.

其实在学之前我也有个疑虑,我为啥要学curator,撇开涨薪这些外在的东西,就单技术层面来讲,学curator能帮我做些什么?这就不得不从zookeeper说起,上篇我已经大篇幅讲了zk是做什么的了,但真正要靠zk去实现多服务器自动拉取更新的配置文件等功能是非常难的,如果没有curator,直接去写的话基本上能把你累哭,就好比连Mybatis或者jpa都没有,让你用原生的代码去写个网站一样,你可以把curator当做一个比较强大的工具,有了它操作zk不再是事,说这么多,是时候进入正题了:

curator 官网:http://curator.apache.org

使用curator去实现的几块内容:

学习目录:  
1.使用curator建立与zk的连接  
2.使用curator添加/递归添加节点  
3.使用curator删除/递归删除节点  
4.使用curator创建/验证 ACL(访问权限列表)  
5.使用curator监听 单个/父 节点的变化(watch事件)  
\---------------------------------------------  
6.基于curator实现zookeeper分布式锁(需要掌握基本的多线程知识)  
  
前置条件:已掌握zookeeper的基本操作,对zookeeper有所了解,如果没有掌握请翻阅我前面的博客去学习.  
本节所需要引入的依赖有以下三个,建议直接全部引入即可:

<dependency>  
<groupId>org.apache.curator</groupId>  
<artifactId>curator-framework</artifactId>  
<version>4.0.0</version>  
</dependency>  
<dependency>  
<groupId>org.apache.zookeeper</groupId>  
<artifactId>zookeeper</artifactId>  
<version>3.4.12</version>  
</dependency>  
<dependency>  
<groupId>org.apache.curator</groupId>  
<artifactId>curator-recipes</artifactId>  
<version>4.0.0</version>  
</dependency>  
1.通过curator建立与zk的连接

需要准备连接zk的url,建议直接写成工具类,因为接下来会频繁用到,功能类似于jdbc.

public class ZkClientUtil \{  
private static final int BASE\_SLEEP\_TIME\_MS = 5000; //定义失败重试间隔时间 单位:毫秒  
private static final int MAX\_RETRIES = 3; //定义失败重试次数  
private static final int SESSION\_TIME\_OUT = 1000000; //定义会话存活时间,根据业务灵活指定 单位:毫秒  
private static final String ZK\_URI = "192.168.174.132:2181";//你自己的zkurl和端口号  
private static final String NAMESPACE = "laohan\_jianshen";  
//工作空间,可以不指定,建议指定,功能类似于项目包,之后创建的所有的节点都会在该工作空间下,方便管理  
  
public static CuratorFramework build()\{  
//创建比较简单,链式编程,很爽,基本上指定点参数就OK了  
RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASE\_SLEEP\_TIME\_MS,MAX\_RETRIES);//重试策略  
CuratorFramework client = CuratorFrameworkFactory  
.builder()  
.connectString(ZK\_URI)  
.retryPolicy(retryPolicy)  
.namespace(NAMESPACE)  
.sessionTimeoutMs(SESSION\_TIME\_OUT)  
.build();  
return client;  
\}  
\}  
2.通过curator添加/递归添加节点

//通过上一步获取到的client,直接启动该client,值得注意的是client在使用前必须先启动:  
client.start;  
client  
.create()//创建节点  
.withMode(CreateMode.xxx)//节点属性:永久节点/临时节点/有序节点 通过CreateMode.即可看到  
.withACL(ZooDefs.Ids.xxx)//节点访问权限,通过Ids.即可看到 默认是OPEN\_ACL\_UNSAFE(开放不安全权限)  
.forPath("/yourpath","yourdata".getBytes());//指明你的节点路径,数据可以不指定,数据必须是byte\[\]  
创建递归节点:

//比如我想一次性创建/yourpath/a/b/c/1/2/3...这样的节点,如果按传统方法会累死你  
//curator可以一次性创建好,只需要在创建时添加creatingParentsIfNeeded即可.  
client  
.create()//创建节点  
.creatingParentsIfNeeded()//创建父节点,如果需要的话  
  
...  
3.使用curator删除/递归删除节点

client  
.delete() //删除  
.guaranteed()//保证一定帮你删了它  
.withVersion(0)//指定要删节点的版本号  
.forPath("/yourpath")//指定要删节点的路径  
递归删除:

//比如我当前的节点结构是这样:/yourpath/a/b/c/1/2/3 我想删除a节点下面的所有目录  
//传统方法累死个人,现在只需要添加deletingChildrenIfNeeded即可  
client  
.delete() //删除  
.deletingChildrenIfNeeded()//如果它有儿子都给删了...  
4.使用curator创建/验证 ACL(访问权限列表)

//为了保证安全,有时需要对节点的访问权限做一些限制,否则可能会引起重要信息泄露/篡改/删除等  
//节点ACL的创建方式有两种,一种是使用ZK提供的,一种是自定义的  
//1.ZK提供的,比较简单,拿来即用,在创建节点时指明withACL即可  
client  
.create()  
.withACL(ZooDefs.Ids.READ\_ACL\_UNSAFE)//指明该节点是只读节点,还有其他属性,可以通过Ids.查看  
//创建自定义ACL,需要自己new Id(),并指明是否是加密的,然后账号和密码是多少,加密策略使用zk提供的:  
List<ACL> aclList = new ArrayList<ACL>();  
ACL acl1 = new ACL(ZooDefs.Perms.READ,new Id("digest",DigestAuthenticationProvider.generateDigest("user:123456")));  
ACL acl2 = new ACL(ZooDefs.Perms.ALL,new Id("digest",DigestAuthenticationProvider.generateDigest("root:123456")));  
aclList.add(acl1);  
aclList.add(acl2);  
//如此我就创建好了两种不同的权限账号,user只能对该节点有读的权限,但root用户对该节点有所有权限  
//ACL验证,创建好节点之后,可以在服务器的zk安装目录的bin目录下 连接客户端./zkCli  
//然后通过ls /该目录 查看是否可以访问 正常是不能访问的 会提示权限不够  
//下面我们通过curator去连接,要想访问该节点需要在创建client时就指明账号和密码:  
CuratorFramework client = CuratorFrameworkFactory  
.builder()  
.authorization("digest","root:123456".getBytes())//指明使用了加密,用户名和密码用:隔开,以byte\[\]输入  
//如此,接下来通过该client可以对刚刚创建的节点具有所有权限,如果登录的是user,则只具有读权限.  
5.通过curator创建单个节点及其父节点的watch事件

由于zk的watch事件是只能被触发一次的,触发完即销毁监听,这显然不是我们想要的,在实际开发中更多的场景是需要对某个节点持续监听,所以这里我只介绍创建持续监听的单节点/父节点

//对单个节点创建watch事件  
//定义NodeCache,指明被监听节点的路径:  
final NodeCache nodeCache = new NodeCache(client,"/yourpath");  
nodeCache.start(true);//开启  
nodeCache  
.getCurrentData()//可以获取该监听节点的数据  
.getPath();//可以获取该监听节点的路径  
  
//对指定父节点创建watch事件,只要其任何一个子节点,或子节点的子节点...发生变化,就会触发watch事件.  
//定义PathChildrenCache,指明要watch的目录  
final PathChildrenCache pathChildrenCache = new PathChildrenCache(client,"yourpath");  
//启动,启动策略有三种:同步,异步提交,异步 用的比较多的就是下面这种,用StartMode.可以查看到  
pathChildrenCache.start(PathChildrenCache.StartMode.POST\_INITIALIZED\_EVENT);  
//对该节点创建监听器  
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() \{  
public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception \{  
//TODO 可以通过PathChildrenCacheEvent.拿到你想要的数据和路径等  
\}  
\});  
至此,curator的常用内容已学习完毕,建议每个都亲自操作一下,为之后的自动配置和分布式锁操作打下基础.

6.基于curator实现zookeeper分布式锁

先来了解下分布式锁应用场景:

比如我有一个电商商城,为了提高系统的服务量,业务被拆分小了,分别部署在不同的服务器上,下单成功后订单系统通知库存系统要及时减库存,但此时并发极高,库存系统还没来得及减库存,又有新的人进来了,读取了库存,数据被脏读了,于是他也下单成功了,但发货的时候发现库存不够了,于是闹得两家都不开心...这时候就需要分布式锁来解决.

为了演示这样的场景,我写了个小项目来模拟这种场景:

我数据库里电脑库存只有10台,然后写了两个下单页面,每个订单买8台电脑,让线程适当休眠来模拟高并发下的延迟,于是我几乎同时访问了这两个下单页面,最后两边都提示我下单成功了,但数据库里的库存数量变成了-6

这显然不是我想要的结果,正确的应该是,这两个人里只有一个人下单成功,另外一个人下单失败,提示库存不足,最后数据库里剩2台电脑库存.

为了解决这个问题,就必须让第一个人下单时,第二个人不能下单,只能等到第一个人下单完成后方可下单,或者第二个人下单时,第一个人不能下单,只能等到第二人下单完成方可下单,由于两套系统是分开部署的,不能像以前那样用同步锁/同步代码块Synchronized来解决了,这个时候就需要引出分布式锁,分布式锁可以用Redis或者zookeeper等实现,这篇主要讲一下用zk去实现.

思路:提供一把全局的锁,所有来购买的请求竞争这一把锁,谁先拿到这把锁,谁就有资格执行下单,没抢到锁的请求被挂起,等待有锁的请求完成下单后释放锁,然后唤醒被挂起的请求继续去竞争这把锁...

可以把这把锁当做是zk上的一个节点,所有请求发起时,创建该节点,第一个创建该节点成功的请求就意味着获得了锁,其他请求创建都会抛出异常,然后捕获该异常,用全局的countDownLatch将该请求挂起,等获得锁的节点完成下单后,把该节点删除(释放锁),然后计数器-1,把挂起的线程都唤醒,继续去竞争该锁...

下面就顺着这个思路一起去实现分布式锁:

这里默认使用上面已经写好的连接ZK的工具类来创建client.

public class ZkLockUtil \{  
//分布式锁,用于挂起当前线程,等待上一把分布式锁释放  
private static CountDownLatch DISTRIBUTE\_LOCK = new CountDownLatch(1);  
//分布式锁的总结点名  
private final static String ZK\_LOCK\_PROJECT = "zk-lock";  
//分布式锁节点名  
private final static String DISTRIBUTE\_LOCK\_NAME = "distribute-lock";  
/\*\*  
\* 获取分布式锁  
\*/  
public static void getLock() \{  
CuratorFramework client = ZkClientUtil.build();  
client.start();  
while (true) \{  
try \{  
client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath("/" + ZK\_LOCK\_PROJECT + "/" + DISTRIBUTE\_LOCK\_NAME);  
System.out.println("获取分布式锁成功...");  
return;  
\} catch (Exception e) \{  
try \{  
//如果没有获取到锁,需要重新设置同步资源值  
if (DISTRIBUTE\_LOCK.getCount() <= 0) \{  
DISTRIBUTE\_LOCK = new CountDownLatch(1);  
\}  
System.out.println("获取分布式锁失败,等待他人释放锁中...");  
DISTRIBUTE\_LOCK.await();  
\} catch (InterruptedException ie) \{  
ie.printStackTrace();  
\}  
\}  
\}  
\}  
  
/\*\*  
\* 释放锁资源  
\*/  
public static void release(String path) \{  
CuratorFramework client = ZkClientUtil.build();  
client.start();  
try \{  
client.delete().forPath(path);  
System.out.println("锁释放成功...");  
\} catch (Exception e) \{  
System.out.println("释放锁失败...");  
e.printStackTrace();  
\} finally \{  
client.close();  
\}  
\}  
  
/\*\*  
\* 为指定路径节点创建watch,观察锁状态  
\*/  
public static void addWatcher2Path(final String path) throws Exception \{  
CuratorFramework client = ZkClientUtil.build();  
client.start();  
final PathChildrenCache pathChildrenCache = new PathChildrenCache(client, path, true);  
pathChildrenCache.start(PathChildrenCache.StartMode.POST\_INITIALIZED\_EVENT);  
System.out.println("创建观察者成功...");  
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() \{  
public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception \{  
if (pathChildrenCacheEvent.getType().equals(PathChildrenCacheEvent.Type.CHILD\_REMOVED)) \{  
String nodePath = pathChildrenCacheEvent.getData().getPath();  
System.out.println("上一会话已释放锁或会话已断开...节点路径为:" + nodePath);  
if (nodePath.contains(DISTRIBUTE\_LOCK\_NAME)) \{  
DISTRIBUTE\_LOCK.countDown();  
System.out.println("释放计数器,计数器值为:"+DISTRIBUTE\_LOCK.getCount()+"让当前请求来获取分布式锁...");  
\}  
\}  
\}  
\});  
\}  
\}  
下面来测试一下,有空的话你可以写一个类似我这种下单的模式去测试,如果时间紧写个测试类模拟也无妨:

public class Test \{  
public static void main(String\[\] args) \{  
final ExecutorService threadpool = Executors.newCachedThreadPool();  
System.out.println("开始购买...");  
for (int i = 0; i <2 ; i++) \{  
threadpool.execute(new Runnable() \{  
public void run() \{  
System.out.println("我是线程:"+Thread.currentThread().getName()+"我开始抢购了...");  
ZkLockUtil.getLock();  
System.out.println(Thread.currentThread().getName()+":我正在疯狂的剁手购买中...");  
try \{  
Thread.sleep(2000);  
\} catch (InterruptedException e) \{  
e.printStackTrace();  
\}  
System.out.println(Thread.currentThread().getName()+":我买完了,有请下一位...");  
try \{  
ZkLockUtil.addWatcher2Path("/zk-lock");  
System.out.println("添加完毕...");  
ZkLockUtil.release("/zk-lock/distribute-lock");  
System.out.println("释放完毕...");  
Thread.sleep(1000);  
  
\} catch (Exception e) \{  
e.printStackTrace();  
\}  
\}  
\});  
\}  
\}  
\}  
测试结果可以看出,在其中一个线程购买时,另外一个线程被挂起等待...

或者用我这种2个服务一起下单模式的:

@RestController  
public class SellController \{  
@Autowired  
private SellRepository sellRepository;  
@RequestMapping("sell")  
public String sell(@RequestParam("num")Integer num)\{  
String msg = "下单成功!";  
try \{  
Thread.sleep(3000);  
ZkLockUtil.getLock();  
Sell sell = sellRepository.getOne(1);  
int remain = sell.getStock()-num;  
sell.setStock(remain);  
if (remain >= 0)\{  
sellRepository.save(sell);  
\}else\{  
msg = "下单失败,库存不足...";  
\}  
ZkLockUtil.addWatcher2Path("/zk-lock");  
ZkLockUtil.release("/zk-lock/distribute-lock");  
\} catch (Exception e) \{  
e.printStackTrace();  
msg = "下单异常...";  
\}  
return msg;  
\}  
测试结果:

数据库中电脑库存由原来的10变为2,达到期望效果:

但这样就大功告成了吗? 其实我觉得没有,原因是因为我在测试的时候发现一个问题,当我在浏览器中按住F5进行刷新页面(模拟高并发下的请求频率),在一开始一切都是正常的,刷一阵子之后八尔哥就出来了:

找了一下原因,发现问题出在了这里:

由于我不断的刷新页面,就意味着不断的去获取锁和释放锁,当锁被释放后计数器减1,会去唤醒线程去竞争锁,然后这个时候来没来的及唤醒,新的请求又进来了,此时新请求创建锁成功了,被唤醒的线程又抢不到锁了,但计数器仍处于0的状态,它会继续去创建锁,此时又有新的请求不断进来,不断创建锁...导致zk认为你是在不断的进行重复操作,于是它就把连接给退出了,然后又有新请求进来了,又要重新建立连接:

新建立的连接又会像上面一样在连绵不断的请求中断开,这样频繁的连接和断开,重复数次之后,ZK直接关闭了连接,导致后台无限报错...

为了解决这个问题,我搜罗各大网站,最后没有找到什么可以参考的东西,我甚至开始怀疑是我写的锁有问题,但后来我在apache curator官网找到了解释,其实curator已经帮我们封装好了一套分布式锁,可以直接拿来用的:

于是我直接调用了zk封装好的这一套分布式锁去做测试,普通情况下跟我自己写的分布式锁没啥两样,效果一样,然后我继续用F5模拟高并发下的情况,结果跟我自己的锁如出一辙...也是报同样的错,错误原因也一样,最后阅读了下源码,其实自己写的跟apache写的没啥两样,原理都一样,只是人家封装的更方便别人使用一些...

@RequestMapping("plus")  
public String sellPlus(@RequestParam("num")int num) throws Exception \{  
CuratorFramework client = ZkClientUtil.build();  
client.start();  
String lockPath = "/plus\_lock";  
InterProcessMutex lock = new InterProcessMutex(client, lockPath);  
String msg = "下单成功!";  
if ( lock.acquire(3, TimeUnit.SECONDS) )  
\{  
try  
\{  
Sell sell = sellRepository.getOne(1);  
int remain = sell.getStock()-num;  
if (remain >= 0)\{  
sell.setStock(remain);  
sellRepository.save(sell);  
\}else \{  
msg = "下单失败,库存不足...";  
\}  
\}  
finally  
\{  
lock.release();  
\}  
\}  
return msg;  
\}  
我总不能去怀疑apache 写的锁也有问题吧,那问题应该就出在了zk建立连接或者zk过滤连接的机制上,应该是zk以为那些频繁关闭又连接的请求是被攻击或者无效的请求,所以强制关闭这些连接,目前尚未去研究zk的连接机制,也不清楚研究了是否能解决该问题,所以基于zk的分布式锁就讲到这里,同时在我心里它已经不是做分布式锁的首选了,我会考虑使用redis或者其它分布式锁去解决,尤其是在高并发的情况下,感兴趣的可以继续关注本博,在redis系列教程中,我会讲如何用redis实现分布式锁.  
\---------------------  
作者：老汉健身  
来源：CSDN  
原文：https://blog.csdn.net/lovexiaotaozi/article/details/83382128  
版权声明：本文为博主原创文章，转载请附上博文链接！

转载于:https://www.cnblogs.com/a-du/p/9870077.html