简单组件介绍：（摘抄自：）

以Dubbo为例

RPC中可以认为有四个角色，消费者（Consumer），提供者（Provider） 注册中心（Registry），监控中心（Monitor），这个还是很好理解的，以前在同一系统的方法的调用者因为网络的存在，变成了消费者，被调的方法成了提供者

同样注册中心Registry的存在，其实就是为了让消费者实时的去感知提供者的存在，去告诉消费者它对应的提供者的地址

Monitor顾名思义，监控者，其实在整个过程中，它并不是一定要存在，只是它可以做统计，做一些数据分析，提供整个系统的可用性，健壮性。

 

好了，我们先简单分析一下Registry / Consumer / Provider / Monitor 这四个角色的定义和每个角色如何各司其职，相互协作完成这整个过程的

分析，分析分成两个方面，一是每个角色在网络的定位，二是每个角色所要完成的职责

1）Registry注册中心

   简述：

    ①注册中心可以有多个，都是无状态的，每个注册中心之间信息不交互

    ②从网络的角度来说，它都是server端，它不需要主动地连接其他的任何实例，只需要像一个地主一样等待别人来连接

    ③消费者随机选择注册中心集群中的任何实例建立长连接，提供者与注册中心中的每一个实例都建立长连接

   职责(与其说职责，还不如说代码要实现的功能)：

   ①接收服务提供者的服务注册信息，接收到信息之后，发送ACK信息给服务提供者，否则服务提供者重新发送注册信息

   ②接收消费者的订阅信息，并把它订阅的结果返回给消费者

   ③如果注册信息变更，会主动通知订阅变更信息的消费者，注册信息的变更包括服务提供者下线，服务被人工降级，或者服务提供者的地址变更

   ④持久化一些服务信息，例如某些服务管理员审核过了，则该服务重新注册后则不需要再审核，再例如，某个服务负载均衡的策略被管理员设置为轮询，那么下次它在注册的时候，则就是轮询，而不是默认的负载策略

2）Provider提供者

   简述：

   ①提供者是一个精神分裂的病人，它在网络上（可以更加明确地说是站在Netty的角度上）饰演两个角色：

        1）一是它是客户端，需要去连接Registry，发送注册信息，它也需要去连接monitor端，去发送一些调用的统计信息

        2）二是它也是服务端，需要作为server端等待Consumer去连接，连接成功后调用服务

   职责：

  ①将自己的信息，提供的接口信息编织成注册信息发送给registry端

  ②能够动态去调自己的方法，可以通过反射，cglib等一些方法去调用自己提供的那些方法

  ③提供服务降级等服务，如果当某些服务调用的失败率高于限定值的时候，可以有一个对应的mock方法，提供降级服务

  ④限流服务，限流的方式有很多种，也有很多实现方式，最简单的就是控制调用次数，比如100w次，其实简单的就是控制单位时间的调用次数，防止业务洪流冲垮服务

  ⑤统计活动，将一些调用信息统计好发送给Monitor端

  ⑥补充......

3)Consumer消费者 

  简述

  ①它也是有两个网络角色，不过并不是精神分裂，它都是作为网络的客户端存在，一它需要去连接registry去获取到订阅信息，二是它需要主动去连接provider端去调用服务

  职责

  ①去向Registry端订阅服务，拿到registry端返回的结果，这个结果也就是provider的网络地址，先建立TCP的长连接，可能是多个地址，因为提供某个服务的可能有多个提供者

  ②当开始系统主动调用该服务的时候，拿到刚才建立的连接的集合，根据某个方法，是随意还是轮询，获取到其中的一个连接，发送方法入参，等待响应

  ③当注册中心发送某个服务的调用的负载策略发生变化过，发送信息给consumer，consumer需要做相应的变更

4）Monitor监控者

  简述

 ①这个与整个系统是没有任何直接的关系的，实现方式也是多样的，可以与上面一样建立长连接，接收每个角色统计的信息，然后展示给用户，可以使用MQ,使用消息队列，每个角色把自己统计的信息放到队列中，Monitor去消费这些信息，这样做的好处就是解耦，如果monitor宕了，不影响服务

 

简单RPC实现：

