準備工作

這次我們用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框架的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。