zookeeper案例

目录

案例一：服务器动态上下线

服务端：

（1）先获取zookeeper连接

（2）注册服务器到zookeeper集群：

（3）业务逻辑（睡眠）：

服务端代码如下：

客户端：

（1）获取zookeeper的连接：

（2）监听/servers下边的子节点的增减：

客户端代码如下：

案例二：ZooKeeper 分布式锁

分布式锁是什么？

锁的实现：

构造函数：

加锁函数：

解锁函数：

整体代码：

测试类代码 ：

Curator 框架实现分布式锁案例：

实现步骤：

代码如下：

---

该案例主要也是客户端监听原理，客户端监听服务器的上下线情况

先在集群上创建/servers 节点（用于存储连接的服务器的主机和该服务器的节点数）相当于zookeeper集群

案例一：服务器动态上下线

服务端：

（1）先获取zookeeper连接

        创建类对象

该类为我们创建的服务端类：

        DistributeServer server = new DistributeServer();

        获取zookeeper连接：

自己创建连接方法：

    private void getconnect() throws IOException {zk = new ZooKeeper(connectstring, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {}});}

 让后server对象在main函数中调用

（2）注册服务器到zookeeper集群：

注册是需要注册到zookeeper集群的/servers路径下，需要指定参数进行创建

private void regestServer(String hostname) throws InterruptedException, KeeperException {
zk.create(parentNode+"/"+hostname,hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);//  需要创建有序的临时节点所以-e（暂时） -s（有序）System.out.println("服务器"+hostname+"已注册连接");}

（3）业务逻辑（睡眠）：

    private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}

服务端代码如下：

package com.tangxiaocong.case1;
import org.apache.zookeeper.*;
import java.io.IOException;
/*** @Date 2023/8/10 19:06* @Author */
public class DistributeServer {private static String connectstring="hadoop102:2181,hadoop103:2181,hadoop104:2181";private static int sessionTimeout=2000;private ZooKeeper zk =null;private String parentNode = "/servers";public static void main(String[] args) throws IOException, InterruptedException, KeeperException {//获取zk连接//创建DistributeServer server = new DistributeServer();server.getconnect();//注册服务器到zk集群//注册是需要在/servers节点下创建所开启的服务器的路径server.regestServer(args[0]);//业务逻辑（实际是延时让它睡觉---不然会注册完成就关闭）server.business();}private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}private void regestServer(String hostname) throws InterruptedException, KeeperException {
zk.create(parentNode+"/"+hostname,hostname.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);//  需要创建有序的临时节点所以-e（暂时） -s（有序）System.out.println("服务器"+hostname+"已注册连接");}private void getconnect() throws IOException {zk = new ZooKeeper(connectstring, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {}});}
}

客户端：

（1）获取zookeeper的连接：

        先创建客户端对象，在进行构建获取zookeeper连接的方法，本方法对process方法进行了重写，填写了再发生上下线的运行逻辑

 private void getConnect() throws IOException {zk= new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {try {getServerList();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}});}

（2）监听/servers下边的子节点的增减：

        构建方法client.getServerList()来进行监听：

代码逻辑就是通过getChildren()方法获取指定目录下的所有子目录并开启监听

