实现一个极简rpc框架

Posted 2021-04-11 BIsland

一、RPC是什么

Remote Procedure Call，远过程调用

二、实现一个简单RPC

分析一下如何实现一个简单rpc的过程。

2.1 C/S模型

方法的请求者就是client方，方法的提供者就是我们的server方，两者通过网络通信来传递数据。

• client端。client方要使用服务，最终就是要使用方法，所以client至少需要提供：

• 调用方法的类，即谁来调用方法；
• 方法名，方法参数，即确定调用哪一个方法；

• server端。server端主要就是提供方法调用的服务，所以需要有以下的职责：

• 注册可用方法，要提供方法首先就要有方法；

• 启动rpc服务，即监听来自用户的网络请求，做出相应方法的调用；

  public interface ServiceCenter {
      // 开始监听网络端口，提供服务
      void start() throws IOException;
      
      // 停止服务
      void stop();
      
      // 注册服务接口及对应实现
      void register(Class<?> serviceInterface, Class<?> impl); 
      
      // 向外暴露服务端口
      int getPort();
  }
  

2.2 网络通信

远程的通信需要通过网络来实现，网络的传输本质上就是字节序列的搬运，所以无论是客户端发出的请求还是服务端返回的服务，都需要序列化成字节序列发送。对象变成字节序列使用的是序列化方法，反之从字节序列还原一个对象使用的就是反射的方式了。

为了能够很好的解释来自客户端的rpc请求到底是什么含义，双方都需要对请求的结构达成一个协议：

public class RpcRequest implements Serializable {
    private String serviceName; // 服务对象的名称
    private String methodName;  // 要使用服务对象的哪个方法
    private Class<?>[] paraTypes;   // 方法参数类型
    private Object[] params;    // 方法参数列表
    
    // constructor, getters and setters ...
}

2.3 server服务提供

服务中心启动之后的主要任务就是：监听网络请求，获取到请求后，根据请求的参数从注册中心中找到方法并调用，接着通过网络返回结果。所以服务中心服务提供的主要逻辑如下：

public void start() throws IOException {
    // 1. 服务中心拥有了socket才打开了网络的通信的门
    try (ServerSocket serverSocket = new ServerSocket()) {
        serverSocket.bind(new InetSocketAddress(port)); // socket = (ip地址，端口)
        System.out.println("Service center start :)");
        while (true) {
            // 2. 阻塞等待客户连接
            // 3. 接收到tcp请求就交给一个线程进行处理
            threadPool.execute(new ServiceTask(serverSocket.accept()));
        }
    }
}

其中ServiceTask就是rpc请求处理的封装，里边的主要逻辑是：

• 通过socket获取tcp的字节流输入，进而得到一个rpc请求对象；
• 根据rpc请求的各个参数从注册中心获取服务对象；
• 服务对象调用方法得到结果，再通过socket传回给客户端。

// 最终分配给线程的是可运行的任务
// 这里就将线程到底要干什么封装
private class ServiceTask implements Runnable {
    // 线程的任务就是要服务远程的客户
    // 网络通信就必须获取ip地址和端口，socket就是这两者的封装
    Socket clientSocket;

    public ServiceTask(Socket client) {
        this.clientSocket = client;
    }

    @Override
    public void run() {
        // TCP是字节流的传输，所以对象必然是序列化成字节码之后发过来
        // 所以需要获取对象的流输入输出
        try (ObjectInputStream ins = new ObjectInputStream(clientSocket.getInputStream());
             ObjectOutputStream outs = new ObjectOutputStream(clientSocket.getOutputStream())) {
            System.out.println("# Service center connected to client");
   
            // 1. 获取rpc请求对象的所有参数
            RpcRequest req = (RpcRequest) ins.readObject();
            String serviceName = req.getServiceName();
            String methodName = req.getMethodName();
            Class<?>[] paraTypes = req.getParaTypes();
            Object[] params = req.getParams();

            // 2. 根据参数从注册中心获取服务对象
            Class<?> serviceClass = registry.get(serviceName);
            if (serviceClass == null) {
                throw new ClassNotFoundException(serviceName + " not found in service center");
            }
            // 3. 获取服务对象的方法
            Method method = serviceClass.getMethod(methodName, paraTypes);
            // 4. 调用方法，获得结果
            Object result = method.invoke(serviceClass.newInstance(), params);
            // 5. 将结果返回给客户端
            outs.writeObject(result);

            System.out.println("# Service center do service and return result");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                if (clientSocket != null) {
                    clientSocket.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

2.4 client服务获取

客户端最终是通过反射字节码的方式得到服务的，可以使用JDK提供的动态代理方法来动态获取一个服务对象的代理，接着通过该代理对象来间接调用远程服务器的方法。所以与服务器的网络通信实际上在代理对象内部进行完成，客户只是把这个代理对象当作其服务的提供者。

public class RpcClient {
    private Socket socket;

    /**
     * 获取远程服务对象
     * @param serviceInterface 服务接口
     * @param serviceAddr   服务中心地址
     * @return  服务代理对象
     */
    public Object getRemoteServiceProxy(
        final Class<?> serviceInterface,
        final InetSocketAddress serviceAddr
    ) {
        // 使用jdk动态代理
        return Proxy.newProxyInstance(
            serviceInterface.getClassLoader(),
            new Class<?>[]{serviceInterface},
            (proxy, method, args) -> {  // 定义代理对象要干啥
                System.out.println("> Proxy do work");
                // 1. 创建socket，与服务器建立连接
                socket = new Socket();
                socket.connect(serviceAddr);

                System.out.println("> Socket connected");

                // 2. 将服务请求序列化成字节流发送给服务中心
                try (ObjectOutputStream outs = new ObjectOutputStream(socket.getOutputStream());
                     ObjectInputStream ins = new ObjectInputStream(socket.getInputStream())) {

                    outs.writeObject(new RpcRequest(
                        serviceInterface.getName(),
                        method.getName(),
                        method.getParameterTypes(),
                        args
                    ));

                    System.out.println("> Proxy sent RPC request");

                    // 3. 阻塞等待服务中心返回服务结果
                    return ins.readObject();
                } finally {
                    if (socket != null) {
                        socket.close();
                    }
                }
            }
        );
    }
}

至此，一个极简的rpc框架就完成了！跑起来试试：

完整源码：https://gitee.com/bankarian/prepare/tree/master/network/src/rpc