大体的RPC的流程稍微理了一下，接下来我们就来一一去实现一个简单的RPC实现远程过程调用，功能并不完善，旨在展示怎么实现“远程过程调用”。

简单介绍下这几个模块包

1）、RPC包 RPC基础功能包：提供RPC中最基本的:  服务暴露 expor ,服务发现引用 refer  功能（方法）

2）、service 包： 自定义的服务接口.  impl 接口的实现类，实现具体服务。

3）、provider包 : 调用expor向中心暴露自己的服务。这里的provider是个功能测试一体包。

4）、consumer包：调用refer发现并使用某个已经暴露的服务方法。这里的consumer也是个功能测试一体包。

代码：（代码摘抄引用自： ）

RpcFramework:

package com.zjl.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;    /**  * RpcFramework  *   */  public class RpcFramework {        /**      * 暴露服务      *       * @param service 服务实现      * @param port 服务端口      * @throws Exception      */      public static void export(final Object service, int port) throws Exception {          if (service == null)              throw new IllegalArgumentException("service instance == null");          if (port <= 0 || port > 65535)              throw new IllegalArgumentException("Invalid port " + port);          System.out.println("Export service " + service.getClass().getName() + " on port " + port);          ServerSocket server = new ServerSocket(port);          for(;;) {              try {                  final Socket socket = server.accept();                  new Thread(new Runnable() {                      @Override                      public void run() {                          try {                              try {                                  ObjectInputStream input = new ObjectInputStream(socket.getInputStream());                                  try {                                      String methodName = input.readUTF();                                      Class
    [] parameterTypes = (Class
    [])input.readObject();                                      Object[] arguments = (Object[])input.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 {                                      input.close();                                  }                              } finally {                                  socket.close();                              }                          } catch (Exception e) {                              e.printStackTrace();                          }                      }                  }).start();              } catch (Exception e) {                  e.printStackTrace();              }          }      }        /**      * 引用服务      *       * @param 
    
      接口泛型      * @param interfaceClass 接口类型      * @param host 服务器主机名      * @param port 服务器端口      * @return 远程服务      * @throws Exception      */      @SuppressWarnings("unchecked")      public static 
     
       T refer(final Class
      
        interfaceClass, final String host, final int port) throws Exception {          if (interfaceClass == null)              throw new IllegalArgumentException("Interface class == null");          if (! interfaceClass.isInterface())              throw new IllegalArgumentException("The " + interfaceClass.getName() + " must be interface class!");          if (host == null || host.length() == 0)              throw new IllegalArgumentException("Host == null!");          if (port <= 0 || port > 65535)              throw new IllegalArgumentException("Invalid port " + port);          System.out.println("Get remote service " + interfaceClass.getName() + " from server " + host + ":" + port);          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();                  }              }          });      }    }
      
     
    

自定义服务接口：

package com.zjl.service;/** * HelloService 自定义的服务接口*  */  public interface HelloService {    String hello(String name);  }

自定义服务接口的实现：

package com.zjl.service.impl;import com.zjl.service.HelloService;/** *  HelloServiceImpl 自定义服务接口的实现类 * */public class HelloServiceImpl implements HelloService {  	      public String hello(String name) {          return "Hello " + name;      }  }

暴露服务（服务提供者）：

package com.zjl.provider;import com.zjl.RPC.RpcFramework;import com.zjl.service.HelloService;import com.zjl.service.impl.HelloServiceImpl;/**  * RpcProvider  暴露服务 *    */  public class RpcProvider {        public static void main(String[] args) throws Exception {          HelloService service = new HelloServiceImpl();          RpcFramework.export(service, 1234);      }    }

消费服务（服务调用者）：

package com.zjl.consumer;import com.zjl.RPC.RpcFramework;import com.zjl.service.HelloService;/**  * RpcConsumer 消费服务 *   */  public class RpcConsumer {            public static void main(String[] args) throws Exception {          HelloService service = RpcFramework.refer(HelloService.class, "127.0.0.1", 1234);          for (int i = 0; i < Integer.MAX_VALUE; i ++) {              String hello = service.hello("World" + i);              System.out.println(hello);              Thread.sleep(1000);          }      }        }

 

调试：先运行RpcProvider 暴露服务。

