自己设计一个的轻量级的RPC框架--zookeeper服务上下线通知和watch机制

╰+攻爆jí腚メ 2022-03-12 19:38 229阅读 0赞

### 自己设计一个的轻量级的RPC框架--zookeeper服务上下线通知和watch机制 ###

*  前言
 *  watch机制
 *   *  watch注册
     *  watch触发
     *  watch事件类型
 *  服务的上下线通知
 *  效果

# 前言 #

这篇博客主要是因为自己如果服务端每次都去访问zookeeper获取一些服务的节点信息感觉zookeeper的压力比较大，所以打算把服务的信息缓存放在本地，这里就要设计到zookeeper的watch机制。（PS 自己也踩了坑所以记录一下）

# watch机制 #

Watch机制：Watch事件是一个一次性的触发器，当被设置了Watch的数据发生了改变的时候，则服务器将这个改变发送给设置了Watch的客户端，以便通知它们。  
例如：如果我RPC的服务有4个，如果某一个节点发生改变的话需要4个服务做出相对于的反应的话需要在这之前4个服务都对这个节点进行Watch的注册。

## watch注册 ##

注册方法 --因为Watch只会触发一次，所以触发玩之后我们需要再次注册  
1.getData 获取数据当这个数据发送改变的时候会触发watch事件  
2.exists 获取数据当这个数据发送改变的时候会触发watch事件  
3.getChildren 获取数据当这个节点的子节点数据发送改变的时候会触发watch事件

## watch触发 ##

触发事件  
1.create 创建节点  
2.delete 删除节点  
3.setData 设置节点内容

## watch事件类型 ##

1.EventType.NodeCreated 节点创建  
2.EventType.NodeDataChanged 节点数据发送改变  
3.EventType.NodeChildrenChaged 子节点发生改变  
4.EventType.NodeDeleted 节点删除

服务上下线 我用的是NodeChildrenChaged 这个类型进行判断。为什么不用前面2个呢，原因是我A服务(一台RPC服务)创建调用exists进行监听的话 确实会触发1事件对于A服务创建的watch事件，但是对于B服务来说(另一台台RPC服务)他没有对这个节点的注册事件,所以他不能触发watch事件。所以我在我在连接的时候先初始化服务列表 先把服务缓存至本地 并将每个节点利用getChildren进行watch注册,这样如果在节点下面注册服务的话,就会触发watch方法。

# 服务的上下线通知 #

缓存至本地的server服务 好像watch是异步触发的 我担心会有线程安全问题所以用了线程安全的线程不知道对不对 有待下次检测

public static ConcurrentHashMap<String, ConcurrentHashMap<String,CopyOnWriteArrayList<String>>> serviceMap = new ConcurrentHashMap<String,ConcurrentHashMap<String,CopyOnWriteArrayList<String>>>();

有些方法我自己封装了一下例如getChilds()方法其实就是getChildren()

if(EventType.None == eventType){
    			System.out.println("连接成功");
    			//初始化服务列表 先把服务缓存至本地
    			try {
    				intiServerMap("/RPCSERVER");
    			} catch (KeeperException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    			countDownLatch.countDown();
    		}else if(EventType.NodeChildrenChanged == eventType){//某节点的其子节点有变化
    			String paths[] = path.split("/");
    			ConcurrentHashMap<String,CopyOnWriteArrayList<String>> Map = ZkServer.serviceMap.get(paths[3]);
    			try {
    				//目前列表中数据
    				CopyOnWriteArrayList<String> childList = Map.get(new String(getData("/"+paths[2]+"/"+paths[3])));
    				//节点数据
    				List<String> childrens = getChilds("/"+paths[2]+"/"+paths[3]);//再次注册watch
    				//上线处理
    				if(childrens.size() !=0){
    					for (String child : childrens) {
    						if(!childList.contains(child)){
    							childList.add(child);
    							System.out.println("服务上线"+path+"/"+child);
    						}
    						
    					}
    				}
    				//下线处理
    				if(childList.size() != 0){
    					for (String str : childList) {
    						if(childrens.size() !=0){
    							for (String child : childrens) {
    								if(!child.equals(str)){
    									childList.remove(str);
    									System.out.println("服务下线"+path+"/"+child);
    								}
    							}
    						}else{
    							childList.remove(str);
    							System.out.println("服务下线"+path);
    						}
    					}
    				}
    			} catch (KeeperException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}

# 效果 #

开启A服务 B服务，关闭B服务，再次开发B服务 日志

服务上线/request/RPCSERVER/serverWorld/172.16.12.32:8886
    服务下线/request/RPCSERVER/serverWorld
    服务上线/request/RPCSERVER/serverWorld/172.16.12.32:8886