自定义 RPC框架4——RMI+Zookeeper实现RPC框架

准备工作

这次我们用RMI+Zookeeper实现一个远程调用的RPC框架，RMI实现远程调用，Zookeeper作为注册中心，具体的操作之前的文章都提到过，这里不再做过多赘述。

自定义 RPC框架2——RMI实现RPC

https://blog.csdn.net/qq_45587153/article/details/124211478?spm=1001.2014.3001.5502

自定义 RPC框架3——JAVA实现Zookeeper节点增删改查

https://blog.csdn.net/qq_45587153/article/details/124225572?spm=1001.2014.3001.5502

代码实现

项目结构

• ZkConnection负责创建一个zookeeper对象并返回

• ShenRpcRegistry

  • 提供注册服务的方法，将服务节点保存到Zookeeper并注册到Registry中，
  • 提供一个获取服务的方法，从Zookeeper中获取节点，根据查询的结果创建一个代理对象返回

• ShenRpcFactory框架入口

  • 使用该框架需要两个配置文件:shen.properties和shen-services.properties

  • 提供快速注册服务方法和快速获取代理对象方法，以及基本的连接和批量注册

  • shen.properties里面写响应的的配置:

    • registry.ip=服务器IP地址，默认为localhost

    • registry.port=服务端端口号，默认为9090

    • zk.server=Zookeeper访问地址，默认为localhost:2181

    • zk.sessionTimeout=Zookeeper连接会话超时，默认为10000

    • 例如：

    • registry.port=9999 zk.server=129.211.65.241:2181 zk.sessionTimeout=20000

  • shen-service.properties里面写需要批量注册的服务，例如：

    com.shen.service.UserService=com.shen.service.impl.UserServiceImpl com.shen.service.CustomService=com.shen.service.impl.CustomServiceImpl

导入POM依赖

导入zookeeper依赖的同时，要排除其中的logback依赖。

<dependency><groupId>org.apache.zookeeper</groupId><artifactId>zookeeper</artifactId><version>3.4.10</version><exclusions><exclusion><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId></exclusion><exclusion><groupId>ch.qos.logback</groupId><artifactId>logback-core</artifactId></exclusion></exclusions> </dependency>

ZkConnection

package com.shen.rpc.connection;import org.apache.zookeeper.WatchedEvent; import org.apache.zookeeper.Watcher; import org.apache.zookeeper.ZooKeeper;import java.io.IOException;//专门提供zookeeper连接的自定义类型 public class ZkConnection {//保存ZK的地址，格式是ip:port,如：129.211.65.241:2181private String zkServer;//保存会话超时时间private int sessionTimeout;public ZkConnection(){super();this.zkServer = "localhost:2181";this.sessionTimeout = 10000;}public ZkConnection(String zkServer,int sessionTimeout){this.zkServer = zkServer;this.sessionTimeout = sessionTimeout;}public ZooKeeper getConnection() throws IOException {return new ZooKeeper(this.zkServer, this.sessionTimeout, new Watcher() {@Overridepublic void process(WatchedEvent watchedEvent) {System.out.println("zookeeper监听");}});} }

ShenRpcRegistry

package com.shen.rpc.registry;import com.shen.rpc.connection.ZkConnection; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs; import org.apache.zookeeper.data.Stat;import java.io.IOException; import java.rmi.Naming; import java.rmi.NotBoundException; import java.rmi.Remote; import java.util.List;//注册器工具 //通过zk连接对象，和传入的Remote接口实现对象，完成RMI地址的拼接，和保存（保存在zk中） //缺少LocateRegistry对象，缺少当前类型中属性赋值过程，整体流程，缺少zkconnection的创建过程 public class ShenRpcRegistry {//连接对象private ZkConnection connection;private String ip;private int port;/*** 注册服务的方法* 1，拼接RMI的地址URI* 2，把访问地址URI存储在zookeeper中* @param serviceInterface-服务接口类的对象，如com.shen.service.UserService.class* 接口必须是Remote接口的子接口* @param remote-f服务实现类型的对象如：new com.shen.service.impl.UserServiceImpl* 实现类型，必须实现serviceInterface,且是Remote接口直接或间接实现类* @throws Exception 抛出异常代表注册失败*/public void registerService(Class<? extends Remote> serviceInterface, Remote remote) throws IOException, InterruptedException, KeeperException {//rmi = rmi://ip:port/com.shen.service.UserServiceString rmi = "rmi://" + ip + ":" + port + "/" + serviceInterface.getName();//拼接一个有规则的zk存储节点命名String path = "/shen/rpc/" + serviceInterface.getName();//如果节点已存在，则删除重建List<String> children = connection.getConnection().getChildren("/shen/rpc",false);if(children.contains(serviceInterface.getName())){//节点存在，需要删除Stat stat = new Stat();connection.getConnection().getData(path,false,stat);connection.getConnection().delete(path,stat.getCversion());}connection.getConnection().create(path,rmi.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);//把服务对象，在RMI的Registry中注册Naming.rebind(rmi,remote);}/*** 根据服务接口类型，访问zk,获取RMI的远程代理对象* 1,拼接一个zk中的节点名称* 2，访问zk，查询节点中存储的数据* 3，根据查询的结果，创建一个代理对象* @return*/public <T extends Remote> T getServiceProxy(Class<T> serviceInterface) throws IOException, InterruptedException, KeeperException, NotBoundException {//拼接zk中的节点名称String path = "/shen/rpc/" + serviceInterface.getName();//查询节点中存储的数据byte[] datas = connection.getConnection().getData(path,false,null);//把查询到的字节数组，翻译成RMI的访问地址String rmi = new String(datas);//创建代理对象Object obj = Naming.lookup(rmi);return (T) obj;}public ZkConnection getConnection() {return connection;}public void setConnection(ZkConnection connection) {this.connection = connection;}public String getIp() {return ip;}public void setIp(String ip) {this.ip = ip;}public int getPort() {return port;}public void setPort(int port) {this.port = port;} }