然后运行RpcConsumer 就可以来消费已经暴露的服务了。

 

但是上面的的代码是很简陋的。

RPC中的基本角色 ：Register 按理说应该能注册多个。在本方法中只是在export中只能存一个其他（服务）方法并不能注册。 Monitor 也没有，这个其实有没有都行，没影响。

如果实现一个注册中心的话，肯定是要注册很多方法。然后带来的新的挑战就是：维护Call ID映射

网络传输的话还应该：序列化和反序列化。

还应该有：网络传输的问题

 

关于“本地过程调用和远程过程调用的区别 ”再理解一下这里可以看看知乎的下的回答

本地过程调用

RPC就是要像调用本地的函数一样去调远程函数。在研究RPC前，我们先看看本地调用是怎么调的。假设我们要调用函数Multiply来计算lvalue * rvalue的结果:

1 int Multiply(int l, int r) {2    int y = l * r;3    return y;4 }5 6 int lvalue = 10;7 int rvalue = 20;8 int l_times_r = Multiply(lvalue, rvalue);

那么在第8行时，我们实际上执行了以下操作：

1. 将 lvalue 和 rvalue 的值压栈
2. 进入Multiply函数，取出栈中的值10 和 20，将其赋予 l 和 r
3. 执行第2行代码，计算 l * r ，并将结果存在 y
4. 将 y 的值压栈，然后从Multiply返回
5. 第8行，从栈中取出返回值 200 ，并赋值给 l_times_r

以上5步就是执行本地调用的过程。

 

远程过程调用带来的新问题

在远程调用时，我们需要执行的函数体是在远程的机器上的，也就是说，Multiply是在另一个进程中执行的。这就带来了几个新问题：

1. Call ID映射。我们怎么告诉远程机器我们要调用Multiply，而不是Add或者FooBar呢？在本地调用中，函数体是直接通过函数指针来指定的，我们调用Multiply，编译器就自动帮我们调用它相应的函数指针。但是在远程调用中，函数指针是不行的，因为两个进程的地址空间是完全不一样的。所以，在RPC中，所有的函数都必须有自己的一个ID。这个ID在所有进程中都是唯一确定的。客户端在做远程过程调用时，必须附上这个ID。然后我们还需要在客户端和服务端分别维护一个 {函数 <--> Call ID} 的对应表。两者的表不一定需要完全相同，但相同的函数对应的Call ID必须相同。当客户端需要进行远程调用时，它就查一下这个表，找出相应的Call ID，然后把它传给服务端，服务端也通过查表，来确定客户端需要调用的函数，然后执行相应函数的代码。
2. 序列化和反序列化。客户端怎么把参数值传给远程的函数呢？在本地调用中，我们只需要把参数压到栈里，然后让函数自己去栈里读就行。但是在远程过程调用时，客户端跟服务端是不同的进程，不能通过内存来传递参数。甚至有时候客户端和服务端使用的都不是同一种语言（比如服务端用C++，客户端用Java或者Python）。这时候就需要客户端把参数先转成一个字节流，传给服务端后，再把字节流转成自己能读取的格式。这个过程叫序列化和反序列化。同理，从服务端返回的值也需要序列化反序列化的过程。
3. 网络传输。远程调用往往用在网络上，客户端和服务端是通过网络连接的。所有的数据都需要通过网络传输，因此就需要有一个网络传输层。网络传输层需要把Call ID和序列化后的参数字节流传给服务端，然后再把序列化后的调用结果传回客户端。只要能完成这两者的，都可以作为传输层使用。因此，它所使用的协议其实是不限的，能完成传输就行。尽管大部分RPC框架都使用TCP协议，但其实UDP也可以，而gRPC干脆就用了HTTP2。Java的Netty也属于这层的东西。

所以，要实现一个RPC框架，其实只需要把以上三点实现了就基本完成了。

Call ID映射可以直接使用函数字符串，也可以使用整数ID。映射表一般就是一个哈希表。

序列化反序列化可以自己写，也可以使用Protobuf或者FlatBuffers之类的。

网络传输库可以自己写socket，或者用asio，ZeroMQ，Netty之类。

 

这里还有一个实现的比上面好的（能注册多个服务）Java Rpc 的小例子：