[大数据] Zookeeper 入门（3）：服务器动态上下线监听案例，Zookeeper 分布式锁案例，Curator 框架实现分布式锁案例

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 大数据 -> Zookeeper 入门（3）：服务器动态上下线监听案例，Zookeeper 分布式锁案例，Curator 框架实现分布式锁案例 -> 正文阅读

[大数据]Zookeeper 入门（3）：服务器动态上下线监听案例，Zookeeper 分布式锁案例，Curator 框架实现分布式锁案例

服务器动态上下线监听案例

某分布式系统中，主节点可以有多台，可以动态上下线，任意一台客户端都能实时感知到主节点服务器的上下线。
在这里插入图片描述
如上图所示，我们希望客户端能够实时监听服务器的节点变化情况。

具体实现

（1）先在集群上创建 /servers 节点

[zk: localhost:2181(CONNECTED) 10] create /servers "servers"
Created /servers

（2）在 Idea 中创建包名：com.atguigu.zkcase1

（3）服务器端向 Zookeeper 注册代码

package com.atguigu.zkcase1;
import java.io.IOException;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooDefs.Ids;
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";
	// 创建到 zk 的客户端连接
	public void getConnect() throws IOException{
		zk = new ZooKeeper(connectString, sessionTimeout, new 
		Watcher() {
		@Override
		public void process(WatchedEvent event) {
		}
		});
	}
	// 注册服务器
	public void registServer(String hostname) throws Exception{
		String create = zk.create(parentNode + "/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		System.out.println(hostname +" is online "+ create);
	}
	// 业务功能
	public void business(String hostname) throws Exception{
		System.out.println(hostname + " is working ...");
		Thread.sleep(Long.MAX_VALUE);
	}
	public static void main(String[] args) throws Exception {
		// 1 获取 zk 连接
		DistributeServer server = new DistributeServer();
		server.getConnect();
		// 2 利用 zk 连接注册服务器信息
		server.registServer(args[0]);
		// 3 启动业务功能
		server.business(args[0]);
	}
}

客户端代码，监听zk的节点变化

package com.atguigu.zkcase1;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
public class DistributeClient {
	private static String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
	private static int sessionTimeout = 2000;
	private ZooKeeper zk = null;
	private String parentNode = "/servers";
	// 创建到 zk 的客户端连接
	public void getConnect() throws IOException {
		zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			@Override
			public void process(WatchedEvent event) {
			// 再次启动监听
			try {
				getServerList();
			} catch (Exception e) {
				e.printStackTrace();
			}
			}
		});
	}
	// 获取服务器列表信息
	public void getServerList() throws Exception {
		 // 1 获取服务器子节点信息，并且对父节点进行监听
		List<String> children = zk.getChildren(parentNode, true);
		 // 2 存储服务器信息列表
		ArrayList<String> servers = new ArrayList<>();
		 // 3 遍历所有节点，获取节点中的主机名称信息
		for (String child : children) {
			byte[] data = zk.getData(parentNode + "/" + child, false, null);
			servers.add(new String(data));
		}
		 // 4 打印服务器列表信息
		System.out.println(servers);
	}
	// 业务功能
	public void business() throws Exception{
		System.out.println("client is working ...");
		Thread.sleep(Long.MAX_VALUE);
	}
	public static void main(String[] args) throws Exception {
		// 1 获取 zk 连接
		DistributeClient client = new DistributeClient();
		client.getConnect();
		// 2 获取 servers 的子节点信息，从中获取服务器信息列表
		client.getServerList();
		// 3 业务进程启动
		client.business();
	}
}

测试

1）在 Linux 命令行上操作增加减少服务器

（1）启动 DistributeClient 客户端

（2）在 hadoop102 上 zk 的客户端/servers目录上创建临时带序号节点

[zk: localhost:2181(CONNECTED) 1] create -e -s /servers/hadoop102 "hadoop102"

[zk: localhost:2181(CONNECTED) 2] create -e -s /servers/hadoop103 "hadoop103"

