RPC远程过程调用学习之路（一）：用最原始代码还原PRC框架

肖哥哥

发布于 2019-02-22 10:40:47

6440

发布于 2019-02-22 10:40:47

文章被收录于专栏：后台及大数据开发后台及大数据开发

RPC: Remote Procedure Call 远程过程调用，即业务的具体实现不是在自己系统中，需要从其他系统中进行调用实现，所以在系统间进行数据交互时经常使用。

rpc的实现方式有很多，可以通过http和tcp协议进行实现

通过http协议的主要有：

webService 可以参考我之前的博客 WebService 学习之路（一）：了解并使用webService

webService学习之路（二）：springMVC集成CXF快速发布webService

webService学习之路（三）：springMVC集成CXF后调用已知的wsdl接口

restful 可以参考我之前的博客 Restful 介绍及SpringMVC+restful 实例讲解

而今天要讲的是通过TCP协议实现的远程调用。

为啥已经掌握了webservice和restful等通过http协议实现rpc技术外，还要研究tcp协议实现rpc呢？

因为网络七层协议中，http位于tcp之上，而从传输上而言，越底层同等条件下传输速度更快

另外影响rpc调用的除了传输方式外，另一个就是序列化，而java的阻塞式IO往往成为瓶颈，所以这里设计到了NIO，

NIO知识点就多了，请自己搜索学习。

言归正传，今天不借助其他仁和框架，用简单的代码还原rpc的过程。

大致可以分为下面几部（先了解过程，再看代码更容易理解）：

书写好服务接口和实现，就是我们项目中的业务层，看着service，service.imp就熟悉了 o(∩_∩)o
把1写好的接口暴露给其他系统，以便调用
根据暴露了接口的地址和接口信息，进行调用

是不是感觉和调用本地的service一样，最终就是要达到的这个效果。

文采不好，就要开始贴代码了：

首先是写接口和接口的实现类，和平时看见的、写的没任何区别

接口定义

package com.xiaochangwei.rpc;

public interface RpcTestService {
    String testRpcCall(int count);
}

接口实现类

package com.xiaochangwei.rpc;

import java.text.SimpleDateFormat;
import java.util.Date;

public class RpcTestServiceImpl implements RpcTestService {

    public final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:sss");
    
    @Override
    public String testRpcCall(int count) {
        return dateFormat.format(new Date())+" 调用rpc次数为：" + count;
    }

}

然后就是rpc的精髓了，怎么把服务暴露给其他系统的

其实说白了就是使用java自带的

import java.net.ServerSocket; import java.net.Socket;

网络编程相关的东西socket和对应的IO，因为我们要向服务器发送请求，然后服务器又要返回数据给请求方，

import java.io.ObjectInputStream; import java.io.ObjectOutputStream;

同时还要用到反射，因为我们不可能每个接口实现类的暴露都去写一套方法吧，得共通化吧，先简单理解为泛型把代码中看见过Class T 吧 o(∩_∩)o

先简单理解为过程和socket调用过程一样吧：

根据约定的端口，服务端起一个ServerSocket，并一直监听该端口
监听到有请求时，server端通过inputStream取得请求的相关信息
根据请求信息调用相应方法处理，并返回结果

简易PRC框架

package com.xiaochangwei.rpc;

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

/**
 * rpc简易框架
 */
public class RpcFramework {

    /**
     * 暴露服务
     */
    @SuppressWarnings("resource")
    public static void export(final Object service, int port) throws Exception {
        if (service == null)
            throw new IllegalArgumentException("服务实例为空");
        if (port <= 0 || port > 65535)
            throw new IllegalArgumentException("端口号不正确");
        System.out.println("通过端口 [" + port +"] 暴露服务[" + service.getClass().getName() + "]" );
        
        ServerSocket server = new ServerSocket(port);
        while(true){
            try {
                final Socket socket = server.accept();
                new Thread(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            try {
                                ObjectInputStream inputStream = new ObjectInputStream(socket.getInputStream());
                                try {
                                    String methodName = inputStream.readUTF();
                                    Class<?>[] parameterTypes = (Class<?>[]) inputStream.readObject();
                                    Object[] arguments = (Object[]) inputStream.readObject();
                                    ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                                    try {
                                        Method method = service.getClass().getMethod(methodName,parameterTypes);
                                        Object result = method.invoke(service,arguments);
                                        output.writeObject(result);
                                    } catch (Throwable t) {
                                        output.writeObject(t);
                                    } finally {
                                        output.close();
                                    }
                                } finally {
                                    inputStream.close();
                                }
                            } finally {
                                socket.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }).start();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 引用服务
     */
    @SuppressWarnings("unchecked")
    public static <T> T refer(final Class<T> interfaceClass, final String host,
            final int port) throws Exception {
        if (interfaceClass == null)
            throw new IllegalArgumentException("接口为空");
        if (!interfaceClass.isInterface())
            throw new IllegalArgumentException(interfaceClass.getName() + " 不是接口");
        if (host == null || host.length() == 0)
            throw new IllegalArgumentException("主机地址为空");
        if (port <= 0 || port > 65535)
            throw new IllegalArgumentException("端口不正确" + port);
        System.out.println("从服务器[" + host + ":" + port + "]取得远程服务[" + interfaceClass.getName() + "]" );
        
        return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
                new Class<?>[] { interfaceClass }, new InvocationHandler() {
                    public Object invoke(Object proxy, Method method,Object[] arguments) throws Throwable {
                        Socket socket = new Socket(host, port);
                        try {
                            ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                            try {
                                output.writeUTF(method.getName());
                                output.writeObject(method.getParameterTypes());
                                output.writeObject(arguments);
                                ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                                try {
                                    Object result = input.readObject();
                                    if (result instanceof Throwable) {
                                        throw (Throwable) result;
                                    }
                                    return result;
                                } finally {
                                    input.close();
                                }
                            } finally {
                                output.close();
                            }
                        } finally {
                            socket.close();
                        }
                    }
                });
    }

    /**
     * 暴露RPC服务主调函数
     */
    public static void main(String[] args) throws Exception {
        RpcTestService rpcTestService = new RpcTestServiceImpl();
        RpcFramework.export(rpcTestService, 3125);
    }
}

指定其中的main后，我们的接口就通过指定的端口暴露给其他系统了

其他系统调用也很简单

package com.xiaochangwei.rpc;

public class RpcConsumer {
    public static void main(String[] args) throws Exception {
        RpcTestService rpcTestService = RpcFramework.refer(RpcTestService.class,"127.0.0.1", 3125);
        for (int i = 0; i < 10; i++) {
            String response = rpcTestService.testRpcCall(i);
            System.out.println(response);
            Thread.sleep(1000);
        }
    }
}

是不是感觉和本地调用一样 o(∩_∩)o

看下效果吧，先暴露接口，再调用