目录

初识Zookeeper

Zookeeper作用

维护配置信息

分布式锁服务

集群管理

生产分布式唯一ID

Zookeeper的设计目标

Zookeeper 工作机制

数据模型

ZooKeeper 命令操作

服务端常用命令

 客户端常用命令

ZooKeeper JavaAPI操作

Curator 介绍

Curator API 常用操作

导入依赖

建立连接

创建节点

查询节点

修改节点

删除节点

Watch事件监听

概念

API演示 

分布式锁实现

分布式锁概述

ZooKeeper分布式锁原理 

​编辑

Curator实现分布式锁

API 

1. 配置

2. 可重入锁InterProcessMutex

3. 不可重入锁InterProcessSemaphoreMutex

4. 可重入读写锁InterProcessReadWriteLock

5. 联锁InterProcessMultiLock

6. 信号量InterProcessSemaphoreV2

7. 共享计数器

7.1. SharedCount

ZooKeeper 集群搭建

选举机制

选举过程

异常情况处理

集群中的角色定位

初识Zookeeper

  Zookeeper是一个经典的分布式数据一致性解决方案，致力于为分布式应用提供一个高性能、高可用,且具有严格顺序访问控制能力的分布式协调存储服务。

• 维护配置信息

• 分布式锁服务

• 集群管理

• 生成分布式唯一ID

Zookeeper作用

维护配置信息

  java编程经常会遇到配置项，比如数据库的url、 schema、user和 password等。通常这些配置项我们会放置在配置文件中，再将配置文件放置在服务器上当需要更改配置项时，需要去服务器上修改对应的配置文件。

        但是随着分布式系统的兴起,由于许多服务都需要使用到该配置文件,因此有必须保证该配置服务的高可用性(highavailability)和各台服务器上配置数据的一致性。

        通常会将配置文件部署在一个集群上，然而一个集群动辄上千台服务器，此时如果再一台台服务器逐个修改配置文件那将是非常繁琐且危险的的操作，因此就需要一种服务，能够高效快速且可靠地完成配置项的更改等操作，并能够保证各配置项在每台服务器上的数据一致性。

  zookeeper就可以提供这样一种服务，其使用Zab这种一致性协议来保证一致性。现在有很多开源项目使用zookeeper来维护配置，如在 hbase中，客户端就是连接一个 zookeeper，获得必要的 hbase集群的配置信息，然后才可以进一步操作。还有在开源的消息队列 kafka中，也便用zookeeper来维护 brokers的信息。在 alibaba开源的soa框架dubbo中也广泛的使用zookeeper管理一些配置来实现服务治理。

分布式锁服务

        一个集群是一个分布式系统，由多台服务器组成。为了提高并发度和可靠性，多台服务器上运行着同一种服务。当多个服务在运行时就需要协调各服务的进度，有时候需要保证当某个服务在进行某个操作时，其他的服务都不能进行该操作，即对该操作进行加锁，如果当前机器挂掉后，释放锁并 fail over到其他的机器继续执行该服务

集群管理

        一个集群有时会因为各种软硬件故障或者网络故障，出现棊些服务器挂掉而被移除集群，而某些服务器加入到集群中的情况，zookeeper会将这些服务器加入/移出的情况通知给集群中的其他正常工作的服务器，以及时调整存储和计算等任务的分配和执行等。此外zookeeper还会对故障的服务器做出诊断并尝试修复。

生产分布式唯一ID

        在过去的单库单表型系统中，通常可以使用数据库字段自带的auto_ increment属性来自动为每条记录生成一个唯一的ID。但是分库分表后，就无法在依靠数据库的auto_ Increment属性来唯一标识一条记录了。此时我们就可以用zookeeper在分布式环境下生成全局唯一ID。

Zookeeper的设计目标

zooKeeper致力于为分布式应用提供一个高性能、高可用，且具有严格顺序访问控制能力的分布式协调服务

高性能

• zookeeper将全量数据存储在内存中，并直接服务于客户端的所有非事务请求，尤其用于以读为主的应用场景

高可用

• zookeeper一般以集群的方式对外提供服务，一般3~5台机器就可以组成一个可用的 Zookeeper集群了，每台机器都会在内存中维护当前的服务器状态，井且每台机器之间都相互保持着通信。只要集群中超过一半的机器都能够正常工作，那么整个集群就能够正常对外服务

严格顺序访问