（3）观察 Idea 控制台变化

# 客户端监控节点变化的输出
[hadoop102, hadoop103]

（4）执行删除操作

[zk: localhost:2181(CONNECTED) 8] delete /servers/hadoop1020000000000

（5）观察 Idea 控制台变化
```
[hadoop103]
```

2）在 Idea 上操作增加减少服务器

（1）启动 DistributeClient 客户端（如果已经启动过，不需要重启
（2）启动 DistributeServer 服务
- ①点击 Edit Configurations…
- ②在弹出的窗口中（Program arguments）输入想启动的主机，例如，hadoop102
回到 DistributeServer 的 main 方法，右键，在弹出的窗口中点击 Run “DistributeServer.main()”
④观察 DistributeServer 控制台，提示 hadoop102 is working
⑤观察 DistributeClient 控制台，提示 hadoop102 已经上线

ZooKeeper 分布式锁案例

获取锁

在 Zookeeper 当中创建一个持久节,当第一个客户端 Client1 想要获得锁时,需要在这个节点下面创建一个临时顺序节点。
Client1查找持久节点下面所有的临时顺序节点并排序，判断自己所创建的节点是不是顺序最靠前的一个。如果是第一个节点，则成功获得锁。
如果再有一个客户端 Client2 前来获取锁，则在持久节点下面再创建一个临时顺序节点Lock2。
Client2查找持久节点下面所有的临时顺序节点并排序，判断自己所创建的节点Lock2是不是顺序最靠前的一个，结果发现节点Lock2并不是最小的。于是，Client2向排序仅比它靠前的节点Lock1注册Watcher，用于监听Lock1节点是否存在。这意味着Client2抢锁失败，进入了等待状态。
如果又有一个客户端Client3前来获取锁，则在持久节点下载再创建一个临时顺序节点Lock3。
Client3查找持久节点下面所有的临时顺序节点并排序，判断自己所创建的节点Lock3是不是顺序最靠前的一个，结果同样发现节点Lock3并不是最小的。于是，Client3向排序仅比它靠前的节点Lock2注册Watcher，用于监听Lock2节点是否存在。这意味着Client3同样抢锁失败，进入了等待状态。

释放锁

释放锁就比较简单了，因为前面创建的临时顺序节点，所以在出现下面两种情况时，都会自动释放锁：

任务完成后，Client 会释放锁。
任务没完成，Client 就崩溃了，也会自动释放锁。

代码编写

分布式锁的编写

