ToB企服应用市场:ToB评测及商务社交产业平台

标题: 【ZooKeeper学习条记】 [打印本页]

---

作者: 用户云卷云舒    时间: 2024-7-15 09:31
标题: 【ZooKeeper学习条记】
1. ZooKeeper基本概念

   Zookeeper官网：https://zookeeper.apache.org/index.html
  

• Zookeeper是Apache Hadoop项目中的一个子项目，是一个树形目录服务
  
• Zookeeper翻译过来就是动物园管理员，用来管理Hadoop（大象）、Hive（蜜蜂）、Pig（小猪）的管理员，简称zk
  
• Zookeeper的本质是一个分布式的、开源的、提供分布式应用程序调和服务的组件
  
• Zookeeper提供的主要功能有：
  
  
  
  • 配置管理
    
  • 分布式锁
    
  • 集群管理
    
  
  

2. ZooKeeper常用命令

2.1 ZooKeeper数据模型

在正式介绍Zookeeper的常用命令之前，我们先来了解一下Zookeeper的相干数据模型：

• Zookeeper的是一个树形目录服务，其数据模型与unix文件系统目录树雷同，是一个条理化的结构
  
• 这里面的每一个节点都被称为ZNode，每个节点上都会保存本身的数据以及元数据信息
  
• 节点也可以拥有子节点，同时答应少量数据（1MB）存储在该节点之下
  
• 节点类型大致可以分为如下四类：
  
  
  
  • PERSISTENT：长期化节点
    
  • EPHEMERAL：临时节点 -e
    
  • PERSISTENT_SEQUENTIAL：长期化次序节点 -s
    
  • EPHEMERAL_SEQUENTIAL：临时次序节点 -e -s
    
  
  

2.2 ZooKeeper常用命令

Zookeeper是一个常见的客户端-服务器模型，我们可以利用命令行大概JavaAPI的方式充当客户端进行访问，其架构如下图所示：

• 服务端命令：
  

• ./zkServer.sh start启动zookeeper服务
  

• ./zkServer.sh status查看zookeeper服务运行状态
  

• ./zkServer.sh restart重启zookeeper服务
  

• ./zkServer.sh stop关闭zookeeper服务
  

• 客户端命令：
  

• ./zkCli.sh -server ip:port连接指定的zookeeper服务（如连接本地可忽略选项直接利用./zkCli.sh）
  

• quit退出客户端交互界面
  

• help查看命令帮助
  

• ls 目录查看指定目录下的znode节点
  

• ls -s 目录查看节点具体信息
  

• create znode [value]创建znode节点（可以携带data）
  

• create znode -e [value]创建临时节点（会话结束后消失）
  
• create znode -s [value]创建次序节点
  

• get znode查看节点携带数据
  

• set znode value设置节点数据
  

• delete znode删除指定的znode节点（必须为空）
  

• deleteall znode删除指定的znode节点及其子节点
  

2.3 ZooKeeper的JavaAPI操纵

2.3.1 Curator介绍

Curator：是一个Zookeeper的Java客户端库

• 常见的Zookeeper Java客户端有如下几种：
  
  
  • 原生JavaAPI
    
  • ZkClient
    
  • Curator
    
  
  
• Curator的目标就是简化Zookeeper客户端的利用
  
• Curator项目最初有Netflix公司研发，后来捐给了Apache基金会，成为顶级项目
  

   Curator官网：http://curator.apache.org/
  2.3.2 Curator API操纵

2.3.2.1 创建连接

我们可以利用CuratorFrameworkFactory静态工厂类进行创建，可以通过如下两种方式配置：

• 利用newClient()方法
  
• 利用build()方法
  

下面我们就给出对应两种代码的实现方式：
newClient：

1. /**
2. * ZooKeeper测试类
3. */
4. public class ZooKeeperTest {
6.     private CuratorFramework client = null;
8.     @Before
9.     public void initByNewClient() {
10.         CuratorFramework client = CuratorFrameworkFactory.newClient("127.0.0.1:2181",
11.                 3000,
12.                 3000,
13.                 new ExponentialBackoffRetry(3000, 1));
14.         this.client = client;
15.         this.client.start();
16.     }
17. }

复制代码

build：

1. /**
2. * ZooKeeper测试类
3. */
4. public class ZooKeeperTest {
6.     private CuratorFramework client = null;
8.     @Before
9.     public void init() {
10.         CuratorFramework client = CuratorFrameworkFactory
11.                 .builder()
12.                 .connectString("127.0.0.1:2181")
13.                 .sessionTimeoutMs(3000)
14.                 .connectionTimeoutMs(3000)
15.                 .retryPolicy(new ExponentialBackoffRetry(3000, 1))
16.                 .namespace("")
17.                 .build();
18.         client.start();
19.         this.client = client;
20.     }
21. }

复制代码

其中各个配置项含义如下：

• connectString：连接字符串，配置服务器地址，格式为ip:port
  
• sessionTimeoutMs：会话超时时间
  
• connectionTimeoutMs：连接超时时间
  
• retryPolicy：重试策略
  
