RPC通讯框架详解

Posted 2022-05-24 _瞳孔

RPC基本理论

从单机走向分布式，产生了很多分布式的通信方式

• 最古老也是最有效，并且永不过时的，TCP/UDP的二进制传输。事实上所有的通信方式归根结底都是TCP/UDP
• CORBA Common Object Request Broker Architecture：古老而复杂的，支持面向对象的通信协议
• web Service (SOA SOAP RDDI WSDL…)：基于http + xml的标准化Web API
• RestFul (Representational State Transfer)，回归简单化本源的Web API的事实标准
• RMI (Remote Method lnvocation)： Java内部的分布式通信协议
• JMS (Java Message Service)：JavaEE中的消息框架标准,为很多MQ所支持
• RPC (Remote Procedure Call)：远程方法调用，这只一个统称，重点在于方法调用（不支持对象的概念），具体实现甚至可以用RMl、RestFul等去实现。但一般不用，因为RMI不能跨语言, 而RestFul效率太低。多用手服务器集群间的通信，因此常使用更加高效短小精悍的传输模式以提高效率。

RPC：
进程间通信（IPC，Inter-Process Communication），指至少两个进程或线程间传送数据或信号的一些技术或方法。进程是计算机系统分配资源的最小单位。每个进程都有自己的一部分独立的系统资源，彼此是隔离的。为了能使不同的进程互相访问资源并进行协调工作，才有了进程间通信。这些进程可以运行在同一计算机上或网络连接的不同计算机上。 进程间通信技术包括消息传递、同步、共享内存和远程过程调用。 IPC是一种标准的Unix通信机制。

有两种类型的进程间通信(IPC)。

• 本地过程调用(LPC)：LPC用在多任务操作系统中，使得同时运行的任务能互相会话。这些任务共享内存空间使任务同步和互相发送信息。
• 远程过程调用(RPC)：RPC类似于LPC，Birrell 和 Nelson 在 1984 发表于 ACM Transactions on Computer Systems 的论文《Implementing remote procedure calls》对 RPC 做了经典的诠释。RPC 是指计算机 A 上的进程，调用另外一台计算机 B 上的进程，其中 A 上的调用进程被挂起，而 B 上的被调用进程开始执行，当值返回给 A 时，A 进程继续执行。调用方可以通过使用参数将信息传送给被调用方，而后可以通过传回的结果得到信息。而这一过程，对于开发人员来说是透明的。

远程过程调用采用客户机/服务器(C/S)模式。请求程序就是一个客户机，而服务提供程序就是一台服务器。和常规或本地过程调用一样，远程过程调用是同步操作，在远程过程结果返回之前，需要暂时中止请求程序。使用相同地址空间的低权进程或低权线程允许同时运行多个远程过程调用。

如果无法在一个进程内，甚至无法在一个计算机内通过本地调用的方式完成的需求，比如比如不同的系统间的通讯，甚至不同的组织间的通讯。由于计算能力需要横向扩展，需要在多台机器组成的集群上部署应用，这时候就可以使用RPC

RPC的核心并不在于使用什么协议。RPC的目的是让你在本地调用远程的方法，而对你来说这个调用是透明的，你并不知道这个调用的方法是部署哪里。通过RPC能解耦服务，这才是使用RPC的真正目的。RPC的原理主要用到了动态代理模式，至于http协议，只是传输协议而已。简单的实现可以参考spring remoting，复杂的实现可以参考dubbo。

简单的说

• RPC就是从一台机器（客户端）上通过参数传递的方式调用另一台机器（服务器）上的一个函数或方法（可以统称为服务）并得到返回的结果。
• RPC 会隐藏底层的通讯细节（不需要直接处理Socket通讯或Http通讯） RPC 是一个请求响应模型。客户端发起请求，服务器返回响应（类似于Http的工作方式） RPC 在使用形式上像调用本地函数（或方法）一样去调用远程的函数（或方法）。

RPC通信过程

默认socket通信。本地机器的RPC框架反序列化出执行结果，函数return这个结果

RPC 的基本操作

让我们看看本地过程调用是如何实现的。考虑下面的 C 语言的调用：

count = read(fd, buf, nbytes);

其中，fd 为一个整型数，表示一个文件。buf 为一个字符数组，用于存储读入的数据。 nbytes 为另一个整型数，用于记录实际读入的字节数。如果该调用位于主程序中，那么在调用之前堆栈的状态如图2(a)所示。为了进行调用，调用方首先把参数反序压入堆栈，即为最后一个参数先压入，如图2(b)所示。在 read 操作运行完毕后，它将返回值放在某个寄存器中，移出返回地址，并将控制权交回给调用方。调用方随后将参数从堆栈中移出，使堆栈还原到最初的状态。

