準(zhǔn)備工作

這次我們用RMI+Zookeeper實(shí)現(xiàn)一個(gè)遠(yuǎn)程調(diào)用的RPC框架，RMI實(shí)現(xiàn)遠(yuǎn)程調(diào)用，Zookeeper作為注冊(cè)中心，具體的操作之前的文章都提到過(guò)，這里不再做過(guò)多贅述。

自定義 RPC框架2——RMI實(shí)現(xiàn)RPC

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

自定義 RPC框架3——JAVA實(shí)現(xiàn)Zookeeper節(jié)點(diǎn)增刪改查

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

代碼實(shí)現(xiàn)

項(xiàng)目結(jié)構(gòu)

• ZkConnection負(fù)責(zé)創(chuàng)建一個(gè)zookeeper對(duì)象并返回

• ShenRpcRegistry

  • 提供注冊(cè)服務(wù)的方法，將服務(wù)節(jié)點(diǎn)保存到Zookeeper并注冊(cè)到Registry中，
  • 提供一個(gè)獲取服務(wù)的方法，從Zookeeper中獲取節(jié)點(diǎn)，根據(jù)查詢的結(jié)果創(chuàng)建一個(gè)代理對(duì)象返回

• ShenRpcFactory框架入口

  • 使用該框架需要兩個(gè)配置文件:shen.properties和shen-services.properties

  • 提供快速注冊(cè)服務(wù)方法和快速獲取代理對(duì)象方法，以及基本的連接和批量注冊(cè)

  • shen.properties里面寫響應(yīng)的的配置:

    • registry.ip=服務(wù)器IP地址，默認(rèn)為localhost

    • registry.port=服務(wù)端端口號(hào)，默認(rèn)為9090

    • zk.server=Zookeeper訪問(wèn)地址，默認(rèn)為localhost:2181

    • zk.sessionTimeout=Zookeeper連接會(huì)話超時(shí)，默認(rèn)為10000

    • 例如：

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

  • shen-service.properties里面寫需要批量注冊(cè)的服務(wù)，例如：

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

導(dǎo)入POM依賴