• namespace：设置根目录位置
  

2.3.2.2 创建节点

创建节点有如下常见的四种方式：
Case1：创建节点（不携带数据）

1. @Test
2. public void testCreate1() throws Exception {
3.     String path = client.create().forPath("/app1");
4.     System.out.println(path);
5. }

复制代码

Case2：创建节点（携带数据）

1. @Test
2. public void testCreate2() throws Exception {
3.     String path = client.create().forPath("/app2", "curator java api".getBytes());
4.     System.out.println(path);
5. }

复制代码

Case3：创建多级节点

1. @Test
2. public void testCreate4() throws Exception {
3.     client.create().creatingParentsIfNeeded().forPath("/test/test1/test2");
4. }

复制代码

Case4：创建节点并指定类型

1. @Test
2. public void testCreate3() throws Exception {
3.     client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");
4.     client.create().withMode(CreateMode.PERSISTENT).forPath("/app4");
5.     client.create().withMode(CreateMode.PERSISTENT_SEQUENTIAL).forPath("/app5");
6.     client.create().withMode(CreateMode.EPHEMERAL_SEQUENTIAL).forPath("/app6");
7. }

复制代码

2.3.2.3 删除节点

删除节点有如下常见的两种方式：
Case1：删除节点（不含子节点）

1. @Test
2. public void testDelete() throws Exception {
3.     client.delete().forPath("/app1");
4. }

复制代码

Case2：删除节点（递归删除子节点）

1. @Test
2. public void testDeleteAll() throws Exception {
3.     client.delete().deletingChildrenIfNeeded().forPath("/test");
4. }

复制代码

2.3.2.4 查询节点

查询节点有如下常见的三种方式：
Case1：查询子节点信息

1. @Test
2. public void testGetChildren() throws Exception {
3.     List<String> childrenList = client.getChildren().forPath("/");
4.     System.out.println(childrenList);
5. }

复制代码

Case2：查询节点数据

1. @Test
2. public void testGetData() throws Exception {
3.     byte[] bytes = client.getData().forPath("/app2");
4.     System.out.println("data: " + new String(bytes));
5. }

复制代码

Case3：查询节点具体信息

1. @Test
2. public void testGetData3() throws Exception {
3.     Stat stat = new Stat();
4.     client.getData().storingStatIn(stat).forPath("/app2");
5.     System.out.println(stat);
6. }

复制代码

2.3.2.5 修改节点

修改节点有如下常见的两种方式：
Case1：修改节点数据

1. @Test
2. public void testSetData() throws Exception {
3.     Stat stat = client.setData().forPath("/app1", "some data".getBytes());
4.     System.out.println(stat);
5. }

复制代码

Case2：修改节点数据（带有版本号）

1. @Test
2. public void testSetData2() throws Exception {
3.     Stat stat = new Stat();
4.     client.getData().storingStatIn(stat).forPath("/app2");
5.     int version = stat.getVersion();
6.     System.out.println(version);
7.     client.setData().withVersion(version).forPath("/app2", "set with version".getBytes());
8. }

复制代码

3. ZooKeeper的变乱监听机制

Watcher变乱监听机制：

• ZooKeeper答应用户在指定节点上注册一些Watcher，当一些特定变乱发生时，ZooKeeper就会将变乱关照给对其感兴趣的客户端，这是ZooKeeper提供分布式调和服务的重要特性
  
• ZooKeeper引入了Watcher机制来实现发布 / 订阅功能，能够让多个订阅者同时监听某一个对象，当一个对象状态发生变革时就会关照所有订阅者
  
• ZooKeeper提供原生Watcher的方式，但是比力贫苦，因此Curator利用Cache数据结构进行了优化实现监听机制
  
• Curator提供了如下三种Cache：
  
  
  • NodeCache：只监听某一个指定的节点变革
    
  • PathChildrenCache：监控一个节点的所有子节点
    
  • TreeCache：监控整个树上的节点，雷同于前两者的组合
    
  
  

3.1 Node Cache

代码实现：

1. @Test
2. public void testCuratorCache() throws Exception {
3.     NodeCache cache = new NodeCache(client, "/app1");
4.     cache.getListenable().addListener(new NodeCacheListener() {
5.         @Override
6.         public void nodeChanged() throws Exception {
7.             System.out.println("监听到节点变化...");
8.         }
9.     });
10.     cache.start();
11.     while (true) {
13.     }
14. }

复制代码

3.2 PathChildren Cache

代码实现：

1. @Test
2. public void testPathChildrenCache() throws Exception {
3.     PathChildrenCache cache = new PathChildrenCache(client, "/app1", true);
4.     cache.getListenable().addListener(new PathChildrenCacheListener() {
5.         @Override
6.         public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
7.             System.out.println("监听到子节点变化...");
8.             PathChildrenCacheEvent.Type type = pathChildrenCacheEvent.getType();
9.             if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)) {
10.                 System.out.println("监听到子节点数据变化...");
11.                 System.out.println("更新后数据: " + pathChildrenCacheEvent.getData().getData());
12.             }
13.         }
14.     });
15.     cache.start();
16.     while (true) {
18.     }
19. }