RPC 背后的思想是尽量使远程过程调用具有与本地调用相同的形式。假设程序需要从某个文件读取数据，程序员在代码中执行 read 调用来取得数据。在传统的系统中， read 例程由链接器从库中提取出来，然后链接器再将它插入目标程序中。 read 过程是一个短过程，一般通过执行一个等效的 read 系统调用来实现。即，read 过程是一个位于用户代码与本地操作系统之间的接口。

虽然 read 中执行了系统调用，但它本身依然是通过将参数压入堆栈的常规方式调用的。如图2(b)所示，程序员并不知道 read 干了啥。

RPC 是通过类似的途径来获得透明性。当 read 实际上是一个远程过程时（比如在文件服务器所在的机器上运行的过程），库中就放入 read 的另外一个版本，称为客户存根（client stub）。这种版本的 read 过程同样遵循图2(b)的调用次序，这点与原来的 read 过程相同。另一个相同点是其中也执行了本地操作系统调用。唯一不同点是它不要求操作系统提供数据，而是将参数打包成消息，而后请求此消息发送到服务器，如图3所示。在对 send 的调用后，客户存根调用 receive 过程，随即阻塞自己，直到收到响应消息。

当消息到达服务器时，服务器上的操作系统将它传递给服务器存根（server stub）。服务器存根是客户存根在服务器端的等价物，也是一段代码，用来将通过网络输入的请求转换为本地过程调用。服务器存根一般先调用 receive ，然后被阻塞，等待消息输入。收到消息后，服务器将参数由消息中提取出来，然后以常规方式调用服务器上的相应过程（如图3所示）。从服务器角度看，过程好像是由客户直接调用的一样：参数和返回地址都位于堆栈中，一切都很正常。服务器执行所要求的操作，随后将得到的结果以常规的方式返回给调用方。以 read 为例，服务器将用数据填充 read 中第二个参数指向的缓冲区，该缓存区是属于服务器存根内部的。

调用完后，服务器存根要将控制权教会给客户发出调用的过程，它将结果（缓冲区）打包成消息，随后调用 send 将结果返回给客户。事后，服务器存根一般会再次调用 receive，等待下一个输入的请求。

客户机器接收到消息后，客户操作系统发现该消息属于某个客户进程（实际上该进程是客户存根，只是操作系统无法区分二者）。操作系统将消息复制到相应的缓存区中，随后解除对客户进程的阻塞。客户存根检查该消息，将结果提取出来并复制给调用者，而后以通常的方式返回。当调用者在 read 调用进行完毕后重新获得控制权时，它所知道的唯一事就是已经得到了所需的数据。它不指导操作是在本地操作系统进行，还是远程完成。

整个方法，客户方可以简单地忽略不关心的内容。客户所涉及的操作只是执行普通的（本地）过程调用来访问远程服务，它并不需要直接调用 send 和 receive 。消息传递的所有细节都隐藏在双方的库过程中，就像传统库隐藏了执行实际系统调用的细节一样。

概况来说，远程过程调用包含如下步骤：

• 客户过程以正常的方式调用客户存根；
• 客户存根生成一个消息，然后调用本地操作系统；
• 客户端操作系统将消息发送给远程操作系统；
• 远程操作系统将消息交给服务器存根；
• 服务器存根调将参数提取出来，而后调用服务器；
• 服务器执行要求的操作，操作完成后将结果返回给服务器存根；
• 服务器存根将结果打包成一个消息，而后调用本地操作系统；
• 服务器操作系统将含有结果的消息发送给客户端操作系统；
• 客户端操作系统将消息交给客户存根；
• 客户存根将结果从消息中提取出来，返回给调用它的客户存根。

以上步骤就是将客户过程对客户存根发出的本地调用转换成对服务器过程的本地调用，而客户端和服务器都不会意识到中间步骤的存在。

RPC 的主要好处是双重的。首先,程序员可以使用过程调用语义来调用远程函数并获取响应。其次,简化了编写分布式应用程序的难度,因为 RPC 隐藏了所有的网络代码存根函数。应用程序不必担心一些细节,比如 socket、端口号以及数据的转换和解析。在 OSI 参考模型,RPC 跨越了会话层和表示层。

以下是简单的RPC代码实现

Server端

package Server;

