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

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

• ZooKeeper java使用方法
• 建立连接
• 创建临时节点
• 创建临时
• 获取子节点
• 删除节点

• 第一步标识将要被注册的服务
• 第二步将建立连接 将服务createnode
• 第三步配置信息

#前言

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>

ZooKeeper java 大致流程

1.将服务标识
2.读取配置文件的ip:host 将zookeeper建立连接
3.读取配置文件扫描的包 将这些包下的服务信息在zookeeper中建立