導(dǎo)入zookeeper依賴的同時(shí)，要排除其中的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;//保存會(huì)話超時(shí)時(shí)間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監(jiān)聽");}});} }

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;//注冊(cè)器工具 //通過(guò)zk連接對(duì)象，和傳入的Remote接口實(shí)現(xiàn)對(duì)象，完成RMI地址的拼接，和保存（保存在zk中） //缺少LocateRegistry對(duì)象，缺少當(dāng)前類型中屬性賦值過(guò)程，整體流程，缺少zkconnection的創(chuàng)建過(guò)程 public class ShenRpcRegistry {//連接對(duì)象private ZkConnection connection;private String ip;private int port;/*** 注冊(cè)服務(wù)的方法* 1，拼接RMI的地址URI* 2，把訪問(wèn)地址URI存儲(chǔ)在zookeeper中* @param serviceInterface-服務(wù)接口類的對(duì)象，如com.shen.service.UserService.class* 接口必須是Remote接口的子接口* @param remote-f服務(wù)實(shí)現(xiàn)類型的對(duì)象如：new com.shen.service.impl.UserServiceImpl* 實(shí)現(xiàn)類型，必須實(shí)現(xiàn)serviceInterface,且是Remote接口直接或間接實(shí)現(xiàn)類* @throws Exception 拋出異常代表注冊(cè)失敗*/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();//拼接一個(gè)有規(guī)則的zk存儲(chǔ)節(jié)點(diǎn)命名String path = "/shen/rpc/" + serviceInterface.getName();//如果節(jié)點(diǎn)已存在，則刪除重建List<String> children = connection.getConnection().getChildren("/shen/rpc",false);if(children.contains(serviceInterface.getName())){//節(jié)點(diǎn)存在，需要?jiǎng)h除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);//把服務(wù)對(duì)象，在RMI的Registry中注冊(cè)Naming.rebind(rmi,remote);}/*** 根據(jù)服務(wù)接口類型，訪問(wèn)zk,獲取RMI的遠(yuǎn)程代理對(duì)象* 1,拼接一個(gè)zk中的節(jié)點(diǎn)名稱* 2，訪問(wèn)zk，查詢節(jié)點(diǎn)中存儲(chǔ)的數(shù)據(jù)* 3，根據(jù)查詢的結(jié)果，創(chuàng)建一個(gè)代理對(duì)象* @return*/public <T extends Remote> T getServiceProxy(Class<T> serviceInterface) throws IOException, InterruptedException, KeeperException, NotBoundException {//拼接zk中的節(jié)點(diǎn)名稱String path = "/shen/rpc/" + serviceInterface.getName();//查詢節(jié)點(diǎn)中存儲(chǔ)的數(shù)據(jù)byte[] datas = connection.getConnection().getData(path,false,null);//把查詢到的字節(jié)數(shù)組，翻譯成RMI的訪問(wèn)地址String rmi = new String(datas);//創(chuàng)建代理對(duì)象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();//連接對(duì)象private static final ZkConnection connection;//注冊(cè)器對(duì)象private static final ShenRpcRegistry registry;//用于讀取初始化的配置對(duì)象private static final Properties services = new Properties();/*** 初始化過(guò)程* 固定邏輯，在classpath下，提供配置文件，命名為，shen.properties* 配置文件結(jié)構(gòu)固化：* registry.ip=服務(wù)器IP地址，默認(rèn)為localhost* registry.port=服務(wù)端端口號(hào)，默認(rèn)為9090* zk.server=Zookeeper訪問(wèn)地址，默認(rèn)為localhost:2181* zk.sessionTimeout=Zookeeper連接會(huì)話超時(shí)，默認(rèn)為10000*/static {try {//獲取classpath類路徑下的配置文件輸入流InputStream input = ShenRpcFactory.class.getClassLoader().getResourceAsStream("shen.properties");//讀取配置文件，初始化配置對(duì)象config.load(input);//獲取服務(wù)端ipString serverIp = config.getProperty("registry.ip") == null ? "localhost" : config.getProperty("registry.ip");//獲取服務(wù)端端口號(hào)int serverPort = config.getProperty("registry.port") == null ?9090 : Integer.parseInt(config.getProperty("registry.port"));//獲取zookeeper服務(wù)器地址String zkServer = config.getProperty("zk.server") == null ? "localhost:2181" : config.getProperty("zk.server");//獲取zookeeper連接會(huì)話超時(shí)時(shí)長(zhǎng)int zkSessionTimeout = config.getProperty("zk.sessionTimeout") == null ?10000 : Integer.parseInt(config.getProperty("zk.sessionTimeout"));//創(chuàng)建連接對(duì)象connection = new ZkConnection(zkServer,zkSessionTimeout);//創(chuàng)建注冊(cè)器對(duì)象registry = new ShenRpcRegistry();//初始化注冊(cè)器對(duì)象屬性registry.setIp(serverIp);registry.setConnection(connection);registry.setPort(serverPort);//創(chuàng)建一個(gè)RMI的注冊(cè)器LocateRegistry.createRegistry(serverPort);//初始化zk中的父節(jié)點(diǎn)/shen/rpcList<String> children = connection.getConnection().getChildren("/",false);//不存在子節(jié)點(diǎn)/shenif(!children.contains("shen")){//創(chuàng)建節(jié)點(diǎn)/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下，是否有一個(gè)配置文件，命名為：shen-services.properties//如果有這個(gè)配置，則自動(dòng)初始化，沒有忽略后續(xù)邏輯//配置文件的格式是：接口全命名=實(shí)現(xiàn)類全命名InputStream servicesInout = ShenRpcFactory.class.getClassLoader().getResourceAsStream("shen-services.properties");if(servicesInout != null){//有配置，初始化services.load(servicesInout);//遍歷集合servicesfor (Object key: services.keySet()) {//通過(guò)key查詢valueObject value = services.get(key);//key是接口的全命名，value是實(shí)現(xiàn)類的全命名Class<Remote> serviceInterface = (Class<Remote>) Class.forName(key.toString());Remote serviceObject = (Remote) Class.forName(value.toString()).newInstance();//有個(gè)接口的類對(duì)象和服務(wù)的對(duì)象，注冊(cè)registry.registerService(serviceInterface,serviceObject);}}}catch (Exception e){e.printStackTrace();//當(dāng)初始化代碼塊發(fā)生異常問(wèn)題，拋出錯(cuò)誤，中斷虛擬機(jī)throw new ExceptionInInitializerError(e);}}//提供一個(gè)快速注冊(cè)服務(wù)和創(chuàng)建客戶端代理對(duì)象的靜態(tài)工具方法public static void registerSercice(Class<? extends Remote> serviceInterface,Remote remote) throws IOException, InterruptedException, KeeperException {registry.registerService(serviceInterface,remote);}//提供一個(gè)快速獲取代理對(duì)象的靜態(tài)工具方法public static <T extends Remote> T getServiceProxy(Class<T> serviceInterface) throws IOException, InterruptedException, KeeperException, NotBoundException{return registry.getServiceProxy(serviceInterface);}}

后續(xù)工作

后面我會(huì)在博客里寫兩個(gè)使用該框架的案例

總結(jié)

以上是生活随笔為你收集整理的自定义 RPC框架4——RMI+Zookeeper实现RPC框架的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。