public interface EchoService 
    String echo(String ping);

package Server;

public class EchoServiceImpl implements EchoService
    @Override
    public String echo(String ping) 
        // TODO Auto-generated method stub
        return ping !=null?ping+"--> I am ok.":"I am bad.";
     

Exporter端

package Exporter;

import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * RPC服务端发布者
 * 作为服务端，监听客户端的TCP连接，接收到新的客户端连接之后，将其封装成Task，由线程池执行
 * 将客户端发送的码流反序列化成对象，反射调用服务实现者，获取执行结果
 * 将执行结果对象发序列化，通过Socket发送给客户端
 * 远程调用完成之后，释放Socket等连接资源，防止句柄泄露
 * @author Administrator
 *
 */
public class RpcExporter 
    //创建一个可重用固定线程数的线程池
    //Runtime.getRuntime().availableProcessors()返回虚拟机可用的处理器数量
    static Executor executor=Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    
    public static void exporter(String hostname,int port) throws IOException 
        //创建一个监听特定端口的Serversocket,负责接收客户连接请求
        ServerSocket server = new ServerSocket();
        //绑定主机名端口号
        server.bind(new InetSocketAddress(hostname,port));
        try
        while(true)
        
            executor.execute(new ExporterTask(server.accept()));
        
        finally
        
            server.close();
        
    
    
    private static class ExporterTask implements Runnable

        Socket client=null;
        public ExporterTask(Socket client)
            this.client=client;
        
        @Override
        public void run() 
            // TODO Auto-generated method stub
            ObjectInputStream input=null;
            ObjectOutputStream output=null;
            try
                //获取输入流
                input=new ObjectInputStream(client.getInputStream());
                //获取调用的接口名
                String interfaceName = input.readUTF();
                //加载接口
                Class<?> service = Class.forName(interfaceName);
                //获取调用的方法名
                String methodName = input.readUTF();
                //获取方法返回类型
                Class<?>[] ParameterTypes = (Class<?>[]) input.readObject();
                //获取参数
                Object[] arguments = (Object[]) input.readObject();
                //通过反射获取方法
                Method method = service.getMethod(methodName, ParameterTypes);
                //通过反射调用方法
                Object result = method.invoke(service.newInstance(), arguments);
                output = new ObjectOutputStream(client.getOutputStream());
                output.writeObject(result);
            catch(Exception e)
                e.printStackTrace();
            
            finally
                if(output != null)
                try
                    output.close();
                catch ( IOException e)
                    e.printStackTrace();
                
                
                if(input !=null)
                try
                    input.close();
                catch(IOException e)
                    e.printStackTrace();
                
                
                if(client != null)
                    try
                        client.close();
                    catch (IOException e)
                        e.printStackTrace();
                    
        
    

Client端

package Client;

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.InetSocketAddress;
import java.net.Socket;
/**
 *本地服务代理
 *将本地的接口调用转换成JDK的动态代理，在动态代理中实现接口的远程调用
 *创建Socket客户端，根据指定地址连接远程服务提供者
 *将远程服务调用所需要的接口类，方法名，参数列表等编码参数发送给服务提供者
 *同步阻塞等待服务端返回应答，获取应答之后返回
 * @author Administrator
 *
 * @param <S>
 */
public class RpcImporter<S> 
    @SuppressWarnings("unchecked")
    public S importer(final Class<?> serviceClass,final InetSocketAddress addr)
    
        return (S) Proxy.newProxyInstance(serviceClass.getClassLoader(), new  Class<?>[] serviceClass.getInterfaces()[0], new InvocationHandler() 
            
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable 
                // TODO Auto-generated method stub
                Socket socket =null;
                ObjectOutputStream output = null;
                ObjectInputStream input = null;
                try
                    socket = new Socket();
                    socket.connect(addr);
                    //将远程服务调用所需要的接口类，方法名，参数列表等编码参数发送给服务提供者
                    output = new ObjectOutputStream(socket.getOutputStream());
                    output.writeUTF(serviceClass.getName());
                    output.writeUTF(method.getName());
                    output.writeObject(method.getParameterTypes());
                    output.writeObject(args);
                    //同步阻塞等待服务端返回应答，获取应答之后返回
                    input= new ObjectInputStream(socket.getInputStream());
                
                return input.readObject();
                
                finally
                    if(socket != null)
                        socket.close();
                    
                    if(output != null)
                        output.close();
                    if(input != null)
                        input.close();
                
            
        );
    

测试代码：

package test;

import java.net.InetSocketAddress;

