Dubbo之手写RPC框架

Love The Way You Lie 2022-12-10 03:53 404阅读 0赞

建议从Dubbo之@SPI开始看。

关键词：Dubbo RPC

纯手写实现一个简单的RPC调用，帮助更好地学习和理解Dubbo

RPC-远程过程调用，我感觉可以理解成客户端（即消费者）通过TCP加上特定的消息协议访问服务端（即提供者），服务端根据消息协议内容调用本地方法并响应给客户端。就好像浏览器采用http协议，通过TCP传输去调用服务端接口一样，只不过http调用的是服务端的接口，接口其实对应着某个特定的方法。而RPC则直接调用服务端的方法。

关于协议：协议就是双方约定好的一种消息格式，这样发送消息的人发送出去的消息，接收消息的人才能够认识。就好比，两个人通信，发信人用的是中文，收信人呢就去查看新华字典来一个个读取信件中的内容，然后便知道和理解发信人的目的和行为了。你想想，如果此时的收信人拿出一本牛津词典查来查去，他是如论如何都不会理解中文的信件的。这里的新华字典就好比协议。

通过简单的代码来实现简单的RPC调用，这样更有助于理解和使用Dubbo，Dubbo包装了很多功能，理解起来还是蛮困难的。我感觉就像上面说的那样，本质还是TCP调用。TCP是传输层的协议了，比较底层，在JAVA中对应着就是Socket和ServerSocket。

█ 总体认识

watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70

  • 我代码都放在一个模块里面了，实际的项目中，客户端代码和服务端代码是分开的，有可能就是放在不同的服务器上。公共代码是客户端和服务端都需要的。
  • 客户端代码就是消费者，也就是服务的调用方；服务端代码就是提供者，也就是服务的提供者

█ 公共代码

  1. package common;
  3. /**
  4.  *
  5.  * 接口，面向接口调用。客户端调用的是接口的代理，
  6.  * 服务端真正调用接口的实现类（即CatService或DogService）
  7.  *
  8.  */
  9. public interface AnimalService {
  11.     String say();
  13.     int age(int age);
  15. }

