大数据学习之zookeeper案例节点动态上下线感知19

需求？

某分布式系统当中，主节点有多台，可以进行动态上下线，当有任何一台机器发生了动态的上下线 任何一台客户端都能感知得到。

思路？

（1）创建客户端与服务端

（2）启动client端 监听

（3）启动server端 注册

（4）当server端 发生上下线

（5）client端都能感知的到

代码编写：

服务端

package com.dawn.上下线感知;

import java.io.IOException;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;

/**
 * @author Dawn
 * @date 2019年5月17日18:08:46
 * @version 1.0
 * 创建服务端，这里服务端就只做了一个事情。创建节点（也就是这里的信息注册）
 */
public class ZkServer {
	
	public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
		//1.连接zkServer
		ZkServer zkServer = new ZkServer();
		zkServer.getConnect();
		
		//2.注册节点信息 服务器ip添加到zk中
		zkServer.regist(args[0]);
		
		//3.业务逻辑处理
		zkServer.build(args[0]);
	}
	
	
	private String connectString = "bigdata11:2181,bigdata12:2181,bigdata13:2181";
	private int sessionTimeout = 3000;
	ZooKeeper zkCli=null;
	//定义父节点
	private String parentNode = "/servers";
	
	//1：连接zkServer
	public void getConnect() throws IOException {
		zkCli = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
			}
		});
	}
	
	//2.注册信息
	public void regist(String hostname) throws KeeperException, InterruptedException {
		//创建临时带序列号的节点
		String node = zkCli.create(parentNode+"/server", hostname.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
		System.out.println(node);
	}
	
	//3.构造服务端
	public void build(String hostname) throws InterruptedException {
		System.out.println(hostname + ":服务器上线了！");
		
		Thread.sleep(Long.MAX_VALUE);
	}
	
	
	
}

　　

客户端：

package com.dawn.上下线感知;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

/**
 * @author Dawn
 * @date 2019年5月17日18:38:03
 * @version 1.0
 * 创建客服端，监听服务端servers节点下的信息。
 */
public class ZkClient {
	
	public static void main(String[] args) throws IOException, KeeperException, InterruptedException {
		//1.获取连接
		ZkClient zkClient = new ZkClient();
		zkClient.getConnect();
		
		//2.监听服务的节点信息
		zkClient.getServers();
		
		//3.业务逻辑（一直监听）
		zkClient.getWatch();
	}

	private String connectString = "bigdata11:2181,bigdata12:2181,bigdata13:2181";
	private int sessionTimeout = 3000;
	ZooKeeper zkCli=null;
	
	//1.连接集群
	public void getConnect() throws IOException {
		zkCli=new ZooKeeper(connectString, sessionTimeout, new Watcher() {
			
			@Override
			public void process(WatchedEvent event) {
				List<String> children;
				
				try {
					//监听父节点
					children = zkCli.getChildren("/servers", true);
				
					//创建集合存储服务器列表
					ArrayList<String> serverList = new ArrayList<>();
					
					//获取每个节点的数据
					for(String c:children) {
						zkCli.getData("/servers/"+c, true, null);
						serverList.add(c);
					}
					
					//打印服务器列表
					System.out.println(serverList);
				
				} catch (KeeperException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
				
			}
		});
	}
	
	//2.监听服务的节点信息
	public void getServers(){
		
		List<String> children=null;
		
		try {
		children = zkCli.getChildren("/servers", true);
		ArrayList<String> serverList = new ArrayList<String>();
		
		//获取每个节点的数据
		for(String c:children) {
			zkCli.getData("/servers/"+c, true, null);
			serverList.add(new String(c));
		}
		
		//打印服务器列表
		System.out.println(serverList);
		}catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	//3.业务逻辑
	public void getWatch() throws InterruptedException {
		Thread.sleep(Long.MAX_VALUE);
	}
}

　　

运行截图：