再进行遍历，把遍历结果封装到一个集合中，最后进行输出

 private void getServerList() throws InterruptedException, KeeperException {List<String> children = zk.getChildren("/servers", true);//该方法会获取指定路径下的所有子节点//true 会走初始化中的watch 也可以自己创建watch//把所有的服务器都封装到一个集合ArrayList<String> list = new ArrayList<>();for (String child : children) {byte[] data = zk.getData("/servers" +"/"+ child, false, null);//上边已经便利到一个服务器对象，再进行添加list.add(new String(data));}System.out.println(list);}

（3）业务逻辑同服务端不在赘述。

客户端代码如下：

package com.tangxiaocong.case1;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
/*** @Date 2023/8/10 21:27* @Author * 客户端的监听功能*/
public class DistributeClient {
private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";private int sessionTimeout=2000;private ZooKeeper zk=null;public static void main(String[] args) throws IOException, InterruptedException, KeeperException {//获取zk连接DistributeClient client = new DistributeClient();client.getConnect();//监听/servers下边的子节点的增减client.getServerList();//业务逻辑（睡眠）client.business();}private void business() throws InterruptedException {Thread.sleep(Long.MAX_VALUE);}private void getServerList() throws InterruptedException, KeeperException {List<String> children = zk.getChildren("/servers", true);//该方法会获取指定路径下的所有子节点//true 会走初始化中的watch 也可以自己创建watch//把所有的服务器都封装到一个集合ArrayList<String> list = new ArrayList<>();for (String child : children) {byte[] data = zk.getData("/servers" +"/"+ child, false, null);//上边已经便利到一个服务器对象，再进行添加list.add(new String(data));}System.out.println(list);}private void getConnect() throws IOException {zk= new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {try {getServerList();} catch (InterruptedException e) {e.printStackTrace();} catch (KeeperException e) {e.printStackTrace();}}});}
}

案例二：ZooKeeper 分布式锁

分布式锁是什么？

日常使用计算机的时候，我们的电脑不会只开一个进程，但是当“进程1”在访问某些资源的时候，不能被其他进程所访问，它就会去获得锁，把她所访问的资源进行锁上，对该资源进行独占。"进程 1"用完该资源以后就将锁释放掉，让其 他进程来获得锁，那么通过这个锁机制，我们就能保证了分布式系统中多个进程能够有序的 访问该临界资源。那么我们把这个分布式环境下的这个锁叫作分布式锁。

锁的实现：

构造函数：

在该类中首先要实现构造方法，构造方法与类名相同，在该方法中需要获取连接，重写process方法，在该方法中实现释放CountDownLatch的类对象，有两种情况，正常连接释放一种，不是正常连接状态，则释放另一种。在构造方法中还要判断是否存在“/locks”路径，存在则正常退出，不存在则创建该路径。

加锁函数：

使用ZooKeeper对象进行创建节点（临时有序），让后获取“/locks”路径下的所有节点序号，对结果进行判断，如果返回的List集合只有一个节点，则直接返回，默认加锁，不用再做监听工作。如果不是只有一个节点，则对List集合进行排序，再获取他的节点名称，通过indexOf函数来获取该名称节点的下标。如果为-1，则数据异常，为0 则为最小节点，则直接退出，进行加锁不需要设置监听，结果为其他则需要设置监听，先设置监听字符串，当状态不发生改变会一致阻塞,只有上锁节点让位后会调用process方法进行释放。

解锁函数：

解锁就是直接删除节点即可

整体代码：

package com.tangxiaocong.case2;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
/*** @Date 2023/8/12 19:56* @Author */
public class DistributedLock {final    private String connectString="hadoop102:2181,hadoop103:2181,hadoop104:2181";final  private int sessionTimeout=2000;final    private   ZooKeeper zk;private String waitPath;private String currentModu;//为了程序的健壮性，创建该对象   等待操作final   private CountDownLatch waitLach=new CountDownLatch(1);final   private CountDownLatch countDownLatch=new CountDownLatch(1);public DistributedLock() throws IOException, InterruptedException, KeeperException {//获取连接zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {//  connectLatch  如果正常连接zk  可以释放if (watchedEvent.getState()==Event.KeeperState.SyncConnected){countDownLatch.countDown();}//检测到删除节点并且是前一个节点则释放waitlatchif (watchedEvent.getType()==Event.EventType.NodeDeleted && watchedEvent.getPath().equals(waitPath)){waitLach.countDown();}}});//等待是否正常连接  正常（已）连接会释放  否则阻塞countDownLatch.await();// 判断是否存在lock锁Stat stat = zk.exists("/locks", false);if (stat==null){//创建该节点String s = zk.create("/locks", "locks".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.CONTAINER);}}//对zk加锁public void zkLock()  {//创建临时的带序号的节点try {currentModu = zk.create("/locks/" + "seq-", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);List<String> children = zk.getChildren("/locks", false);//如果只有一个节点   则直接获取if(children.size()==1){return;}else {//排序Collections.sort(children);//直接从s后边开始   开始的下标就是length的长度String substring = currentModu.substring("/locks/".length());//通过substring来获取在List集合中的下标位置int index = children.indexOf(substring);if (index==-1){System.out.println("数据异常");}else if (index==0){return;}else {//  需要监听上一个节点waitPath="/locks/"+children.get(index-1);zk.getData(waitPath,true,new Stat());//等待监听waitLach.await();return;}}} catch (KeeperException e) {e.printStackTrace();} catch (InterruptedException e) {e.printStackTrace();}//判断创建的节点是否是最小序号的节点 如果是则获取锁  不是则监听他的前一个节点}//对zk解锁public void unzkLock(){
//删除节点try {//-1  是版本号zk.delete(this.currentModu,-1);} catch (InterruptedException  | KeeperException e) {e.printStackTrace();}}
}

测试类代码 ：

package com.tangxiaocong.case2;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
/*** @Date 2023/8/12 22:31* @Author 唐晓聪*/
public class DistributedLockTest
{public static void main(String[] args) throws IOException, InterruptedException, KeeperException {//创建两个客户端对象final    DistributedLock lock1 = new DistributedLock();final   DistributedLock lock2 = new DistributedLock();new Thread(new Runnable() {@Overridepublic void run() {try {  lock1.zkLock();System.out.println("线程1启动获得锁");Thread.sleep(5*1000);lock1.unzkLock();System.out.println("线程1释放锁");} catch (Exception e) {e.printStackTrace();}}}).start();new Thread(new Runnable() {@Overridepublic void run() {try {lock2.zkLock();System.out.println("线程2启动获得锁");Thread.sleep(5*1000);lock2.unzkLock();System.out.println("线程2释放锁");} catch (Exception e) {e.printStackTrace();}}}).start();}
}

Curator 框架实现分布式锁案例：

该案例是直接使用API进行实现分布式锁

实现步骤：

先创建分布式锁对象，new InterProcessMutex(),参数1为所要连接的客户端，参数2为监听路径

参数1传入的为getCuratorFramework()自定义函数，

该函数通过工厂类的方式进行建立连接，返回创建好的客户端，让后start启动客户端

创建完分布式锁对象后创建两个线程，在线程中进行获得锁，释放锁的操作。

代码如下：

package com.tangxiaocong.case3;
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;
/*** @Date 2023/8/13 20:07* @Author */
public class CuratorLockTest {public static void main(String[] args) {//创建分布式锁1//参数1   所连接的客户端 参数2 监听路径InterProcessMutex lock1 = new InterProcessMutex(getCuratorFramework(), "/locks");//创建分布式锁2InterProcessMutex lock2 = new InterProcessMutex(getCuratorFramework(), "/locks");//创建线程new Thread(new Runnable() {@Overridepublic void run() {try {lock1.acquire();System.out.println("thread 1 acquire lock");lock1.acquire();System.out.println("thread 1 again acquire lock");Thread.sleep(5*1000);lock1.release();System.out.println("thread 1 relax lock");lock1.release();System.out.println("thread 1 again relax lock");System.out.println();} catch (Exception e) {e.printStackTrace();}}}).start();new Thread(new Runnable() {@Overridepublic void run() {try {lock2.acquire();System.out.println("thread 2 acquire lock");lock2.acquire();System.out.println("thread 2 again acquire lock");Thread.sleep(5*1000);lock2.release();System.out.println("thread 2 relax lock");lock2.release();System.out.println("thread 2 again relax lock");} catch (Exception e) {e.printStackTrace();}}}).start();}private static CuratorFramework getCuratorFramework() {ExponentialBackoffRetry policy = new ExponentialBackoffRetry(3000, 3);//通过工厂类的方式进行建立连接CuratorFramework client = CuratorFrameworkFactory.builder().connectString("hadoop102:2181,hadoop102:2181,hadoop104:2181").connectionTimeoutMs(2000).sessionTimeoutMs(2000).retryPolicy(policy)//连接失败后  间隔多少秒下次间隔.build();client.start();System.out.println("zookeeper  success start  !!!!!");return client;}
}