基于Zookeeper服务注册和发现

素颜马尾好姑娘i 2022-07-13 15:25 236阅读 0赞

1、项目的目录结构

![20170114153550425][]

2、基于zk服务注册和发现的架构图

![20170114154232276][]

3、服务端(像zk提供服务的访问地址)

package cn.zk.distribute;
    
    import org.apache.zookeeper.CreateMode;
    import org.apache.zookeeper.KeeperException;
    import org.apache.zookeeper.ZooDefs.Ids;
    import org.apache.zookeeper.ZooKeeper;
    
    public class DistributedSystemServer {
    
    	private ZooKeeper zk = null;
    
    	private void getZkClient() throws Exception {
    
    		// 服务器在需求中并不需要做任何监听
    		zk = new ZooKeeper(GlobalConstants.zkhosts,
    				GlobalConstants.sessionTimeout, null);
    
    	}
    
    	/**
    	 * 向zookeeper中的/servers下创建子节点
    	 * 
    	 * @throws InterruptedException
    	 * @throws KeeperException
    	 */
    	private void connectZK(String serverName, String port) throws Exception {
    
    		// 先创建出父节点
    		if (zk.exists(GlobalConstants.parentZnodePath, false) == null) {
    			zk.create(GlobalConstants.parentZnodePath, null,
    					Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
    		}
    
    		// 连接zk创建znode
    		zk.create(GlobalConstants.parentZnodePath + "/",
    				(serverName + ":" + port).getBytes(), Ids.OPEN_ACL_UNSAFE,
    				CreateMode.EPHEMERAL_SEQUENTIAL);
    		System.out.println("server " + serverName + " is online ......");
    
    	}
    
    	// 服务器的具体业务处理功能
    	private void handle(String serverName) throws Exception {
    		System.out.println("server " + serverName
    				+ " is waiting for task process......");
    		Thread.sleep(Long.MAX_VALUE);
    
    	}
    
    	public static void main(String[] args) throws Exception {
    
    		DistributedSystemServer server = new DistributedSystemServer();
    
    		// 获取与zookeeper通信的客户端连接
    		server.getZkClient();
    
    		// 一启动就去zookeeper上注册服务器信息，参数1： 服务器的主机名 参数2：服务器的监听端口
    		server.connectZK(args[0], args[1]);
    
    		// 进入业务逻辑处理流程
    		server.handle(args[0]);
    
    	}
    
    }

2、客户端(获取zk提供的服务地址，并调用服务)

package cn.zk.distribute;
    
    import java.util.ArrayList;
    import java.util.List;
    
    import org.apache.zookeeper.WatchedEvent;
    import org.apache.zookeeper.Watcher;
    import org.apache.zookeeper.ZooKeeper;
    import org.apache.zookeeper.Watcher.Event.EventType;
    
    public class DistributedSystemClient {
    
    	private volatile List<String> servers = null;
    	private ZooKeeper zk = null;
    
    	// 获取zk连接
    	private void getZkClient() throws Exception {
    
    		// 服务器在需求中并不需要做任何监听
    		zk = new ZooKeeper(GlobalConstants.zkhosts,
    				GlobalConstants.sessionTimeout, new Watcher() {
    
    					@Override
    					public void process(WatchedEvent event) {
    
    						if (event.getType() == EventType.None)
    							return;
    
    						try {
    							// 获取新的服务器列表,重新注册监听
    							updateServers();
    
    						} catch (Exception e) {
    
    							e.printStackTrace();
    						}
    
    					}
    				});
    
    	}
    
    	/**
    	 * 从zk中获取在线服务器信息
    	 */
    	public void updateServers() throws Exception {
    
    		// 从servers父节点下获取到所有子节点，并注册监听
    		List<String> children = zk.getChildren(GlobalConstants.parentZnodePath,
    				true);
    
    		ArrayList<String> serverList = new ArrayList<String>();
    
    		for (String child : children) {
    
    			byte[] data = zk.getData(GlobalConstants.parentZnodePath + "/"
    					+ child, false, null);
    
    			serverList.add(new String(data));
    
    		}
    
    		// 如果客户端是一个多线程程序，而且各个线程都会竞争访问servers列表，所以，在成员中用volatile修饰了一个servers变量
    		// 而在更新服务器信息的这个方法中，是用一个临时List变量来进行更新
    		servers = serverList;
    
    		// 将更新之后的服务器列表信息打印在控制台观察一下
    		for (String server : serverList) {
    
    			System.out.println(server);
    		}
    		System.out.println("===================");
    
    	}
    
    	/**
    	 * 业务逻辑
    	 * 
    	 * @throws InterruptedException
    	 */
    	private void requestService() throws InterruptedException {
    		Thread.sleep(Long.MAX_VALUE);
    
    	}
    
    	public static void main(String[] args) throws Exception {
    
    		DistributedSystemClient client = new DistributedSystemClient();
    
    		// 先构造一个zk的连接
    		client.getZkClient();
    
    		// 获取服务器列表
    		client.updateServers();
    
    		// 客户端进入业务流程，请求服务器的服务
    		client.requestService();
    
    	}
    
    }

3、用到的常量配置信息

package cn.zk.distribute;
    
    public class GlobalConstants {
    	// zk服务器列表
    	public static final String zkhosts = "192.168.2.118:2181";
    	// 连接的超时时间
    	public static final int sessionTimeout = 2000;
    	// 服务在zk下的路径
    	public static final String parentZnodePath = "/servers";
    
    }

4、将服务端，导出为可以运行的jar文件

![20170114154323122][]

![20170114154333951][]

![20170114154345138][]

jar的运行

[root@localhost Desktop]# java -jar server.jar 192.168.2.11 4567

5、将客户端，导出为可以运行的jar文件

步骤与上面的4相同，可以参照上面的步骤过程。

7、源码下载

链接：http://pan.baidu.com/s/1eRHHGBK 密码：9d90

[20170114153550425]: /images/20220713/c502ce84d3044ed38224f00efeda2497.png
[20170114154232276]: /images/20220713/969e2c5ef8c84dcab47e94afceb3286d.png
[20170114154323122]: /images/20220713/081106eb585a495a90b868ef64363d09.png
[20170114154333951]: /images/20220713/9d2fd18bbb924363a5fe570cccd89256.png
[20170114154345138]: /images/20220713/169d2d31057e43328f5415dc667ecc8e.png