《云计算平台与技术》｜实验二：Zookeeper Curator

type

date

slug

summary

status

题目一：Zookeeper Curator实现生产者消费者模型

[!cite]- 题目在分布式系统中，生产者消费者模型是一种常用的设计模式，其中生产者生成数据并放入缓冲区，消费者从缓冲区取出数据进行处理。在一个多线程或者多进程的环境中，同步和互斥是保证数据一致性的关键。请使用Apache Curator框架中提供的高级API来实现这个模型，并使用Zookeeper来完成同步和互斥。
题目要求
使用Java语言编写程序。
通过Zookeeper Curator的InterProcessMutex实现互斥，确保生产者和消费者在访问共享资源时的同步。
共享资源可以是一个在Zookeeper中维护的计数器，代表缓冲区中的项目数量。
生产者应当在缓冲区不满时添加项目，并更新计数器。
消费者应当在缓冲区不空时消费项目，并更新计数器。
请处理好异常情况，如Zookeeper连接中断等。
预期输出（控制台输出包含本人的标识信息，比如学号或姓名等）
程序应当能够在多线程或多进程环境下无误地运行。
当生产者生产数据时，消费者能够正确消费数据，且整个过程中共享资源的访问应该是同步的。

1 环境配置

1.1 安装ZooKeeper

知乎｜mac（m1版）环境下zookeeper安装使用