• 对于来自客户端的每个更新请求，Zookeeper都会分配一个全局唯一的递增编号，这个编号反应了所有事务操作的先后顺序

Zookeeper 工作机制

• Zookeeper从设计模式角度来理解：是一个基于观察者模式设计的分布式服务管理框架，它负责存储和管理大家都关心的数据，然后接受观察者的注册，一旦这些数据的状态发生变化，Zookeeper就将负责通知已经在Zookeeper上注册的那些观察者做出相应的反应。也就是说 Zookeeper = 文件系统 + 通知机制。

数据模型

ZooKeeper 命令操作

本节主要基于命令行操作Zookeeper

服务端常用命令

 客户端常用命令

ZooKeeper JavaAPI操作

Curator 介绍

Curator API 常用操作

注意：如果你采用的是Zookeeper3.5及以上的版本，需要采用的是Curator4.0版本来作为其客户端，对于Zookeeper3.5以下的版本，Curator4.0也兼容，所以无脑4.0

导入依赖

<!--curator-->
<dependency><groupId>org.apache.curator</groupId><artifactId>curator-framework</artifactId><version>4.0.0</version>
</dependency><dependency><groupId>org.apache.curator</groupId><artifactId>curator-recipes</artifactId><version>4.0.0</version>
</dependency>

建立连接