package com.atguigu.lock2;
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;
public class DistributedLock {
	 // zookeeper server 列表
	 private String connectString = "hadoop102:2181,hadoop103:2181,hadoop104:2181";
	 // 超时时间
	 private int sessionTimeout = 2000;
	 private ZooKeeper zk;
	 private String rootNode = "locks";
	 private String subNode = "seq-";
	 // 当前 client 等待的子节点
	 private String waitPath;
	 //ZooKeeper 连接
	 private CountDownLatch connectLatch = new CountDownLatch(1);
	//ZooKeeper 节点等待
	 private CountDownLatch waitLatch = new CountDownLatch(1);
	 // 当前 client 创建的子节点
	 private String currentNode;
	 // 和 zk 服务建立连接，并创建根节点
	 public DistributedLock() throws IOException, InterruptedException, KeeperException {
		 zk = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			 @Override
			 public void process(WatchedEvent event) {
			 // 连接建立时, 打开 latch, 唤醒 wait 在该 latch 上的线程
			 if (event.getState() == Event.KeeperState.SyncConnected) {
			 	connectLatch.countDown();
			 }
		 // 发生了 waitPath 的删除事件
		 if (event.getType() == Event.EventType.NodeDeleted && event.getPath().equals(waitPath)) 
			{
				 waitLatch.countDown();
			 }
	 		}
		 });
		 // 等待连接建立
		 connectLatch.await();
		 //获取根节点状态
		 Stat stat = zk.exists("/" + rootNode, false);
		 //如果根节点不存在，则创建根节点，根节点类型为永久节点
		 if (stat == null) {
			 System.out.println("根节点不存在");
			 zk.create("/" + rootNode, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
		 }
	 }
	 // 加锁方法
	 public void zkLock() {
		 try {
			 //在根节点下创建临时顺序节点，返回值为创建的节点路径
			 currentNode = zk.create("/" + rootNode + "/" + subNode, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
			 // wait 一小会, 让结果更清晰一些
			 Thread.sleep(10);
		 	// 注意, 没有必要监听"/locks"的子节点的变化情况
			List<String> childrenNodes = zk.getChildren("/" + rootNode, false);
			 // 列表中只有一个子节点, 那肯定就是 currentNode , 说明client 获得锁
			 if (childrenNodes.size() == 1) {
			 		return;
			 } else {
				 //对根节点下的所有临时顺序节点进行从小到大排序
				 Collections.sort(childrenNodes);
				 //当前节点名称
				 String thisNode = currentNode.substring(("/" + rootNode + "/").length());
				 //获取当前节点的位置
		 		int index = childrenNodes.indexOf(thisNode);
		 		if (index == -1) {
		 		System.out.println("数据异常");
				 } else if (index == 0) {
		 			// index == 0, 说明 thisNode 在列表中最小, 当前client 获得锁
		 			return;
		 		} else {
				 // 获得排名比 currentNode 前 1 位的节点
		 		this.waitPath = "/" + rootNode + "/" + childrenNodes.get(index - 1);
				 // 在 waitPath 上注册监听器, 当 waitPath 被删除时, zookeeper 会回调监听器的 process 方法
				 zk.getData(waitPath, true, new Stat());
				 //进入等待锁状态
				waitLatch.await();
		 		return;
				 }
			 }
			 } catch (KeeperException e) {
			 	e.printStackTrace();
			 } catch (InterruptedException e) {
				 e.printStackTrace();
			 }
		 }
	 // 解锁方法
	 public void zkUnlock() {
		 try {
			 zk.delete(this.currentNode, -1);
		 } catch (InterruptedException | KeeperException e) {
			 e.printStackTrace();
		 }
	 }
}

测试zookeeper 的分布式锁