import Client.RpcImporter;
import Exporter.RpcExporter;
import Server.EchoService;
import Server.EchoServiceImpl;

public class run 

    public static void main(String[] args) 
        // TODO Auto-generated method stub

        //创建异步发布服务端的线程并启动，用于接受PRC客户端的请求，根据请求参数调用服务实现类，返回结果给客户端
        new Thread(new Runnable() 
            
            @Override
            public void run() 
                // TODO Auto-generated method stub
                try
                    RpcExporter.exporter("localhost", 8088);
                catch (Exception e)
                    e.printStackTrace();
                
            
        ).start();
        //创建客户端服务代理类，构造RPC求情参数，发起RPC调用
        RpcImporter<EchoService> importer=new RpcImporter<EchoService>();
        EchoService echo = importer.importer(EchoServiceImpl.class, new InetSocketAddress("localhost",8088));
        System.out.println(echo.echo("Are u ok?"));
    

RPC与restful api对比

REST是一种设计风格，它的很多思维方式与RPC是完全冲突的。 RPC的思想是把本地函数映射到API，也就是说一个API对应的是一个function，我本地有一个getAllUsers，远程也能通过某种约定的协议来调用这个getAllUsers。至于这个协议是Socket、是HTTP还是别的什么并不重要； RPC中的主体都是动作，是个动词，表示我要做什么。 而REST则不然，它的URL主体是资源，是个名词。而且也仅支持HTTP协议，规定了使用HTTP Method表达本次要做的动作，类型一般也不超过那四五种。这些动作表达了对资源仅有的几种转化方式。

RPC的根本问题是耦合。RPC客户端以多种方式与服务实现紧密耦合，并且很难在不中断客户端的情况下更改服务实现。RPC更偏向内部调用，REST更偏向外部调用。

Web 服务应该算是 RPC 的一个子集，理论上 RPC 能实现的功能， 用 Web 服务也能实现，甚至很多 RPC 框架选用 HTTP 协议作为传输层。

现在很多网站的 API 都是以 HTTP 服务的形式提供的，这也算是 RPC 的一种形式。

区别主要在这 2 个东西设计的出发点不太一样：

• HTTP 是面向浏览器设计的应用层协议，操作的核心在资源。我们更多的用 Web 服务在做网站。
• RPC 是为了在像在本地调用一个函数那样调用远程的代码而设计的，所以更关注减少本地调用和远程调用的差异，像 SOAP(简单对象访问协议) 这种东西是可以把对象当参数传的。

我们讨论 RPC 和 Web 的区别，其实是在谈论 2 个东西：序列化协议和传输协议。序列化协议比如常见的 XML，JSON 和比较现代的 Protocol Buffers、Thrift。 传输协议比如 TCP、UDP 以及更高层的 HTTP 1.1、HTTP 2.0。

一般我们考虑用 RPC 而不是 HTTP 构建自己的服务，通常是考虑到下面的因素：

• 接口是否需要 Schema 约束
• 是否需要更高效的传输协议（TCP，HTTP 2.0）
• 是否对数据包的大小非常敏感

比如 HTTP 是基于文本的协议，头部有非常多冗余（对于 RPC 服务而言）。HTTP 中我们用的最多就是 RESTful ，而 RESTful 是个弱 Schema 约束，大家通过文档沟通，但是如果我就是不在实现的时候对接口文档约定的参数做检查，你也不能把我怎么样。这个时候 Thrift 这种序列化协议的优势就体现出来了，由于 Schema 的存在，可以保证服务端接受的参数和 Schema 保持一致。

常用RPC框架

• Netty - Netty框架不局限于RPC，更多的是作为一种网络协议的实现框架，比如HTTP，由于RPC需要高效的网络通信，就可能选择以Netty作为基础
• brpc是一个基于protobuf接口的RPC框架，在百度内部称为“baidu-rpc”，它囊括了百度内部所有RPC协议，并支持多种第三方协议，从目前的性能测试数据来看，brpc的性能领跑于其他同类RPC产品
• Dubbo是Alibaba开发的一个RPC框架，远程接口基于Java Interface, 依托于Spring框架
• gRPC的Java实现的底层网络库是基于Netty开发而来，其Go实现是基于net库
• Thrift是Apache的一个项目(http://thrift.apache.org)，前身是Facebook开发的一个RPC框架，采用thrift作为IDL (Interface description language)
• jsonrpc

如果有兴趣了解更多相关内容，欢迎来我的个人网站看看：瞳孔的个人网站