/*** 建立连接*/@Beforepublic void testConnect() {/*** @param connectString       连接字符串。zk server 地址和端口 "192.168.149.135:2181,192.168.149.136:2181"* @param sessionTimeoutMs    会话超时时间 单位ms* @param connectionTimeoutMs 连接超时时间 单位ms* @param retryPolicy         重试策略*//* //重试策略RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);//1.第一种方式CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.149.135:2181",60 * 1000, 15 * 1000, retryPolicy);*///重试策略RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
————————————————————————————————————————————————————————————————————————————————//2.第二种方式//CuratorFrameworkFactory.builder();client = CuratorFrameworkFactory.builder().connectString("192.168.149.135:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retryPolicy).namespace("itheima") // 设置命名空间，后续创建的节点默认以此为根节点.build();//开启连接client.start();}

创建节点

题外小知识点：如果我们在Junit单元测试的时候，一个测试方法的需要调用到其他@Test方法来进行初始化 or 资源释放操作，可以通过@Before 和 @After 注解来标注执行该@Test方法前先执行什么方法，后执行什么方法

@Before
public void connect(){建立连接的逻辑
}@Test
public void consum(){建立连接后执行的逻辑
}@After
public void release(){释放连接逻辑
}

上述代码，运行consum方法后，他的加载顺序是 connect() ——> consum() ——> release() 

   /*** 创建节点：create 持久 临时 顺序 数据* 1. 基本创建 ：create().forPath("")* 2. 创建节点 带有数据:create().forPath("",data)* 3. 设置节点的类型：create().withMode().forPath("",data)* 4. 创建多级节点  /app1/p1 ：create().creatingParentsIfNeeded().forPath("",data)*/@Testpublic void testCreate2() throws Exception {//1. 基本创建//如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储String path = client.create().forPath("/app1");  // 这里是在根节点后的节点路径System.out.println(path);}@Testpublic void testCreate() throws Exception {//2. 创建节点 带有数据//如果创建节点，没有指定数据，则默认将当前客户端的ip作为数据存储String path = client.create().forPath("/app2", "hehe".getBytes());System.out.println(path);}@Testpublic void testCreate3() throws Exception {//3. 设置节点的类型//默认类型：持久化String path = client.create().withMode(CreateMode.EPHEMERAL).forPath("/app3");System.out.println(path);}@Testpublic void testCreate4() throws Exception {//4. 创建多级节点  /app1/p1//creatingParentsIfNeeded():如果父节点不存在，则创建父节点String path = client.create().creatingParentsIfNeeded().forPath("/app4/p1");System.out.println(path);}

查询节点

/*** 查询节点：* 1. 查询数据：get: getData().forPath()* 2. 查询子节点： ls: getChildren().forPath()* 3. 查询节点状态信息：ls -s:getData().storingStatIn(状态对象).forPath()*/@Testpublic void testGet1() throws Exception {//1. 查询数据：getbyte[] data = client.getData().forPath("/app1");System.out.println(new String(data));}@Testpublic void testGet2() throws Exception {// 2. 查询子节点： lsList<String> path = client.getChildren().forPath("/"); // 查询整棵树System.out.println(path);}@Testpublic void testGet3() throws Exception {Stat status = new Stat();System.out.println(status);//3. 查询节点状态信息：ls -sclient.getData().storingStatIn(status).forPath("/app1");System.out.println(status);}

修改节点

    /*** 修改数据* 1. 基本修改数据：setData().forPath()* 2. 根据版本修改: setData().withVersion().forPath()* * version 是通过查询出来的。目的就是为了让其他客户端或者线程不干扰我。** @throws Exception*/@Testpublic void testSet() throws Exception {client.setData().forPath("/app1", "itcast".getBytes());}@Testpublic void testSetForVersion() throws Exception {Stat status = new Stat();//3. 查询节点状态信息：ls -sclient.getData().storingStatIn(status).forPath("/app1");int version = status.getVersion();//查询出来的 3System.out.println(version);client.setData().withVersion(version).forPath("/app1", "hehe".getBytes());}

删除节点

 /*** 删除节点： delete deleteall* 1. 删除单个节点:delete().forPath("/app1");* 2. 删除带有子节点的节点:delete().deletingChildrenIfNeeded().forPath("/app1");* 3. 必须成功的删除:为了防止网络抖动。本质就是重试。  client.delete().guaranteed().forPath("/app2");* 4. 回调：inBackground* @throws Exception*/@Testpublic void testDelete() throws Exception {// 1. 删除单个节点client.delete().forPath("/app1");}@Testpublic void testDelete2() throws Exception {//2. 删除带有子节点的节点client.delete().deletingChildrenIfNeeded().forPath("/app4");}@Testpublic void testDelete3() throws Exception {//3. 必须成功的删除client.delete().guaranteed().forPath("/app2");}@Testpublic void testDelete4() throws Exception {//4. 回调client.delete().guaranteed().inBackground(new BackgroundCallback(){@Overridepublic void processResult(CuratorFramework client, CuratorEvent event) throws Exception {System.out.println("我被删除了~");System.out.println(event);}}).forPath("/app1");}@Afterpublic void close() {if (client != null) {client.close();}}

Watch事件监听

概念

        Zookeeper的Watch事件监听是一种机制，用于在Zookeeper集群中监视和观察节点状态的变化。当某个节点发生变化时（例如数据内容的更改、节点的创建或删除等），Zookeeper会通知与该节点相关联的应用程序，以便应用程序能够及时作出相应的处理。 Watch事件监听可以帮助应用程序实时感知和处理节点状态的变化，并在节点状态发生变化时触发相应的回调函数或事件处理机制。这种机制使得应用程序能够根据事件的发生而采取适当的行动，从而实现分布式系统中各个节点之间的协调和同步。

Watch事件监听可以类比一下 volatile 关键字修饰的变量的保证多线程可见性的操作

API演示 

 /*** 建立连接*/@Beforepublic void testConnect() {/*** @param connectString       连接字符串。zk server 地址和端口 "192.168.149.135:2181,192.168.149.136:2181"* @param sessionTimeoutMs    会话超时时间 单位ms* @param connectionTimeoutMs 连接超时时间 单位ms* @param retryPolicy         重试策略*//* //重试策略RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000,10);//1.第一种方式CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.149.135:2181",60 * 1000, 15 * 1000, retryPolicy);*///重试策略RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);//2.第二种方式//CuratorFrameworkFactory.builder();client = CuratorFrameworkFactory.builder().connectString("192.168.149.135:2181").sessionTimeoutMs(60 * 1000).connectionTimeoutMs(15 * 1000).retryPolicy(retryPolicy).namespace("itheima").build();//开启连接client.start();}@Afterpublic void close() {if (client != null) {client.close();}}/*** 演示 NodeCache：给指定一个节点注册监听器*/@Testpublic void testNodeCache() throws Exception {//1. 创建NodeCache对象final NodeCache nodeCache = new NodeCache(client,"/app1");//2. 注册监听nodeCache.getListenable().addListener(new NodeCacheListener() {@Overridepublic void nodeChanged() throws Exception {System.out.println("节点变化了~");//获取修改节点后的数据byte[] data = nodeCache.getCurrentData().getData();System.out.println(new String(data));}});//3. 开启监听.如果设置为true，则开启监听是，加载缓冲数据nodeCache.start(true);}/*** 演示 PathChildrenCache：监听某个节点的所有子节点们*/@Testpublic void testPathChildrenCache() throws Exception {//1.创建监听对象PathChildrenCache pathChildrenCache = new PathChildrenCache(client,"/app2",true);//2. 绑定监听器pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {@Overridepublic void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {System.out.println("子节点变化了~");System.out.println(event);//监听子节点的数据变更，并且拿到变更后的数据//1.获取类型PathChildrenCacheEvent.Type type = event.getType();//2.判断类型是否是updateif(type.equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){System.out.println("数据变了！！！");byte[] data = event.getData().getData();System.out.println(new String(data));}}});//3. 开启pathChildrenCache.start();}/*** 演示 TreeCache：监听某个节点自己和所有子节点们*/@Testpublic void testTreeCache() throws Exception {//1. 创建监听器TreeCache treeCache = new TreeCache(client,"/app2");//2. 注册监听treeCache.getListenable().addListener(new TreeCacheListener() {@Overridepublic void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {System.out.println("节点变化了");System.out.println(event);}});//3. 开启treeCache.start();}

分布式锁实现

分布式锁概述

ZooKeeper分布式锁原理 

基于节点顺序依次获取锁可见，Zookeeper实现的锁是公平锁

Curator实现分布式锁

API 

在Curator中有五种锁方案：

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

1. 配置

添加curator客户端配置：

@Configuration
public class CuratorConfig {@Beanpublic CuratorFramework curatorFramework(){// 重试策略，这里使用的是指数补偿重试策略，重试3次，初始重试间隔1000ms，每次重试之后重试间隔递增。RetryPolicy retry = new ExponentialBackoffRetry(1000, 3);// 初始化Curator客户端：指定链接信息 及 重试策略CuratorFramework client = CuratorFrameworkFactory.newClient("192.168.1.111:2181", retry);client.start(); // 开始链接，如果不调用该方法，很多方法无法工作return client;}
}

2. 可重入锁InterProcessMutex

Reentrant和JDK的ReentrantLock类似， 意味着同一个客户端在拥有锁的同时，可以多次获取，不会被阻塞。它是由类InterProcessMutex来实现。

// 常用构造方法
public InterProcessMutex(CuratorFramework client, String path)
// 获取锁
public void acquire();
// 带超时时间的可重入锁
public boolean acquire(long time, TimeUnit unit);
// 释放锁
public void release();

注意：如想重入，则需要使用同一个InterProcessMutex对象。

3. 不可重入锁InterProcessSemaphoreMutex

public InterProcessSemaphoreMutex(CuratorFramework client, String path);
public void acquire();
public boolean acquire(long time, TimeUnit unit);
public void release();

// 构造方法
public InterProcessReadWriteLock(CuratorFramework client, String basePath);
// 获取读锁对象
InterProcessMutex readLock();
// 获取写锁对象
InterProcessMutex writeLock();

// 构造函数需要包含的锁的集合，或者一组ZooKeeper的path
public InterProcessMultiLock(List<InterProcessLock> locks);
public InterProcessMultiLock(CuratorFramework client, List<String> paths);// 获取锁
public void acquire();
public boolean acquire(long time, TimeUnit unit);// 释放锁
public synchronized void release();

// 构造方法
public InterProcessSemaphoreV2(CuratorFramework client, String path, int maxLeases);// 注意一次你可以请求多个租约，如果Semaphore当前的租约不够，则请求线程会被阻塞。
// 同时还提供了超时的重载方法
public Lease acquire();
public Collection<Lease> acquire(int qty);
public Lease acquire(long time, TimeUnit unit);
public Collection<Lease> acquire(int qty, long time, TimeUnit unit)// 租约还可以通过下面的方式返还
public void returnAll(Collection<Lease> leases);
public void returnLease(Lease lease);

7.1. SharedCount

// 构造方法
public SharedCount(CuratorFramework client, String path, int seedValue);
// 获取共享计数的值
public int getCount();
// 设置共享计数的值
public void setCount(int newCount) throws Exception;
// 当版本号没有变化时，才会更新共享变量的值
public boolean  trySetCount(VersionedValue<Integer> previous, int newCount);
// 通过监听器监听共享计数的变化
public void addListener(SharedCountListener listener);
public void addListener(final SharedCountListener listener, Executor executor);
// 共享计数在使用之前必须开启
public void start() throws Exception;
// 关闭共享计数
public void close() throws IOException;