10分钟实现简易RPC – 汪明鑫的个人博客

Rpc调用流程图

实现一版简易的RPC,揭开RPC的面纱，感受一波

首先要定义我们的Rpc方法

package pers.wmx.springbootfreemarkerdemo.rpc.server;

/**
 * @author wangmingxin03
 * Created on 2021-12-02
 */
public interface SayHelloRpcService {
    public String hello(String clientName);
}

package pers.wmx.springbootfreemarkerdemo.rpc.server;

/**
 * @author wangmingxin03
 * Created on 2021-12-02
 */
public class SayHelloRpcServiceImpl implements SayHelloRpcService {
    @Override
    public String hello(String clientName) {
        return "hello, " + clientName;
    }
}

然后我们需要把定义的服务发布出去，让上游服务可以通过网络远程调用

package pers.wmx.springbootfreemarkerdemo.rpc.server;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.ServerSocket;
import java.net.Socket;

/**
 * @author wangmingxin03
 * Created on 2021-12-02
 */
public class RpcServer {
    /**
     * 发布服务
     *
     * @param service 要发布的服务
     * @param port server端口号
     **/
    public void publishService(Object service, int port) throws Exception {
        if (service == null) {
            throw new IllegalArgumentException("empty service");
        }

        if (port <= 0) {
            throw new IllegalArgumentException("invalid port");
        }

        ServerSocket serverSocket = new ServerSocket(port);
        while (true) {
            Socket socket = serverSocket.accept();
            new Thread(() -> {
                try {

                    // 获取Socket输入输出流
                    ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                    ObjectOutputStream outputStream = new ObjectOutputStream(socket.getOutputStream());

                    try {
                        String methodName = inputStream.readUTF();
                        Class[] parameterTypes = (Class[]) inputStream.readObject();
                        Object[] arguments = (Object[]) inputStream.readObject();

                        Method method = service.getClass().getMethod(methodName, parameterTypes);
                        Object response = method.invoke(service, arguments);

                        outputStream.writeObject(response);
                    } catch (Throwable t) {
                        // 把异常抛给服务调用方
                        outputStream.writeObject(t);
                    }

                } catch (Exception e) {
                    e.printStackTrace();
                }

            }).start();
        }
    }
}

实际上就是起一个Sockct

等待客户端连接发起远程调用

启动Rpc服务的入口

package pers.wmx.springbootfreemarkerdemo.rpc.server;

/**
 * @author wangmingxin03
 * Created on 2021-12-02
 */
public class RpcServerMain {
    public static void main(String[] args) throws Exception {
        SayHelloRpcService sayHelloRpcService = new SayHelloRpcServiceImpl();
        RpcServer rpcServer = new RpcServer();
        rpcServer.publishService(sayHelloRpcService, 6666);
    }
}

通过反射完成下面的操作

拿到调用方的方法名

String methodName = inputStream.readUTF();

再通过调用方的参数类型，找到接口中指定的方法

Class[] parameterTypes = (Class[]) inputStream.readObject();

再读到调用发传的参数

Object[] arguments = (Object[]) inputStream.readObject();

有了方法和参数就可以发起调用

Method method = service.getClass().getMethod(methodName, parameterTypes);

Object response = method.invoke(service, arguments);

最后把结果写回调用方

写一个客户端发起调用

package pers.wmx.springbootfreemarkerdemo.rpc.client;

import pers.wmx.springbootfreemarkerdemo.rpc.server.SayHelloRpcService;

/**
 * @author wangmingxin03
 * Created on 2021-12-02
 */
public class RpcClientMain {
    public static void main(String[] args) throws Exception {
        RpcClient client = new RpcClient();
        SayHelloRpcService service = client
                .callService(SayHelloRpcService.class, "127.0.0.1", 6666);

        String request = "xinye";
        String response = service.hello(request);
        System.out.println(response);
    }
}

package pers.wmx.springbootfreemarkerdemo.rpc.client;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Proxy;
import java.net.Socket;

/**
 * @author wangmingxin03
 * Created on 2021-12-02
 */
public class RpcClient {
    public <T> T callService(Class<T> rpcServiceClass, String host, int port) throws Exception {
        return (T) Proxy.newProxyInstance(rpcServiceClass.getClassLoader(),
                new Class<?>[] {rpcServiceClass}, (proxy, method, args) -> {
                    // 创建客户端到服务端的连接
                    Socket socket = new Socket(host, port);

                    ObjectOutputStream outputStream = null;
                    ObjectInputStream inputStream = null;

                    try {
                        // 拿到输出流，开始调RPC
                        outputStream = new ObjectOutputStream(socket.getOutputStream());
                        outputStream.writeUTF(method.getName());
                        outputStream.writeObject(method.getParameterTypes());
                        outputStream.writeObject(args);

                        // 拿到输入流，获取调用结果
                        inputStream = new ObjectInputStream(socket.getInputStream());
                        Object response = inputStream.readObject();

                        if (response instanceof Throwable) {
                            throw (Throwable)response;
                        }
                        return response;
                    } catch (Exception e) {
                        e.printStackTrace();
                        return "";
                    } finally {
                        inputStream.close();
                        outputStream.close();
                        socket.close();
                    }

                });
    }
}

Proxy.newProxyInstance

这里用了代理模式

基于定义的接口生成一个代理对象

对代码对象操作就会发起远程调用

把方法、方法参数类型、参数都发给服务端，服务端就能找到对应的实现类做本地调用，

再把结果返回完事

debug 调用方发向Socket的数据