协议类，客户端和服务端都需要使用。客户端根据协议的规定格式创建请求信息，服务端根据协议的规定格式解析请求内容。我这里的协议比较简单，就是字符串拼接，包含了请求的接口，方法和参数等信息（这样都是定位到调用的具体方法必不可少的条件）。dubbo-start的作用是用来标识一段请求消息的内容的。当多个客户端同时请求了服务端，服务端获取到的请求内容可能会被放到同一个字节数组里面，这样dubbo-start方面拆分每一条请求。请求的唯一标识的作用差不多，能够方便客户端解析响应消息的时候，对应上自己的哪一次请求。

  1. package common;
  3. import java.util.ArrayList;
  4. import java.util.HashMap;
  5. import java.util.List;
  6. import java.util.Map;
  8. /**
  9.  * 协议类，用于组装、拆解交流的信息
  10.  *
  11.  * 协议规定：一个请求必须以dubbo-start开头，第二个是请求的唯一标识 第三个是类的全限定名，第四个是方法名 第五个是参数类型列表
  12.  * 第六个是参数值，（参数类型和参数值，多个用逗号分割）
  13.  * 值与值中间用一个空行分割
  14.  *
  15.  */
  16. public class Protocol {
  18.     /**
  19.      * 构建请求信息，即客户端按照协议规定的格式构建请求信息
  20.      * @param requestId 请求唯一标识
  21.      * @param targetClass 请求的目标类
  22.      * @param methodName 请求的目标方法
  23.      * @param paramTypes 目标方法的参数类型
  24.      * @param paramValues 传给目标方法的参数
  25.      * @return
  26.      */
  27.     public byte[] packRequest(String requestId, Class targetClass, String methodName, Class[] paramTypes, Object[] paramValues) {
  28.         StringBuilder sb = new StringBuilder();
  29.         // 拼接协议信息
  30.         sb.append("dubbo-start").append(" ").append(requestId).append(" ").append(targetClass.getName()).append(" ").
  31.                 append(methodName).append(" ");
  33.         // 处理参数类型
  34.         if (paramTypes!=null && paramTypes.length>0) {
  35.             StringBuilder paramType = new StringBuilder();
  36.             for (Class type : paramTypes) {
  37.                 paramType.append(type.getName()).append(",");
  38.             }
  39.             // 去掉结尾的逗号
  40.             String substring = paramType.toString().substring(0, paramType.toString().length() - 1);
  41.             sb.append(substring).append(" ");
  42.         }
  43.         // 处理参数值
  44.         if (paramValues!=null && paramValues.length>0) {
  45.             StringBuilder paramValue = new StringBuilder();
  46.             for (Object value : paramValues) {
  47.                 paramValue.append(value).append(",");
  48.             }
  49.             // 去掉结尾的逗号
  50.             String substring = paramValue.toString().substring(0, paramValue.toString().length() - 1);
  51.             sb.append(substring);
  52.         }
  53.         return sb.toString().getBytes();
  54.     }
  56.     /**
  57.      * 拆解请求信息，即服务端解析客户端的请求信息
  58.      * @param bytes
  59.      * @param len
  60.      * @return
  61.      */
  62.     public Map<String, Object> unpackRequest(byte[] bytes, int len) {
  63.         Map<String, Object> map = new HashMap<>();
  64.         String recv = new String(bytes, 0, len);
  65.         // 根据协议约定，按照空格分隔各个请求信息
  66.         String[] split = recv.split(" ");
  67.         System.out.println("请求信息："+split);
  68.         // 请求ID
  69.         map.put(Const.REQUEST_ID, split[1]);
  70.         // 接口名
  71.         map.put(Const.INTERFACE_NAME, split[2]);
  72.         // 方法名
  73.         map.put(Const.METHOD_NAME, split[3]);
  74.         // 参数类型
  75.         List<Class> paramTypelist = new ArrayList<>();
  76.         if (split.length>4 && split[4]!=null) {
  77.             String[] types = split[4].split(",");
  78.             for (String type : types) {
  79.                 paramTypelist.add(convertParamType(type));
  80.             }
  81.             map.put(Const.PARAM_TYPES, paramTypelist);
  82.         }
  83.         // 参数值
  84.         if (split.length>5 && split[5]!=null) {
  85.             String[] values = split[5].split(",");
  86.             List<Object> valueList = new ArrayList<>(values.length);
  87.             for (int i=0; i<values.length; i++) {
  88.                 valueList.add(convertParamValue(paramTypelist.get(i), values[i]));
  89.             }
  90.             map.put(Const.PARAM_VALUES, valueList);
  91.         }
  93.         return map;
  94.     }
  96.     /**
  97.      * 参数类型转换
  98.      * @param type
  99.      * @return
  100.      */
  101.     private Class convertParamType(String type) {
  102.         // 简单地举了几个例子
  103.         if ("int".equals(type)) {
  104.             return int.class;
  105.         } else if ("java.util.ArrayList".equals(type) || "ArrayList".equals(type)) {
  106.             return ArrayList.class;
  107.         }
  108.         return String.class;
  109.     }
  111.     /**
  112.      * 参数值转换。解析的时候都转成了字符串，
  113.      * 这里需要根据参数具体的类型转换
  114.      * @param type
  115.      * @param value
  116.      * @return
  117.      */
  118.     private Object convertParamValue(Class type, String value) {
  119.         // 简单的举了几个例子
  120.         if (type==int.class) {
  121.             return Integer.parseInt(value);
  122.         } else if (type==double.class) {
  123.             return Double.parseDouble(value);
  124.         }
  125.         return value;
  126.     }
  128. }
  130. package common;
  132. /**
  133.  *
  134.  * 常量类，抽出字符串
  135.  *
  136.  */
  137. public class Const {
  139.     // 请求ID
  140.     public static final String REQUEST_ID = "requestId";
  142.     // 请求目标接口类型
  143.     public static final String INTERFACE_NAME = "interfaceName";
  145.     // 请求目标方法
  146.     public static final String METHOD_NAME = "methodName";
  148.     // 请求方法参数类型
  149.     public static final String PARAM_TYPES = "paramTypes";
  151.     // 请求方法参数值
  152.     public static final String PARAM_VALUES = "paramValues";
  154. }

█ 服务端

