Zookeeper的学习与应用

1初始zookeeper

zookeeper是一个树形目录结构；通俗地讲，ZooKeeper是动物园管理员，它是拿来管大象 Hadoop、鲸鱼 HBase、Kafka等的管理员；
Zookeeper主要是一个分布式、开源的、分布式应用程序的协调服务；其主要功能：
1配置管理；

2分布式锁；

以上是单机加锁的情况，是可以应用的；但是，

这样的就出了问题，B服务出了问题；下面的方案是可以解决的：这个分布式锁就是zookeeper

3集群管理；

2zookeeper的安装（pass）

3数据模型

zookeeper是一个树形目录服务，其数据模型和linux的文件系统目录结构很类似，拥有一个层次结构；
这里的每一个阶段都称为 : ZNode --> 每个节点都会保存自己的数据和节点信息；
节点可以拥有子节点，同时也允许少量（1M）数据存储在该节点之下；
节点的类：
persistent 持久化节点
ephemeral 临时节点
persistent_sequential 持久化顺序节点
ephemeral_sequential 临时顺序节点

4zookeeperCli的命令

[zk: localhost:2181(CONNECTED) 40] ls /
[zookeeper]
[zk: localhost:2181(CONNECTED) 41] create /app1 null
Created /app1
[zk: localhost:2181(CONNECTED) 42] create /app2 null
Created /app2
[zk: localhost:2181(CONNECTED) 43] ls /
[zookeeper, app2, app1]
[zk: localhost:2181(CONNECTED) 45] create /app1/p1 null
Created /app1/p1
[zk: localhost:2181(CONNECTED) 46] create /app1/p2 null
Created /app1/p2
[zk: localhost:2181(CONNECTED) 47] create /app1/p3 null
Created /app1/p3
[zk: localhost:2181(CONNECTED) 49] ls /app1
[p1, p2, p3]

创建临时节点：

[zk: localhost:2181(CONNECTED) 50] create -e /app3 null
Created /app3
[zk: localhost:2181(CONNECTED) 51] ls /
[zookeeper, app2, app1, app3]
[zk: localhost:2181(CONNECTED) 52] quit
Quitting...[zk: localhost:2181(CONNECTED) 0] ls /
[zookeeper, app2, app1]
[zk: localhost:2181(CONNECTED) 1]

创建顺序节点（持久化）：

[zk: localhost:2181(CONNECTED) 0] create -s /app1 ""
Created /app10000000005
[zk: localhost:2181(CONNECTED) 1] create -s /app1 ""
Created /app10000000006
[zk: localhost:2181(CONNECTED) 2] ls /
[zookeeper, app2, app1, app10000000006, app10000000005]

创建临时的顺序节点：

[zk: localhost:2181(CONNECTED) 4] create -e -s /app4 ""
Created /app40000000009
[zk: localhost:2181(CONNECTED) 5] quit
Quitting...[zk: localhost:2181(CONNECTED) 0] ls /
[app10000000008, zookeeper, app2, app1, app10000000006, app10000000005, app4]

5 Zookeeper JavaAPI

Curator的介绍：javak客户端库；为了简化zookeeper的客户端的使用；
常见的zookeeper java api:

原生java api
ZkClient
Curator

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.zs</groupId><artifactId>zookeeper-test</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target></properties><dependencies><!-- https://mvnrepository.com/artifact/org.junit.jupiter/junit-jupiter-api --><dependency><groupId>org.junit.jupiter</groupId><artifactId>junit-jupiter-api</artifactId><version>5.5.1</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.curator/curator-recipes --><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.2.0</version></dependency><!-- https://mvnrepository.com/artifact/org.apache.curator/curator-framework --><dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.2.0</version></dependency><!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-api --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>2.0.0-alpha1</version></dependency><!-- https://mvnrepository.com/artifact/org.slf4j/slf4j-log4j12 --><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.30</version><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>RELEASE</version><scope>compile</scope></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.1</version><configuration><source>1.8</source><target>1.8</target><encoding>UTF-8</encoding></configuration></plugin></plugins></build></project>

public class Curator {private  static final int  BASESLEEPTIMEMS = 3000;private  static final int  MAXRETRIES = 3000;@Testpublic void test() {RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASESLEEPTIMEMS, MAXRETRIES);/*CuratorFramework client =CuratorFrameworkFactory.newClient("127.0.0.1:2181", 60 * 1000, 15 * 1000, retryPolicy);client.start();*/CuratorFramework client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").sessionTimeoutMs(60 * 1000).retryPolicy(retryPolicy).namespace("zs").build();client.start();}
}

创建节点：

/*** @author zhaoshuai06 <zhaoshuai06@kuaishou.com>* Created on 2021-06-04*/
public class Curator {private  static final int  BASESLEEPTIMEMS = 3000;private  static final int  MAXRETRIES = 3000;private  static CuratorFramework client;@BeforeAllpublic static void test() {RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASESLEEPTIMEMS, MAXRETRIES);client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").sessionTimeoutMs(60 * 1000).retryPolicy(retryPolicy).namespace("zs").build();client.start();}/*** create node*/@Testpublic void testCreate() throws Exception {/*如果创建节点， 没有创建数据，则默认将当前客户端的ip作为数据存储String s = client.create().forPath("/app1");System.out.println(s);client.create().forPath("/app2", "hello".getBytes(StandardCharsets.UTF_8));// 默认的是持久化client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3", "app3".getBytes(StandardCharsets.UTF_8));// 创建多级目录client.create().creatingParentContainersIfNeeded().forPath("/app4/p1");*/}@AfterAllpublic static void close() {if (client != null) {client.close();}}
}

