手把手教你写一个RPC

1.1 RPC 是什么

定义：RPC（Remote Procedure Call Protocol）——远程过程调用协议 ，RPC协议假定某些传输协议的存在，如TCP或UDP，为通信程序之间携带信息数据。在OSI网络通信模型中，RPC跨越了传输层和应用层 ，RPC使得开发包括网络分布式多程序在内的应用程序更加容易。

成都创新互联公司是一家专业提供成都企业网站建设,专注与网站制作、做网站、H5网站设计、小程序制作等业务。10年已为成都众多企业、政府机构等服务。创新互联专业网络公司优惠进行中。

我的理解：与其说把RPC 看作是一种协议，倒不如把它看作是一种客户机/服务器交互的模式，但是 RPC一定是基于 TCP 或者其他通信协议的

下面我们来看一下一个RPC调用的流程涉及哪些通信细节：

手把手教你写一个RPC

服务消费方（client）调用以本地调用方式调用服务；(1)
client stub接收到调用后负责将方法、参数等组装成能够进行网络传输的消息体；(2)
client stub找到服务地址，并将消息发送到服务端；(3)
server stub收到消息后进行解码；(4)
server stub根据解码结果调用本地的服务；(5)
本地服务执行并将结果返回给server stub；(6)
server stub将返回结果打包成消息并发送至消费方；(7)
client stub接收到消息，并进行解码；(8)
服务消费方得到最终结果。(9)

RPC的目标就是要2~8这些步骤都封装起来，让用户对这些细节透明。

1.2 手动实现

1.2.1 先做一个空接口实现序列化接口

public interface IRpcService extends Serializable{

}

1.2.2 做一个需要被远程调用的接口以及对应的接口实现类

public interface IHelloService extends IRpcService{
    String sayHi(String name,String message); 
}

public class HelloServiceImpl implements IHelloService{

    private static final long serialVersionUID = 146468468464364698L;

    @Override
    public String sayHi(String name, String message) {
        return new StringBuilder().append("hi~!").append(",").append(message).toString();
    }

}

1.2.3 需要写一个服务端，主要的作用是进行服务注册(接口注册) 以及接收客户端的调用参数执行调用请求返回结果

注：这个地方我没有采用dom4j 解析配置文件的形式进行接口注册有时间的朋友可以多加一层

public interface Server {

    //Socket端口
    int PORT = 8080;

    //启动服务端
    void start() throws IOException;

    //停止服务端
    void stop();

    /**
     * 服务注册
     * -- serviceInterface 对外暴露接口
     * -- 内部实现类
     */
    void regist(Class serviceInterface,Class impl);

}

public class ServerCenter implements Server{

    /**线程池 接收客户端调用**/
    private static ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 20, 200, TimeUnit.MILLISECONDS,new ArrayBlockingQueue(10));
    /**服务注册缓存**/
    public static final Map> serviceRegistry = new HashMap<>();

    /**
     * 启动服务
     */
    @Override
    public void start() throws IOException {
        ServerSocket server = new ServerSocket();
        server.bind(new InetSocketAddress(PORT));
        try {
            while(true){
                executor.execute(new ServiceTask(server.accept()));
            }
        } finally {
            server.close();
        }
    }

    /**
     * 停止服务
     */
    @Override
    public void stop() {
        executor.shutdown();
    }

    /**
     * 注册服务
     */
    @Override
    public void regist(Class serviceInterface, Class impl) {
        serviceRegistry.put(serviceInterface.getName(), impl);
    }

    private static class ServiceTask implements Runnable{

        Socket client = null;

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

        @Override
        public void run() {
            ObjectInputStream input = null;
            ObjectOutputStream output = null;

            try {

                input = new ObjectInputStream(client.getInputStream());
                String serviceName = input.readUTF();
                String methodName = input.readUTF();

                Class[] parameterTypes = (Class[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();
                Class serviceClass = serviceRegistry.get(serviceName);
                if(serviceClass == null){
                    throw new ClassNotFoundException(serviceName + "not found");
                }
                Method method = serviceClass.getMethod(methodName, parameterTypes);
                Object result = method.invoke(serviceClass.newInstance(), arguments);

                //将执行结果反序列化 通过socket返给客户端
                output = new ObjectOutputStream(client.getOutputStream());
                output.writeObject(result);

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

                if(input != null){
                    try {
                        input.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

                if(output != null){
                    try {
                        output.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

                if(client != null){
                    try {
                        client.close();
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

            }
        }

    }

    public static void main(String[] args) throws Exception {
        ServerCenter center = new ServerCenter();
        center.regist(IHelloService.class,new HelloServiceImpl().getClass());
        center.start();
    }
}

1.2.4 写一个客户端，用动态代理获取被代理接口的各种参数传输给服务端，接收返回结果，打印到控制台

public class Client {

    @SuppressWarnings("unchecked")
    public static T getRemoteProxyObj(final Class serviceInterface,final InetSocketAddress addr){
        return (T) Proxy.newProxyInstance(serviceInterface.getClassLoader(), new Class[]{serviceInterface}, new InvocationHandler() {

            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                Socket socket = null;
                ObjectOutputStream output = null;
                ObjectInputStream input = null;
                try {
                    //1.创建Socket客户端，根据指定地址连接远程服务提供者
                    socket = new Socket();
                    socket.connect(addr);

                    //2.将远程服务调用所需的接口类、方法名、参数列表等编码后发送给服务提供者
                    output = new ObjectOutputStream(socket.getOutputStream());
                    output.writeUTF(serviceInterface.getName());
                    output.writeUTF(method.getName());
                    output.writeObject(method.getParameterTypes());
                    output.writeObject(args);

                    //3.同步阻塞等待服务器返回应答 获取应答后返回
                    input = new ObjectInputStream(socket.getInputStream());
                    return input.readObject();
                } finally{
                    if(socket != null){
                        socket.close();
                    }

                    if(output != null){
                        output.close();
                    }

                    if(input != null){
                        input.close();
                    }
                }
            }
        });
    }

}

1.2.5 测试

注：测试之前需要开启服务端

public class RpcTest {
    public static void main(String[] args) throws IOException {
        IHelloService service  = Client.getRemoteProxyObj(IHelloService.class, new InetSocketAddress(8080));
        System.out.println(service.sayHi("张三", "新年快乐!"));
    }
}

就这样我们实现了一个简陋的RPC

本文意在通过实现简单的RPC，去真正意义上对RPC框架的实现原理有初步的了解，而不是人云亦云。

此RPC实现有诸多缺点，但是我们只要明白RPC的基座其他的RPC框架只是完善基座以及扩展而已。

本文的重点是你有没有收获与成长，其余的都不重要，希望读者们能谨记这一点。同时我经过多年的收藏目前也算收集到了一套完整的学习资料，包括但不限于：分布式架构、高可扩展、高性能、高并发、Jvm性能调优、Spring，MyBatis，Nginx源码分析，redis，ActiveMQ、、Mycat、Netty、Kafka、MySQL、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多个知识点高级进阶干货，希望对想成为架构师的朋友有一定的参考和帮助