Service-包装了服务发布的功能，即通过创建的ServerSocket来接收和响应客户端的请求。

  1. package server;
  3. import common.Const;
  4. import common.Protocol;
  6. import java.io.InputStream;
  7. import java.io.OutputStream;
  8. import java.lang.reflect.Method;
  9. import java.net.ServerSocket;
  10. import java.net.Socket;
  11. import java.util.List;
  12. import java.util.Map;
  13. import java.util.concurrent.ConcurrentHashMap;
  15. /**
  16.  *
  17.  * 包装服务端服务发布的功能
  18.  * 
  19.  */
  20. public class Service<T> {
  21.     // 用于存放接口和实现类对象的关系，作为服务端服务字典，方便查找
  22.     private static ConcurrentHashMap<String, Object> beanMap = new ConcurrentHashMap<>();
  23.     // 协议，在服务端，协议的作用就是解析消费端发送过来的消息
  24.     private Protocol protocol = new Protocol();
  26.     // 接口的实现类
  27.     private T obejct;
  28.     // 接口类型
  29.     private Class interfaceClass;
  31.     public Service(Class interfaceClass, T object) {
  32.         // 前置校验，借实现类是否实现了接口
  33.         boolean assignableFrom = interfaceClass.isAssignableFrom(object.getClass());
  34.         if (!assignableFrom) {
  35.             throw new IllegalArgumentException("object 必须实现interfaceClass接口");
  36.         }
  37.         this.interfaceClass = interfaceClass;
  38.         this.obejct = object;
  39.         // 缓存起来，方便客户端调用的时候，查找目标类
  40.         beanMap.put(interfaceClass.getName(), object);
  41.     }
  43.     public void export() {
  44.         try {
  45.             ServerSocket serverSocket = ServiceServer.getServerSocket();
  46.             while (true) {
  47.                 // 这里会阻塞，等待客户端的连接
  48.                 Socket socket = serverSocket.accept();
  49.                 // 响应结果
  50.                 byte[] result = new byte[1];
  51.                 // 接收消费者的请求
  52.                 InputStream inputStream = socket.getInputStream();
  53.                 byte[] recv = new byte[1024];
  54.                 // 这里会阻塞，等待客户端的请求信息
  55.                 int len = inputStream.read(recv);
  56.                 // 根据协议，转换请求信息
  57.                 Map<String, Object> requestMap = protocol.unpackRequest(recv, len);
  58.                 // 获取调用的接口
  59.                 Object interfaceName = requestMap.get(Const.INTERFACE_NAME);
  60.                 // 获取接口实现类对象
  61.                 Object object = beanMap.get(interfaceName);
  62.                 if (object==null) {
  63.                     result = "请求的接口不存在".getBytes();
  64.                 } else {
  65.                     // 方法名
  66.                     Object methodName = requestMap.get(Const.METHOD_NAME);
  67.                     // 参数类型
  68.                     Class[] paramTypes = null;
  69.                     List<Class> paramTypeList = (List<Class>)requestMap.get(Const.PARAM_TYPES);
  70.                     if (paramTypeList!=null && paramTypeList.size()>0) {
  71.                         paramTypes = paramTypeList.toArray(new Class[paramTypeList.size()]);
  72.                     }
  73.                     // 参数值
  74.                     Object[] paramValues = null;
  75.                     List<Object> paramValueList = (List<Object>)requestMap.get(Const.PARAM_VALUES);
  76.                     if (paramValueList!=null && paramValueList.size()>0) {
  77.                         paramValues = paramValueList.toArray();
  78.                     }
  79.                     // 获取到调用的方法
  80.                     Method method = object.getClass().getMethod(methodName.toString(), paramTypes);
  81.                     // 方法调用结果
  82.                     Object invoke = method.invoke(object, paramValues);
  83.                     if (invoke!=null) {
  84.                         result = invoke.toString().getBytes();
  85.                     }
  86.                 }
  87.                 // 将结果返回给客户端
  88.                 OutputStream os = socket.getOutputStream();
  89.                 os.write(result);
  90.                 os.flush();
  91.                 os.close();
  92.                 inputStream.close();
  93.             }
  94.         } catch (Exception e) {
  95.             e.printStackTrace();
  96.         }
  97.     }
  99. }
  101. package server;
  103. import java.net.ServerSocket;
  105. /**
  106.  * 创建ServerSocket，暴露服务
  107.  * 多次服务暴露，都使用一个ServerSocket
  108.  * 这里的端口号就固定写了8899
  109.  *
  110.  */
  111. public class ServiceServer {
  113.     private static volatile ServerSocket serverSocket;
  115.     private ServiceServer() {
  116.         throw new IllegalStateException();
  117.     }
  119.     /**
  120.      * 创建一个ServerSocket，用于暴露服务。
  121.      * @return
  122.      */
  123.     public static ServerSocket getServerSocket() {
  124.         if (serverSocket==null) {
  125.             synchronized (ServiceServer.class) {
  126.                 if (serverSocket==null) {
  127.                     try {
  128.                         // 因为没有注册中心的功能，没法让消费者去感知服务的端口，这里端口就写死了
  129.                         serverSocket = new ServerSocket(8899);
  130.                     } catch (Exception e) {
  131.                         e.printStackTrace();
  132.                     }
  133.                 }
  134.             }
  135.         }
  136.         return serverSocket;
  137.     }
  139. }
  141. package server;
  143. import common.AnimalService;
  145. public class DogService implements AnimalService {
  147.     @Override
  148.     public String say() {
  149.         return "this is a dog";
  150.     }
  152.     @Override
  153.     public int age(int age) {
  154.         return age;
  155.     }
  157. }
  161. package server;
  163. import common.AnimalService;
  165. public class CatService implements AnimalService {
  167.     @Override
  168.     public String say() {
  169.         return "this is a cat";
  170.     }
  172.     @Override
  173.     public int age(int age) {
  174.         return age + 10;
  175.     }
  177. }