在官网(https://zookeeper.apache.org/releases.html) 下载3.8.4的压缩包，解压后移动到/usr/local文件夹下。

进入目录/usr/local/apache-zookeeper-3.8.4-bin/conf，将文件zoo_sample.cfg改名为zoo.cfg。

1.2 安装Maven+配置VSCode

官网下载Maven二进制压缩包apache-maven-3.9.6，解压并移动到/usr/local/目录。

配置环境变量，执行vim ~/.bash_profile并在末尾加入：

保存退出后执行source ~/.bash_profile更新配置。

之后参考CSDN｜使用VSCode实现Java项目管理 Maven相关插件及配置（Maven换源），将仓库地址更改为/usr/local/apache-maven-repository/repository，并配置VSCode插件“Maven for Java”。

在VSCode主页，点击左栏“新建JAVA项目”，使用Maven创建一个初始项目“producer-consumer”。

1.3 导入Curator

CSDN｜Zookeeper客户端ZkClient、Curator的使用，史上最详细的教程来啦~

修改VSCode项目中的pom.xml文件，加入依赖：

之后终端进入项目“producer-consumer”根目录，更新依赖项：

2 运行

在项目“producer-consumer”中完成代码编写后，打开终端启动zookeeper服务：

可知本地zookeeper服务端的端口为localhost:2181

运行项目“producer-consumer”，并手动开关zookeeper服务，终端输出如下：

题目二：Curator Recipes数据结构和锁的使用

[!cite]- 题目 Apache Curator是一个Zookeeper的客户端库，它包含了一系列简化Zookeeper使用的recipes。请描述以下至少三个Curator提供的锁或者其他数据结构，它们的原理以及在实际场景中的使用。
题目要求
选择并描述Curator Recipes中的queue或者barriers中的一种。
说明选定数据结构的工作原理。
提供实际的使用场景例子。
简述在这些场景中，这些数据结构如何使用。

2.1 Queue: DistributedQueue

Curator实现的一种分布式队列数据结构

注意，官方建议不使用Zookeeper来创建队列，因为zk的每个节点都有1MB的大小限制，但实际使用的队列中往往存放很多消息；而如果一个节点下有上千个子节点，也会严重降低运行性能

数据结构

根据Apache Curator官方文档，通过构建器QueueBuilder构建Distributed Queue实例，参数成员如下——

工作原理

DistributedQueue将队列的元素存储在Zookeeper的有序临时节点中，使用ZooKeeper的Watch机制监听队列的变化，并且维护一个特殊的ều持久节点来标识队列状态，确保队列路径非空而防止队列被强制删除。

对于每个新创建的有序临时节点，Zookeeper都会通过递增的计数器值为其分配节点名，因此节点序号能够表示队列元素的先后顺序（最小编号的节点对应队列头部的元素,而最大编号的节点对应队列尾部的元素）。如此，便可以利用有序节点的特性来模拟队列的入队、出队操作。

入队时，会先在路径下创建一个新的临时节点，将新元素的内容序列化并存储在这个新节点的数据中。

出队时，会先获取路径下所有现有子节点的列表，按照节点序号从小到大排列，并取出最小序号的节点（即队首），读取其中的元素数据（即反序列化），并将这个节点删除。

示例：分布式任务调度

假设有一个分布式的任务调度系统，有一个负责任务接收和分发的调度节点（Scheduler）和多个负责执行和结果报告的工作节点（Worker）。

Scheduler创建一个DistributedQueue实例并维护，每当接收新任务，就调用queue.put()将任务入队，DistributedQueue会创建新的有序临时节点，以存储任务内容。

每次Worker调用queue.remove()将一个任务出队，DistributedQueue获取最小编号节点即首节点，将其中的任务内容返回Worker，并删除该节点。若当前队列为空，则Worker阻塞等待，直到有新的任务加进来。

2.2 Barrier: DistributedBarrier

Curator实现的一种分布式同步机制（屏障），用于阻止一系列进程的执行过程，直到某个条件得到满足。

数据结构

工作原理

DistributedBarrier通过Zookeeper的有序临时节点来计数屏障中的等待进程，也同样使用ZooKeeper的Watch机制监听队列的变化。

首先，在指定路径下创建一个持久节点作为barrier的根路径节点，并指定大小参数N（需要等待的进程数）。

当某进程想要加入屏障时，调用enter()在指定路径下创建一个有序临时节点，写入屏障大小N并在根节点注册一个监听器Watcher，以监听其他进程改变子节点的情况。之后检查子节点数，若总数未达到N，则阻塞等待。

当最后一个进程加入屏障，子节点总数达到N，会触发所有在根节点上注册的Watcher，所有收到消息的进程同时调用leave()离开屏障，删除临时节点以重置屏障，并继续执行后续任务。

示例：分布式系统初始化

假设有一个分布式系统，其中包含1个配置节点和2个应用节点，且应用节点应当在配置节点启动后才启动。

首先创建一个DistributedBarrier实例，并将屏障大小设置为3，在路径下创建根节点；

配置节点在启动后调用barrier.enter()进入屏障，DistributedBarrier在根节点下创建一个有序临时节点。此时子节点总数不足3，因此配置节点阻塞等待；

两个应用节点加入屏障，根节点下创建两个临时节点。此时子节点总数达到3，Zookeeper触发所有节点的Watcher，所有节点调用barrier.leave()离开屏障，于是系统正常启动。

2.3 Lock: InterProcessSemaphoreV2

Curator实现的一种分布式信号量，用于控制能同时访问某个资源的并发进程数量。

数据结构

其中后者更灵活

工作原理

InterProcessSemaphoreV2同样基于Zookeeper的有序临时节点和Watch监听机制。

首先，在指定路径下创建一个持久节点作为根路径节点，用于存储该信号量的信息，如允许的最大并发数N等；

当某进程希望获取信号量以访问共享资源，调用acquire()请求一个信号量，在根路径下创建一个新的有序临时节点表示该请求。之后计算当前持有信号量的子节点数，若未达最大并发数N，则该请求成功获取信号量并立即返回；若当前持有信号量的子节点数已达到N，则当前请求阻塞等待，并在根路径注册一个Watcher监听是否有信号量被释放。

当某进程希望释放信号量，调用returnLease()，删除之前创建的临时子节点，以触发所有阻塞的Watcher，而这些阻塞的进程会按照拟定的先后顺序重新尝试获取信号量，并由其中的一个进程获得信号量并成功执行，其他进程继续阻塞。

示例：审批并发流程

假设有一个审批系统，可能有多个工作人员同时处理用户请求，需要限制同时处理的请求数，以确保人员工作的负载均衡。

首先创建一个InterProcessSemaphoreV2实例，并设置允许的最大并发数N，在路径下创建持久根节点；

当某工作人员希望处理用户请求，调用acquire()请求一个信号量，在根路径下创建一个新的有序临时节点。若此时信号量计数器未达最大值N，则获取成功，工作人员可以开始处理用户请求；否则，该请求会阻塞等待，并监听其他人员释放信号量的情况。

当某工作人员希望完成处理，调用returnLease()归还信号量，删除之前创建的临时子节点，并通知正在阻塞的所有工作人员，按照先后顺序优先分配，其他的工作人员继续阻塞。