2024年最全Zookeeper详解（最详细的zookeeper解析+项目实例），大数据开发事件分发面试_zk项目

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，涵盖了95%以上大数据知识点，真正体系化！

由于文件比较多，这里只是将部分目录截图出来，全套包含大厂面经、学习笔记、源码讲义、实战项目、大纲路线、讲解视频，并且后续会持续更新

需要这份系统化资料的朋友，可以戳这里获取

//3、创建临时节点（当断开连接时临时节点会自动删除,withMode中的属性可选择）
String path =client.create().withMode(CreateMode.EPHEMERAL).forPath(“/a”,“helloworld”.getBytes());
System.out.println(“创建节点：”+path);

###### 4）、查询节点



1
2
3
4
5
6

//1、查询节点的数据
byte[] bytes = client.getData().forPath(“/zuxia”);
System.out.println(new String(bytes));
​
//2、查询节点的数据（详情信息）
Stat stats=new Stat();
System.out.println(stats);//为了区分两个结果的不同
byte[] be = client.getData().storingStatIn(stats).forPath(“/zuxia”);
System.out.println(stats);

###### 5）、更新节点



1
2
3
4
5
6

//给节点赋值（返回值为Stat,可写可不写）
client.setData().forPath(“/ab”, “hello”.getBytes());

###### 6）、删除节点



1
2
3
4
5
6

//1、删除节点
System.out.println(“删除节点：”+client.delete().forPath(“/wjh”));
​
//2、删除带有子节点的目录节点
System.out.println(“删除子节点：”+client.delete().deletingChildrenIfNeeded().forPath(“/zuxia”));

###### 7）、**Watch**事件监听


•ZooKeeper 允许用户在指定节点上注册一些Watcher，并且在一些特定事件触发的时候，ZooKeeper 服务端会将事件通知到感兴趣的客户端上去，该机制是 ZooKeeper 实现分布式协调服务的重要特性。


•ZooKeeper 中引入了Watcher机制来实现了发布/订阅功能能，能够让多个订阅者同时监听某一个对象，当一个对象自身状态变化时，会通知所有订阅者。


•ZooKeeper提供了三种Watcher：


NodeCache : 只是监听某一个特定的节点


PathChildrenCache : 监控一个ZNode的子节点.


TreeCache : 可以监控整个树上的所有节点，类似于PathChildrenCache和NodeCache的组合


**1、NodeCache 监听事件**



1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

@Test
void testNodeCache() throws Exception {
// 1. 创建NodeCache
NodeCache nodeCache = new NodeCache(client, “/ab”);
// 2. 注册监听
nodeCache.getListenable().addListener(new NodeCacheListener() {
   @Override
   public void nodeChanged() throws Exception {
       System.out.println(“/ab节点发生变更”);
       byte[] dataBytes = nodeCache.getCurrentData().getData();
       System.out.println(“节点修改后的数据：” + new String(dataBytes));
  }
});
// 3. 开启监听，如果设置为true，则开启监听时，加载缓冲数据
nodeCache.start(true);
while(true){}
}

**2、PathChildrenCache 监听事件**



1
2
3
4
5

@Test
void testPathChildrenCache() throws Exception {
//创建监听对象(监听指定节点下的)
PathChildrenCache pathChildrenCache= new PathChildrenCache(client, “/zuxia”, true);
//注册监听事件
pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
   @Override
   public void childEvent(CuratorFramework cf, PathChildrenCacheEvent event) throws Exception {
       System.out.println(“节点发生变化了”);
       PathChildrenCacheEvent.Type type = event.getType();
       //当前判断的是当节点发生更新时进入改方法，可以选择添加或者删除的方法
       if (type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
           byte[] bytes = event.getData().getData();
           System.out.println(“节点修改后的数据”+new String(bytes));
      }
  }
});
//开启监听
pathChildrenCache.start();
while (true){}
}

**3、TreeCache 监听事件**



1
2
3
4
5

@Test
void testTreeCache() throws Exception {
   //创建监听对象
   TreeCache treeCache = new TreeCache(client, “/zuxia”);
   //注册监听
   treeCache.getListenable().addListener(new TreeCacheListener() {
       @Override
       public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
           System.out.println(“节点发生变化了”);
           TreeCacheEvent.Type type = treeCacheEvent.getType();
           if (type.equals(TreeCacheEvent.Type.NODE_ADDED)){
               System.out.println(“节点添加了”);
          }
      }
  });
   //开启监听
   treeCache.start();
   while (true){}
}

###### 8）、分布式锁实现


* 首先我们要了解什么是分布式锁？

 在我们进行单机应用开发，涉及并发同步的时候，我们往往采用synchronized或者Lock的方式来解决多线程间的代码同步问题，这时多线程的运行都是在同一个JVM之下，没有任何问题。

 但当我们的应用是分布式集群工作的情况下，属于多JVM下的工作环境，跨JVM之间已经无法通过多线程的锁解决同步问题。

 那么就需要一种更加高级的锁机制，来**处理种跨机器的进程之间的数据同步问题**——这就是分布式锁。