ShenRpcFactory

package com.shen.rpc;import com.shen.rpc.connection.ZkConnection; import com.shen.rpc.registry.ShenRpcRegistry; import org.apache.zookeeper.CreateMode; import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs;import java.io.IOException; import java.io.InputStream; import java.rmi.NotBoundException; import java.rmi.Remote; import java.rmi.registry.LocateRegistry; import java.util.List; import java.util.Properties;/*** 框架入口*/ public class ShenRpcFactory {//用于保存配置信息private static final Properties config = new Properties();//连接对象private static final ZkConnection connection;//注册器对象private static final ShenRpcRegistry registry;//用于读取初始化的配置对象private static final Properties services = new Properties();/*** 初始化过程* 固定逻辑，在classpath下，提供配置文件，命名为，shen.properties* 配置文件结构固化：* registry.ip=服务器IP地址，默认为localhost* registry.port=服务端端口号，默认为9090* zk.server=Zookeeper访问地址，默认为localhost:2181* zk.sessionTimeout=Zookeeper连接会话超时，默认为10000*/static {try {//获取classpath类路径下的配置文件输入流InputStream input = ShenRpcFactory.class.getClassLoader().getResourceAsStream("shen.properties");//读取配置文件，初始化配置对象config.load(input);//获取服务端ipString serverIp = config.getProperty("registry.ip") == null ? "localhost" : config.getProperty("registry.ip");//获取服务端端口号int serverPort = config.getProperty("registry.port") == null ?9090 : Integer.parseInt(config.getProperty("registry.port"));//获取zookeeper服务器地址String zkServer = config.getProperty("zk.server") == null ? "localhost:2181" : config.getProperty("zk.server");//获取zookeeper连接会话超时时长int zkSessionTimeout = config.getProperty("zk.sessionTimeout") == null ?10000 : Integer.parseInt(config.getProperty("zk.sessionTimeout"));//创建连接对象connection = new ZkConnection(zkServer,zkSessionTimeout);//创建注册器对象registry = new ShenRpcRegistry();//初始化注册器对象属性registry.setIp(serverIp);registry.setConnection(connection);registry.setPort(serverPort);//创建一个RMI的注册器LocateRegistry.createRegistry(serverPort);//初始化zk中的父节点/shen/rpcList<String> children = connection.getConnection().getChildren("/",false);//不存在子节点/shenif(!children.contains("shen")){//创建节点/shenconnection.getConnection().create("/shen",null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);}List<String> shenChildren = connection.getConnection().getChildren("/shen",false);if(!shenChildren.contains("rpc")){connection.getConnection().create("/shen/rpc",null,ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}//判断在classpath下，是否有一个配置文件，命名为：shen-services.properties//如果有这个配置，则自动初始化，没有忽略后续逻辑//配置文件的格式是：接口全命名=实现类全命名InputStream servicesInout = ShenRpcFactory.class.getClassLoader().getResourceAsStream("shen-services.properties");if(servicesInout != null){//有配置，初始化services.load(servicesInout);//遍历集合servicesfor (Object key: services.keySet()) {//通过key查询valueObject value = services.get(key);//key是接口的全命名，value是实现类的全命名Class<Remote> serviceInterface = (Class<Remote>) Class.forName(key.toString());Remote serviceObject = (Remote) Class.forName(value.toString()).newInstance();//有个接口的类对象和服务的对象，注册registry.registerService(serviceInterface,serviceObject);}}}catch (Exception e){e.printStackTrace();//当初始化代码块发生异常问题，抛出错误，中断虚拟机throw new ExceptionInInitializerError(e);}}//提供一个快速注册服务和创建客户端代理对象的静态工具方法public static void registerSercice(Class<? extends Remote> serviceInterface,Remote remote) throws IOException, InterruptedException, KeeperException {registry.registerService(serviceInterface,remote);}//提供一个快速获取代理对象的静态工具方法public static <T extends Remote> T getServiceProxy(Class<T> serviceInterface) throws IOException, InterruptedException, KeeperException, NotBoundException{return registry.getServiceProxy(serviceInterface);}}

后续工作

后面我会在博客里写两个使用该框架的案例

总结

以上是生活随笔为你收集整理的自定义 RPC框架4——RMI+Zookeeper实现RPC框架的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。