█ 客户端

  1. package client;
  3. import common.Protocol;
  5. import java.io.InputStream;
  6. import java.io.OutputStream;
  7. import java.lang.reflect.InvocationHandler;
  8. import java.lang.reflect.Method;
  9. import java.lang.reflect.Proxy;
  10. import java.net.Socket;
  11. import java.util.UUID;
  13. /**
  14.  * 服务端调用代理类
  15.  * 客户端并不知道在服务端上接口具体的实现类是哪个，只能通过调用接口来获取。
  16.  * 创建接口的代理，主要工作就是创建Socket去连接服务端，并按照协议格式发送请求
  17.  *
  18.  */
  19. public class Reference implements InvocationHandler {
  21.     // 协议
  22.     private Protocol protocol = new Protocol();
  24.     private Class interfaceClass;
  26.     public Reference(Class interfaceClass) {
  27.         this.interfaceClass = interfaceClass;
  28.     }
  30.     public Object getReference() {
  31.         return Proxy.newProxyInstance(Reference.class.getClassLoader(), new Class[]{interfaceClass}, this);
  32.     }
  34.     @Override
  35.     public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
  36.         Object result = null;
  38.         String name = method.getName();
  39.         // 如果是父类Object中的方法就直接返回方法名，这里举几个例子，不全
  40.         if ("equals".equals(name) || "toString".equals(name)) {
  41.             return name;
  42.         }
  43.         Socket socket = new Socket("localhost", 8899);
  44.         // 客户端发送请求
  45.         OutputStream os = socket.getOutputStream();
  46.         // 通过约定好的协议格式构建请求信息
  47.         byte[] requestByte = protocol.packRequest(UUID.randomUUID().toString(), interfaceClass, name, method.getParameterTypes(), args);
  48.         os.write(requestByte);
  49.         os.flush();
  50.         os.close();
  51.         // 客户端接收响应结果
  52.         InputStream is = socket.getInputStream();
  53.         byte[] recv = new byte[1024];
  54.         int read = is.read(recv);
  55.         is.close();
  56.         // 这里应该还有解码器的，我就简单地写了一下
  57.         Class<?> returnType = method.getReturnType();
  58.         if (String.class==returnType) {
  59.             result = new String(recv, 0, read);
  60.         } else if (int.class==returnType || Integer.class==returnType) {
  61.             return Integer.parseInt(new String(recv, 0, read));
  62.         }
  64.         return result;
  65.     }
  66. }

█ 运行服务端

  1. package server;
  3. import common.AnimalService;
  5. public class Main {
  7.     public static void main(String[] args) {
  8.         // 暴露DogService
  9.         Service service = new Service(AnimalService.class, new DogService());
  10.         // 启动一个ServerSocket，等待客户端的连接和接收消息
  11.         service.export();
  12.     }
  14. }

█ 运行客户端

  1. package client;
  3. import common.AnimalService;
  5. public class Main {
  7.     public static void main(String[] args) {
  8.         // 客户端创建代理对象，代理去实现请求服务端的逻辑
  9.         Reference reference = new Reference(AnimalService.class);
  10.         AnimalService animalService = (AnimalService)reference.getReference();
  11.         // 当调用say方法，实际会调用Reference的invoke方法
  12.         String say = animalService.say();
  13.         System.out.println(say);
  14.         System.out.println("获取到的age="+animalService.age(10));
  16.         // 输出结果：
  17.         // this is a dog
  18.         //获取到的age=10
  19.     }
  21. }