查询节点：

@Testpublic void getTest() throws Exception {/*// get data-infobyte[] bytes = client.getData().forPath("/app1");System.out.println(new String(bytes));// get clidren-pathList<String> path = client.getChildren().forPath("/");System.out.println(path);*/Stat stat = new Stat();byte[] bytes = client.getData().storingStatIn(stat).forPath("/app1");System.out.println(new String(bytes));}

修改节点信息：

    @Testpublic void updateTest() throws Exception {client.setData().forPath("/app1", "app1-test-update".getBytes(StandardCharsets.UTF_8));}

根据版本修改信息：

   @Testpublic void updateTest() throws Exception {Stat stat = new Stat();client.getData().storingStatIn(stat).forPath("/app1");int version = stat.getVersion();client.setData().withVersion(version).forPath("/app1", "app1-test-update-version-hello".getBytes(StandardCharsets.UTF_8));}

删除节点：

  @Testpublic void deleteTest() throws Exception {/*// delete one nodeclient.delete().forPath("/app1");// Delete nodes with childrenclient.delete().deletingChildrenIfNeeded().forPath("/app4");// guaranteed deleteclient.delete().guaranteed().forPath("/zs");*/}

带回调的删除：

    @Testpublic void testCallDelete() throws Exception {/*client.delete().guaranteed().inBackground(new BackgroundCallback() {@Overridepublic void processResult(CuratorFramework curatorFramework, CuratorEvent curatorEvent) throws Exception {System.out.println("delete app1");}}).forPath("/app2");*/client.delete().guaranteed().inBackground((curatorFramework, curatorEvent) -> System.out.println("delete app1")).forPath("/app2");}

6Watch事件监听

Curator引入了Cache来实现对Zookeeper服务端事件的监听。

NodeCache事件监听：

 @Testpublic void testNodeCache() throws Exception {/*** NodeCache*/String path = "/root/app1";NodeCache nodeCache = new NodeCache(client, path);nodeCache.getListenable().addListener(() -> {System.out.print("The nodes have changed == ");// Gets the value of the modified nodeChildData currentData = nodeCache.getCurrentData();byte[] data = currentData.getData();System.out.println(new String(data));});// If it is set to true, the buffer data will be loaded when listening is turned onnodeCache.start(true);while (true) {}}

PathChildrenCache事件监听：感知子节点，但是感知不到自己：

 @Testpublic void testPathChildrenCache() throws Exception {String path = "/root";// 1 Create listening objectPathChildrenCache childrenCache = new PathChildrenCache(client, path, true);childrenCache.getListenable().addListener((curatorFramework, pathChildrenCacheEvent) -> {System.out.print("Child node changed == ");System.out.println(pathChildrenCacheEvent);});childrenCache.start();while (true) {}}

PathChildrenCache 监听特定的事件（update）:

  @Testpublic void testPathChildrenCache() throws Exception {String path = "/root";// 1 Create listening objectPathChildrenCache childrenCache = new PathChildrenCache(client, path, true);childrenCache.getListenable().addListener((curatorFramework, pathChildrenCacheEvent) -> {// Monitor the data changes of the child node, and get the changed data// --> set /root/app1 "hello-testPathChildrenCache-1"PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();if (type.equals(Type.CHILD_UPDATED)) {byte[] data = pathChildrenCacheEvent.getData().getData();System.out.println(new String(data));}});childrenCache.start();while (true) {}}

TreeCache = 子节点 + 自己节点

  @Testpublic void testTreeCache() throws Exception {String path = "/root";TreeCache treeCache = new TreeCache(client, path);treeCache.getListenable().addListener((curatorFramework, treeCacheEvent) -> {System.out.print("The nodes have changed == ");byte[] data = treeCacheEvent.getData().getData();System.out.println(new String(data));});treeCache.start();while (true) {}}

7分布式锁

这里我们使用zookeeper的分布式锁 == Curator的分布式锁 ==：

8zookeeper分布式锁

package com.zs;import java.util.concurrent.TimeUnit;import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;/*** @author zhaoshuai06 <zhaoshuai06@kuaishou.com>* Created on 2021-06-04*/
public class Curator {public static void main(String[] args) {Ticket12306 ticket12306 = new Ticket12306();//create clientThread t1 = new Thread(ticket12306, "xiecheng");Thread t2 = new Thread(ticket12306, "feizhu");t1.start();t2.start();}
}
class Ticket12306 implements Runnable {private  static final int  BASESLEEPTIMEMS = 3000;private  static final int  MAXRETRIES = 3000;private  static CuratorFramework client;private int tickets = 10; // database nums_ticketprivate InterProcessMutex lock;public Ticket12306() {String path = "/lock";RetryPolicy retryPolicy = new ExponentialBackoffRetry(BASESLEEPTIMEMS, MAXRETRIES);client = CuratorFrameworkFactory.builder().connectString("127.0.0.1:2181").sessionTimeoutMs(60 * 1000).retryPolicy(retryPolicy).build();client.start();lock = new InterProcessMutex(client, path);}@Overridepublic void run() {while (true) {try {// add locklock.acquire(3, TimeUnit.SECONDS);if (tickets > 0) {System.out.println(Thread.currentThread().getName() + ":" + tickets);Thread.sleep(100);tickets--;}} catch (Exception e) {e.printStackTrace();} finally {try {//release locklock.release();} catch (Exception e) {e.printStackTrace();}}}}
}

9搭建集群