自己设计一个的轻量级的RPC框架–服务端zookeeper的发现和注册

#前言

ZooKeeper是一个分布式的，开放源码的分布式应用程序协调服务。
常用场景
1.数据发布与订阅（配置中心）
2. 命名服务
3. 分布式协调服务/通知
4. Master选举
5. 分布式锁

本项目主要用zookeeper的数据发布与订阅

ZooKeeper java使用方法

建立连接

@Override
	public void process(WatchedEvent event) {
		if(event.getState()==KeeperState.SyncConnected){
			System.out.println("连接成功");
			countDownLatch.countDown();
		}
	}
 
	public ZookeeperBase(String host) throws IOException, InterruptedException{
		this.zookeeper = new ZooKeeper(host, SESSION_TIME_OUT, this);
		countDownLatch.await();
	}

创建临时节点

//创建临时node 当服务挂的之后会自动吧node给删除 避免了服务宕机之后出现还能请求服务的情况
	public Boolean createNodeForTemporary(String path, String data) throws KeeperException, InterruptedException{
		path = this.pathChange(path);
		if(!this.nodeExists(path)) {
			String listPath[] = path.split("/");
			String prePath = "";
			for(int i=1; i<listPath.length-1; i++){
				prePath = prePath + "/" + listPath[i];
				if(!this.nodeExists(prePath)){
				//CreateMode.EPHEMERAL 创建临时节点
					this.zookeeper.create(prePath, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);	
				}
			}
			this.zookeeper.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
			return true;
		}else{
			return false;
		}
	}

创建临时

//创建node
	public Boolean createNode(String path, String data) throws KeeperException, InterruptedException{
		path = this.pathChange(path);
		if(!this.nodeExists(path)) {
			String listPath[] = path.split("/");
			String prePath = "";
			for(int i=1; i<listPath.length-1; i++){
				prePath = prePath + "/" + listPath[i];
				if(!this.nodeExists(prePath)){
					this.zookeeper.create(prePath, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);	
				}
			}
			this.zookeeper.create(path, data.getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			return true;
		}else{
			return false;
		}
	}

获取子节点

//获取子节点
	public List<String> getChilds(String path) throws KeeperException, InterruptedException{
		path = this.pathChange(path);
		if(this.nodeExists(path)){
			return this.zookeeper.getChildren(path, false);
		}else{
			return null;
		}
	}

删除节点

//由于zk节点下有子节点是不能直接删除的
public void rmr(String path) throws Exception {
        //获取路径下的节点
		path = this.pathChange(path);
        List<String> children = this.zookeeper.getChildren(path, false);
        for (String pathCd : children) {
            //获取父节点下面的子节点路径
            String newPath = "";
            //递归调用,判断是否是根节点
            if (path.equals("/")) {
                newPath = "/" + pathCd;
            } else {
                newPath = path + "/" + pathCd;
            }
            rmr(newPath);
        }
        //删除节点,并过滤zookeeper节点和 /节点
        if (path != null && !path.trim().startsWith("/zookeeper") && !path.trim().equals("/")) {
        	this.zookeeper.delete(path, -1);
            //打印删除的节点路径
            System.out.println("被删除的节点为：" + path);
        }
    }

ZooKeeper的使用

第一步标识将要被注册的服务

//将服务用 @RPCServer 用来表示该服务将被注册
@Service
@RPCServer
public class serverWorld2 {
	public String message(String world){
		return "Hello world";
	} 
}

第二步将建立连接将服务createnode

			代码比较多 剪切其中一块
			//连接zk
			ZookeeperBase zk = new ZookeeperBase(ZookeeperIpHost);
			//创建节点 获取当前项目路径
			Enumeration<URL> urls =Thread.currentThread().getContextClassLoader().getResources(baseackage.replace(".", "/"));
			while (urls.hasMoreElements()){
                URL url =  urls.nextElement();
                if(null != url){
                    String  protocol = url.getProtocol();
                    if(protocol.equals("file")){
                        String packagePath = url.getPath().replaceAll("%20"," ");//去空格
                       System.out.println("server"+packagePath);
                       File file = new File(packagePath);
                       //遍历目录将服务放入map中
                       func(file,baseackage);
                    }
                }
            }
			if(!zk.nodeExists("/RPCSERVER")){
				zk.createNode("/RPCSERVER", "ROOT");
			}
			Iterator<Map.Entry<String, String>> it = map.entrySet().iterator();
	        while (it.hasNext()) {
	             Map.Entry<String, String> entry = it.next();
	             if(!zk.nodeExists("/RPCSERVER/"+entry.getKey())){
	 				zk.createNode("/RPCSERVER/"+entry.getKey(), entry.getValue());
	 			 }
	             InetAddress address = InetAddress.getLocalHost();//获取的是本地的IP地址 //PC-20140317PXKX/192.168.0.121
	             String hostAddress = address.getHostAddress();//192.168.0.121     
	             if(!zk.nodeExists("/RPCSERVER/"+entry.getKey()+"/"+hostAddress+":"+rpc.getPort())){
	            	 //使用临时节点当zk断开连接的时候会自动消失
	            	 zk.createNodeForTemporary("/RPCSERVER/"+entry.getKey()+"/"+hostAddress+":"+rpc.getPort(),hostAddress+":"+rpc.getPort());
	            	 System.out.println("/RPCSERVER/"+entry.getKey()+"/"+hostAddress+":"+rpc.getPort()+" 创建");
	             }
	        }

第三步配置信息

<bean id="zkServer" class="main.java.zkServer.ZkServer" init-method="start" > 
		<property name="ZookeeperIpHost" value="172.16.12.34:2181"></property>
		<!-- 服务注册扫描包 -->
		<property name="baseackage" value="main.java.work.service"></property>
	</bean>