█ 总结

上面代码写的粗糙，功能也粗糙，只是简单地模拟了一下RPC调用，实现了服务端和客户端，根据协议构建和解析协议的功能。希望看完能够帮助更好的去理解Dubbo。在Dubbo中还实现了注册中心，路由选择，负载均衡，容错等等强大的功能，这些功能有机会的话会在后面介绍（希望我能坚持写下去）。我想，只有理解了RPC的工作原理，才能更好的去学习和理解Dubbo的功能。

关于Dubbo的使用、服务暴露、服务注册、负载均衡等，官网已经有很详细的介绍了。我就不重复写了，写了也就是分析源码。具体请移步Dubbo官网慢慢品味：我简述一下自己关于Dubbo的理解：

服务端：

  • 服务暴露：将ServiceConfig对应的接口实现类对外提供调用服务，让客户端（消费者）能够请求调用。

  • 服务注册：将ServiceConfig对应的接口实现类注册到注册中心上，这样是对客户端透明，即提供了一个公示栏，让客户端调用者查看其可以调用哪些服务。

  • 一个或@Service，就对应一个ServiceConfig实例。ServiceConfig是Dubbo服务暴露和注册的起点。开始于export方法

  • ServiceConfig在暴露和注册服务时，使用了Invoker这样的代理类，能够灵活地调用具体的接口实现类，而不用写死代码。

  • 服务暴露时，一个Invoker又会被封装成一个Exporter，由这个Exporter负责具体的服务暴露和引用工作。
  • 所有被暴露的服务都会缓存在Map> exporterMap = new ConcurrentHashMap()，exporterMap之中，key是服务的标识信息（端口，版本号，分组名）。这样当消费者待着请求信息过来请求时，Dubbo提取一些参数值，组装成key，去本地缓存中查找对应的Exporter，而Exporter中又包含了Invoker，这样自然就调用了实际的接口实现类方法了。
  • 真正做服务暴露和注册的是协议Protocol，不如DubboProtocol，RegistryProtocol。

Service ——-> ServiceConfig ———-> Invoker ——-> Exporter

消费端：

  • 一个或@Reference对应一个ReferenceConfig。起点是get方法。
  • 一个ReferenceConfig也是对应一个Invoker，通过Invoker代理服务端的调用。
  • Invoker外包装了一层Directory。这个可以理解成消费者端的本地缓存，缓存了服务端的服务调用。当发起远程调用时，要去Directory这个缓存中去查找对应的Invoker，由Invoker完成具体的调用过程。
  • Directory又被Cluster所有。Cluster根据Directory中缓存的Invoker列表完成路由、负载均衡和容错功能

Reference ——-> ReferenceConfig ———> Cluster ———-> Directory ——-> Invoker———->（Exporter ———>Invoker——->Service）

发表评论

表情：
评论列表 （有 0 条评论，404人围观）

还没有评论，来说两句吧...

相关阅读

    相关 手写RPC框架之泛化调用

    一、背景 前段时间了解了泛化调用这个玩意儿，又想到自己之前写过一个RPC框架（参考《[手写一个RPC框架][RPC]》），于是便想小试牛刀。 二、泛化调用简介 什

    电玩女神电玩女神/ 0 赞/ 186 阅读

    相关 Java手写dubbo框架

    前言 在对dubbo有了较为深入的使用和理解后，来尝试从dubbo框架的角度重新认识下它，对照着dubbo官方的这张图进行反复的理解后，我们可以从已有掌握的技术出发，来尝

    超、凢脫俗超、凢脫俗/ 0 赞/ 158 阅读

    相关 手写RPC框架(netty+zookeeper)

    　　RPC是什么？远程过程调用，过程就是业务处理、计算任务，像调用本地方法一样调用远程的过程。 　　RMI和RPC的区别是什么？RMI是远程方法调用，是oop领域中RPC的一

    Bertha 。Bertha 。/ 0 赞/ 251 阅读

    相关 【RPC框架】之 Dubbo

    一、分布式基础理论 -------------------- 1、什么是分布式系统 《分布式系统原理与范型》定义： 分布式系统是若干独立计算机的集合，这些计算机

    小咪咪小咪咪/ 0 赞/ 359 阅读