[网络协议]自定义 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:2181
    private 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() {
            @Override
            public 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.UserService
        String 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);
            //获取服务端ip
            String 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/rpc
            List<String> children = connection.getConnection().getChildren("/",false);
            //不存在子节点/shen
            if(!children.contains("shen")){
                //创建节点/shen
                connection.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);
                //遍历集合services
                for (Object key: services.keySet()) {
                    //通过key查询value
                    Object 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);
    }

}

后续工作

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

创作挑战赛

新人创作奖励来咯，坚持创作打卡瓜分现金大奖