* ![](https://img-blog.csdnimg.cn/8715ebfefbd3477bbec3a73f6460e7e9.png)



* 其次也要悉知分布式锁的原理：

 核心思想：当客户端要获取锁，则创建节点，使用完锁，则删除该节点。

 1.客户端获取锁时，在lock节点下创建临时顺序节点。

 2.然后获取lock下面的所有子节点，客户端获取到所有的子节点之后，如果发现自己创建的子节点序号最小，那么就认为该客户端获取到了锁。使用完锁后，将该节点删除。

 3.如果发现自己创建的节点并非lock所有子节点中最小的，说明自己还没有获取到锁，此时客户端需要找到比自己小的那个节点，同时对其注册事件监听器，监听删除事件。

 4.如果发现比自己小的那个节点被删除，则客户端的

 Watcher会收到相应通知，此时再次判断自己创建的节点

 是否是lock子节点中序号最小的，如果是则获取到了锁，

 如果不是则重复以上步骤继续获取到比自己小的一个节点

 并注册监听。
* ![](https://img-blog.csdnimg.cn/22f8c7292ec14fa599330095b5e95342.png)



* 案例操作----模拟12306售票：

  •在Curator中有五种锁方案：

 •InterProcessSemaphoreMutex：分布式排它锁（非可重入锁）

 •InterProcessMutex：分布式可重入排它锁

 •InterProcessReadWriteLock：分布式读写锁

 •InterProcessMultiLock：将多个锁作为单个实体管理的容器

 •InterProcessSemaphoreV2：共享信号量

  方法类：

 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

package com.wjh;
​
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;
​
import java.util.concurrent.TimeUnit;
​
public class TickTest implements Runnable{
   private int x=10;//票数
   //创建分布式可重入排它锁对象
   private InterProcessMutex lock;
   CuratorFramework client;
​
   //当前方法的构造方法
   public TickTest() {
       //超时重试（连接间隔时间和超时连接次数）
       RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
       //连接zookeeper对象
       client = CuratorFrameworkFactory.newClient(
               “ip:port”,
               1000,
               60*1000,
               retryPolicy);
       //开始连接
       client.start();
       //创建分布式可重入排它锁对象连接zookeeper注册中心客户端
       //客户端中不用创建，这里会自动创建
       lock = new InterProcessMutex(client, “/lock”);
  }
​
   @Override
   public void run() {
       try {
           //设置锁
           lock.acquire(3, TimeUnit.SECONDS);
           while (true) {
               if(x>0){
                   //输出的调用线程的对象以及票数的数量
                   System.out.println(Thread.currentThread()+“票数:” + x);
                   //间隔200毫秒输出一次
                   Thread.sleep(200);
                   x–;
              }
          }
      } catch (Exception e) {
           throw new RuntimeException(e);
      }finally {
           try {
               //释放锁
               lock.release();
          } catch (Exception e) {
               throw new RuntimeException(e);
​
          }
      }
  }
}
​

1
2
3

测试类：

1
2

package com.wjh;
​
public class MaiTest {
​
   //使用main方法调用
   public static void main(String[] args) {
       //实现线程方法
       TickTest tick = new TickTest();
       //创建线程对象
       Thread t1 = new Thread(tick,“携程”);
       Thread t2 = new Thread(tick,“飞猪”);
       //启动线程
       t1.start();
       t2.start();
  }
}
​

#### 三、Zookeeper集群搭建


###### 1、zookeeper集群介绍


Leader选举：



 •  
 Serverid
：服务器
ID

          比如有三台服务器，编号分别是1,2,3。


          编号越大在选择算法中的权重越大。



 •  
 Zxid
：数据
ID

          服务器中存放的最大数据ID.值越大说明数据  越新，在选举算法中数据越新权重越大。



 •  
 在
Leader
选举的过程中，如果某台
ZooKeeper



![img](https://img-blog.csdnimg.cn/img_convert/3a2029a02739e25cd353aef702a27776.png)
![img](https://img-blog.csdnimg.cn/img_convert/e572d6f1e8b9943128068283fe8e20eb.png)

**网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**

**[需要这份系统化资料的朋友，可以戳这里获取](https://bbs.csdn.net/topics/618545628)**


**一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**

选举的过程中，如果某台
ZooKeeper



[外链图片转存中...(img-nJPyJnbt-1714897203034)]
[外链图片转存中...(img-49GZAQrH-1714897203034)]

**网上学习资料一大堆，但如果学到的知识不成体系，遇到问题时只是浅尝辄止，不再深入研究，那么很难做到真正的技术提升。**

**[需要这份系统化资料的朋友，可以戳这里获取](https://bbs.csdn.net/topics/618545628)**


**一个人可以走的很快，但一群人才能走的更远！不论你是正从事IT行业的老鸟或是对IT行业感兴趣的新人，都欢迎加入我们的的圈子（技术交流、学习资源、职场吐槽、大厂内推、面试辅导），让我们一起学习成长！**

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67