复制代码

3.3 Tree Cache

代码实现：

1. @Test
2.     public void testTreeCache() throws Exception {
3.         TreeCache cache = new TreeCache(client, "/app1");
4.         cache.getListenable().addListener(new TreeCacheListener() {
5.             @Override
6.             public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
7.                 System.out.println("监听到节点发生变化...");
8.                 System.out.println(treeCacheEvent);
9.             }
10.         });
11.         cache.start();
12.         while (true) {
14.         }
15.     }

复制代码

4. ZooKeeper分布式锁

4.1 ZooKeeper分布式锁原理

• 核心思想：当用户获取到锁时就创建节点，利用完锁就删除节点
  

• 每当一个用户想要获取锁时就在/lock节点下创建一个 **临时次序 **节点
  
• 然后获取/lock节点下的全部子节点，假如发现当前节点编号是最小的，则该节点对应的客户端获取到锁，利用完锁后，删除该节点
  
• 假如发现节点编号不是最小的，则对前一个比本身小的编号节点，并注册变乱监听器，监听删除变乱
  
• 假如后续发现比本身小的节点被删除，则客户端会吸收到来自ZooKeeper的关照，然后再次判定所对应节点编号是否是最小的，重复上述步调
  

   注意：这里创建临时节点是因为防止获取到锁的客户端宕机了，进而导致锁永远不会被删的情况；这是创建次序节点是方便编号的排序
  Cutator提供了下面五种分布式锁的方式：

• InterProcessMutex（分布式可重入排他锁）
  
• InterProcessSemaphoreMutex（分布式不可重入排他锁）
  
• InterProcessReadWriteLock（分布式读写锁）
  
• InterProcessMutliLock（将多个锁作为单个实体管理的容器）
  
• InterProcessSemaphoreV2（共享信号量）
  

4.2 分布式锁实战（模拟12306抢票）

代码如下：

1. package org.example;
3. import org.apache.curator.framework.CuratorFramework;
4. import org.apache.curator.framework.CuratorFrameworkFactory;
5. import org.apache.curator.framework.recipes.locks.InterProcessMutex;
6. import org.apache.curator.retry.ExponentialBackoffRetry;
8. import java.util.concurrent.TimeUnit;
10. public class ZooKeeperLockTest {
12.     private static int tickets = 10; // 票数
14.     public static void main(String[] args) {
15.         // 建立连接
16.         CuratorFramework client = CuratorFrameworkFactory
17.                 .builder()
18.                 .connectString("127.0.0.1:2181")
19.                 .sessionTimeoutMs(3000)
20.                 .connectionTimeoutMs(3000)
21.                 .retryPolicy(new ExponentialBackoffRetry(3000, 1))
22.                 .namespace("")
23.                 .build();
24.         client.start();
25.         // 获取分布式锁
26.         InterProcessMutex lock = new InterProcessMutex(client, "/lock");
27.         Thread t1 = new Thread(() -> {
28.             while (true) {
29.                 try {
30.                     boolean hasLock = lock.acquire(3, TimeUnit.SECONDS);
31.                     if (hasLock && tickets > 0) {
32.                         // 不断抢票
33.                         System.out.println("线程" + Thread.currentThread().getName() + "抢到了当前第" + tickets + "张票");
34.                         tickets--;
35.                         if (tickets <= 0) {
36.                             break;
37.                         }
38.                     }
39.                 } catch (Exception e) {
40.                     throw new RuntimeException(e);
41.                 } finally {
42.                     try {
43.                         lock.release();
44.                     } catch (Exception e) {
45.                         throw new RuntimeException(e);
46.                     }
47.                 }
48.             }
49.         }, "携程");
51.         Thread t2 = new Thread(() -> {
52.             while (true) {
53.                 try {
54.                     boolean hasLock = lock.acquire(3, TimeUnit.SECONDS);
55.                     if (hasLock && tickets > 0) {
56.                         // 不断抢票
57.                         System.out.println("线程" + Thread.currentThread().getName() + "抢到了当前第" + tickets + "张票");
58.                         tickets--;
59.                         if (tickets <= 0) {
60.                             break;
61.                         }
62.                     }
63.                 } catch (Exception e) {
64.                     throw new RuntimeException(e);
65.                 } finally {
66.                     try {
67.                         lock.release();
68.                     } catch (Exception e) {
69.                         throw new RuntimeException(e);
70.                     }
71.                 }
72.             }
73.         }, "飞猪");
74.         t1.start();
75.         t2.start();
76.     }
77. }

复制代码

5. ZooKeeper集群管理

Leader推举过程：

• ServerId：服务器ID
  

比如有三台服务器，编号分别是1，2，3。则编号越大在选择算法中的权重就越大

• Zxid：数据ID
  

服务器中存放的数据ID越大，值越大阐明更新的越频繁，则在选择算法中的权重就越大

• 在Leader推举的过程中假如某台ZooKeeper凌驾了半数选票，则直接当选为Leader
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4