package com.atguigu.lock2;
import org.apache.zookeeper.KeeperException;
import java.io.IOException;
public class DistributedLockTest {
	public static void main(String[] args) throws
		InterruptedException, IOException, KeeperException {
		// 创建分布式锁 1
		final DistributedLock lock1 = new DistributedLock();
		// 创建分布式锁 2
		final DistributedLock lock2 = new DistributedLock();
		new Thread(new Runnable() {
			@Override
			public void run() {
			// 获取锁对象
				try {
					lock1.zkLock();
					System.out.println("线程 1 获取锁");
					Thread.sleep(5 * 1000);
					lock1.zkUnlock();
					System.out.println("线程 1 释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
		new Thread(new Runnable() {
			@Override
			public void run() {
			// 获取锁对象
				try {
					lock2.zkLock();
					System.out.println("线程 2 获取锁");
					Thread.sleep(5 * 1000);
					lock2.zkUnlock();
					System.out.println("线程 2 释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
	}
}

观察控制台变化：

线程 1 获取锁
线程 1 释放锁
线程 2 获取锁
线程 2 释放锁

Curator 框架实现分布式锁案例

1）原生的 Java API 开发存在的问题

（1）会话连接是异步的，需要自己去处理。比如使用 `CountDownLatch
（2）Watch 需要重复注册，不然就不能生效
（3）开发的复杂性还是比较高的
（4）不支持多节点删除和创建。需要自己去递归

2）Curator 是一个专门解决分布式锁的框架

Curator 解决了原生 JavaAPI 开发分布式遇到的问题
详情请查看官方文档：https://curator.apache.org/index.html

3）Curator 案例实操

1、添加依赖

<dependency>
	 <groupId>org.apache.curator</groupId>
	 <artifactId>curator-framework</artifactId>
	 <version>4.3.0</version>
</dependency>
<dependency>
	 <groupId>org.apache.curator</groupId>
	 <artifactId>curator-recipes</artifactId>
	 <version>4.3.0</version>
</dependency>
<dependency>
	 <groupId>org.apache.curator</groupId>
	 <artifactId>curator-client</artifactId>
	 <version>4.3.0</version>
</dependency>

代码编写

package com.atguigu.lock;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import
org.apache.curator.framework.recipes.locks.InterProcessLock;
import
org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
public class CuratorLockTest {
	private String rootNode = "/locks";
	// zookeeper server 列表
	private String connectString ="hadoop102:2181,hadoop103:2181,hadoop104:2181";
	// connection 超时时间
	private int connectionTimeout = 2000;
	// session 超时时间
	private int sessionTimeout = 2000;
	public static void main(String[] args) {
		new CuratorLockTest().test();
	}
	// 测试
	private void test() {
		// 创建分布式锁 1
		final InterProcessLock lock1 = new
		InterProcessMutex(getCuratorFramework(), rootNode);
		// 创建分布式锁 2
		final InterProcessLock lock2 = new
		InterProcessMutex(getCuratorFramework(), rootNode);
		new Thread(new Runnable() {
			@Override
			public void run() {
			// 获取锁对象
				try {
					lock1.acquire();
					System.out.println("线程 1 获取锁");
					// 测试锁重入
					lock1.acquire();
					System.out.println("线程 1 再次获取锁");
					Thread.sleep(5 * 1000);
					lock1.release();
					System.out.println("线程 1 释放锁");
					lock1.release();
					System.out.println("线程 1 再次释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
		new Thread(new Runnable() {
			@Override
			public void run() {
			// 获取锁对象
				try {
					lock2.acquire();
					System.out.println("线程 2 获取锁");
					// 测试锁重入
					lock2.acquire();
					System.out.println("线程 2 再次获取锁");
					Thread.sleep(5 * 1000);
					lock2.release();
					System.out.println("线程 2 释放锁");
					lock2.release();
					System.out.println("线程 2 再次释放锁");
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
		}).start();
	}
	// 分布式锁初始化
	public CuratorFramework getCuratorFramework () {
		//重试策略，初试时间 3 秒，重试 3 次
		RetryPolicy policy = new ExponentialBackoffRetry(3000, 3);
		//通过工厂创建 Curator
		CuratorFramework client =
		    CuratorFrameworkFactory.builder()
		    .connectString(connectString)
		    .connectionTimeoutMs(connectionTimeout)
		    .sessionTimeoutMs(sessionTimeout)
		    .retryPolicy(policy).build();
		//开启连接
		client.start();
		System.out.println("zookeeper 初始化完成...");
		return client;
	}
}

观察控制台变化

观察控制台变化：
线程 1 获取锁
线程 1 再次获取锁
线程 1 释放锁
线程 1 再次释放锁
线程 2 获取锁
线程 2 再次获取锁
线程 2 释放锁
线程 2 再次释放锁

大数据最新文章

实现Kafka至少消费一次

亚马逊云科技：还在苦于ETL？Zero ETL的时代

初探MapReduce

【SpringBoot框架篇】32.基于注解+redis实现

Elasticsearch：如何减少 Elasticsearch 集

Go redis操作

Redis面试题

专题五 Redis高并发场景

基于GBase8s和Calcite的多数据源查询

Redis——底层数据结构原理

加:2021-11-20 18:27:52 更:2021-11-20 18:29:08

360图书馆购物三丰科技阅读网日历万年历 2024年11日历

-2024/11/24 6:47:49-

图片自动播放器
↓图片自动播放器↓

TxT小说阅读器
↓语音阅读,小说下载,古典文学↓

一键清除垃圾
↓轻轻一点,清除系统垃圾↓

图片批量下载器
↓批量下载图片,美女图库↓

网站联系: qq:121756557 email:121756557@qq.com IT数码