• Dubbo-聊聊Dubbo协议

    前言

    Dubbo源码阅读分享系列文章，欢迎大家关注点赞

    SPI实现部分

    1. Dubbo-SPI机制

    2. Dubbo-Adaptive实现原理

    3. Dubbo-Activate实现原理

    4. Dubbo SPI-Wrapper

    注册中心

    1. Dubbo-聊聊注册中心的设计

    2. Dubbo-时间轮设计

    通信

    1. Dubbo-聊聊通信模块设计

    什么是协议

    在网络交互中是以字节流的形式传递的，对于字节流都是二进制格式，这样我们就面临一个问题就是如何转化为我们可以识别的字符，协议就是来解决这个问题的，协议用通俗易懂地解释就是通信双方需要遵循的约定。 在日常开发中，我们常见的网络传输协议有TCP、UDP、HTTP。常用的中间件也会定义对应的协议，如Redis、Mysql、Zookeeper等都有自己约定的协议，同样Dubbo的通信也采用一种协议，这些都是应用层协议，都是基于TCP或者UDP设计的。

    如何定义协议

    应用层协议一般的形式有三种：定长协议、特殊结束符和变长协议，聊到这里就可以抛出来一个常见的面试题，如何解决网络通信粘包和拆包的问题？该问题的解决方案也就是通过约定协议，下面我们就来聊聊这三种模式优缺点以及使用场景。

    定长协议

    定长的协议是指协议内容的长度是固定的，比如协议byte长度是50，当从网络上读取50个byte后，就进行decode解码操作。

    优点

    定长协议在读取或者写入时，效率比较高，因为数据大小都是确定的。

    缺点

    定长协议的缺点在于适应性不足，网络传输中传输的内容的大小不可能都是相同的，因此对于一些长度不够的消息，明显过于的浪费带宽。

    特殊结束符

    特殊结束符就是在每次传输结束的时候使用一个特殊的结束符，在Redis中的协议采用了特殊结束符，客户端和服务器发送的命令一律使用\r\n（CRLF）结尾。

    优点

    与定长协议一样读取或者写入时，效率比较高，同时解决定长协议的尴尬。

    缺点

    特殊结束符方式的问题是必须要有一个完整的消息体才能进行传输，除此之外必须要防止用户传输的数据不能同结束符相同，否则就会出现紊乱。

    变长协议

    变长协议由定长以及不定长两部分组成，定长部分一般是协议头，此部分会包含变长部分的描述，变长协议我们经常使用的HTTP协议采用变长协议，HTTP请求报文格式是由三部分组成：

    1. 请求行：包括Url、Version等，由空格分隔，\r\n结尾;

    2. 请求头：多行，每行是key:value的格式，以\r\n结尾;

    3. 请求体：请求头与请求体直接由一个空白行分隔，请求体的长度在请求头中由content-length给出;

    优点

    灵活性比较高，解决了定长协议以及特殊结束符的所有缺点。

    缺点

    复杂性比较高，需要自定义一套标准，所有消息都需要按照该格式发送以及解析。

    Dubbo协议

    Dubbo框架支持很多协议，默认采用Dubbo协议，Dubbo协议采用的是变长协议的设计，整体的格式如下:

    1. 0~7位和8~15位分别是Magic High和Magic Low，是固定魔数值（0xdabb），我们可以通过这两个Byte，判断是否为Dubbo协议;

    2. 16位是Req/Res标识，用于标识当前消息是请求还是响应；

    3. 17位是2Way标识，用于标识当前消息是单向还是双向，如果需要来自服务器的返回值，则设置为1；

    4. 18位是Event标识，用于标识当前消息是否为事件消息；

    5. 19~23位是序列化类型的标志，用于标识当前消息使用哪一种序列化算法；

    6. 24~31位是Status状态，用于记录响应的状态，当Req/Res为0时才有用；

    7. 32~95位是Request ID，用于记录请求的唯一标识；

    8. 96~127位是序列化后的内容长度，该值是按字节计算；

    9. 128位之后是可变的数据，被特定的序列化类型序列化后，每个部分都是一个 byte [] 或者byte，如果是请求包，则每个部分依次为：Dubbo version、Service name、Service version、Method name、Method parameter types、Method arguments 和 Attachments。如果是响应包，则每个部分依次为：返回值类型、返回值；

    image.png

    优点

    Dubbo协议整体设计比较简洁，能采用1个bit表示的，不会用一个byte来表示；此外请求头和响应头一致，整体采用一套解析标准就可以，代码实现起来相对简单。

    缺点

    由于整体的设计相对简洁，导致扩展性不够；

    Dubbo协议是如何解析的

    在通信篇中我们讲过Codec2该接口，该接口提供了encode和decode个方法来实现消息与字节流之间的相互转换，关于该接口的实现我们没有讲解，这里我们来看看此部分和Dubbo协议有什么关系。

    AbstractCodec抽象类没有实现Codec2中定义的接口方法，而是提供了几个给子类用的基础方法。

    1. getSerialization方法：通过SPI获取当前使用的序列化方式；

    2. checkPayload方法：检查编解码数据的长度，如果数据超长，会抛出异常；

    3. isClientSide、isServerSide方法：判断当前是Client端还是Server端；

    接下来我们就来聊聊子类如何被解析的，我们可以看到四个子类的继承关系，重点介绍的是ExchangeCodec和DubboCodec，其他就是做一下简单介绍。 TransportCodec该类已经被标注为弃用，该类内部也就是根据getSerialization方法选择的序列化方法，对传入消息或ChannelBuffer进行序列化或反序列化。 TelnetCodec继承了TransportCodec的能力，该类主要是提供了对Telnet命令处理的能力，该功能主要是对服务进行治理的功能，这里后续我们画一点时间来进行介绍。

    ExchangeCodec

    ExchangeCodec继承了TelnetCodec，在该类基础上增加Dubbo协议头的处理能力,接下来我们首先来看下其核心字段，

    1. //协议头长度
    2. protected static final int HEADER_LENGTH = 16;
    3. //魔数 判断是否是Dubbo协议
    4. protected static final short MAGIC = (short) 0xdabb;
    5. protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
    6. protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
    7. //设置请求响应标志位
    8. protected static final byte FLAG_REQUEST = (byte) 0x80;
    9. //单向还是双向标志位
    10. protected static final byte FLAG_TWOWAY = (byte) 0x40;
    11. //是否事件消息标志位
    12. protected static final byte FLAG_EVENT = (byte) 0x20;
    13. //序列化协议标志位
    14. protected static final int SERIALIZATION_MASK = 0x1f;

    通过核心字段我们可以发现其实和我们介绍的Dubbo的协议是一致的，因此接下来的encode和decode就是对Dubbo协议头的解密和编码，我们来下看encode方法，在encode方法中会根据需要编码的消息类型进行分类， 分为三类:Request、Response、telenet，encodeRequest方法专门对Request对象进行编码，encodeResponse方法对Response对象进行编码。

    1. @Override
    2.   public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
    3.     //Request
    4.     if (msg instanceof Request) {
    5.       encodeRequest(channel, buffer, (Request) msg);
    6.       //Response
    7.     } else if (msg instanceof Response) {
    8.       encodeResponse(channel, buffer, (Response) msg);
    9.     } else {
    10.       //telenet
    11.       super.encode(channel, buffer, msg);
    12.     }
    13.   }
    14. protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
    15.   Serialization serialization = getSerialization(channel, req);
    16.   //存储协议头
    17.   byte[] header = new byte[HEADER_LENGTH];
    18.   // set magic number.
    19.   Bytes.short2bytes(MAGIC, header);
    20.  
    21.   //设置协议头标志位
    22.   header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
    23.  
    24.   if (req.isTwoWay()) {
    25.     header[2] |= FLAG_TWOWAY;
    26.   }
    27.   if (req.isEvent()) {
    28.     header[2] |= FLAG_EVENT;
    29.   }
    30.  
    31.   //记录请求ID
    32.   Bytes.long2bytes(req.getId(), header, 4);
    33.  
    34.   //序列化请求 并统计序列化以后字节数
    35.   int savedWriteIndex = buffer.writerIndex();
    36.   //将写入位置后移16位
    37.   buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
    38.   //请求序列化
    39.   ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
    40.  
    41.   //是否心跳检查 为空就是心跳检查
    42.   if (req.isHeartbeat()) {
    43.     // heartbeat request data is always null
    44.     bos.write(CodecSupport.getNullBytesOf(serialization));
    45.   } else {
    46.     ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
    47.     //事件序列化
    48.     if (req.isEvent()) {
    49.       //事件序列化
    50.       encodeEventData(channel, out, req.getData());
    51.     } else {
    52.       //Dubbo请求序列化
    53.       encodeRequestData(channel, out, req.getData(), req.getVersion());
    54.     }
    55.     out.flushBuffer();
    56.     if (out instanceof Cleanable) {
    57.       ((Cleanable) out).cleanup();
    58.     }
    59.   }
    60.  
    61.   bos.flush();
    62.   bos.close();
    63.   //获取字节数
    64.   int len = bos.writtenBytes();
    65.   //检查字节长度
    66.   checkPayload(channel, len);
    67.   //将字节数写入header数组中
    68.   Bytes.int2bytes(len, header, 12);
    69.  
    70.   //重置写入位置
    71.   buffer.writerIndex(savedWriteIndex);
    72.   //写入消息头
    73.   buffer.writeBytes(header);
    74.   //buffer写出去的位置从writeIndex开始 加上header长度 数据长度
    75.   buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    76. }
    77.  
    78. protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
    79.   int savedWriteIndex = buffer.writerIndex();
    80.   try {
    81.     //序列化
    82.     Serialization serialization = getSerialization(channel, res);
    83.     //协议头  长度为16字节
    84.     byte[] header = new byte[HEADER_LENGTH];
    85.     //魔数
    86.     Bytes.short2bytes(MAGIC, header);
    87.     //序列化方式
    88.     header[2] = serialization.getContentTypeId();
    89.     //心跳还是正常消息
    90.     if (res.isHeartbeat()) {
    91.       header[2] |= FLAG_EVENT;
    92.     }
    93.     //响应状态
    94.     byte status = res.getStatus();
    95.     header[3] = status;
    96.     //设置请求ID
    97.     Bytes.long2bytes(res.getId(), header, 4);
    98.  
    99.     //写入时候真需要加上协议头长度
    100.     buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
    101.     ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
    102.  
    103.     //对响应信息进行编码
    104.     if (status == Response.OK) {
    105.       if(res.isHeartbeat()){
    106.         //心跳
    107.         bos.write(CodecSupport.getNullBytesOf(serialization));
    108.       }else {
    109.         //正常响应
    110.         ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
    111.         if (res.isEvent()) {
    112.           encodeEventData(channel, out, res.getResult());
    113.         } else {
    114.           encodeResponseData(channel, out, res.getResult(), res.getVersion());
    115.         }
    116.         out.flushBuffer();
    117.         if (out instanceof Cleanable) {
    118.           ((Cleanable) out).cleanup();
    119.         }
    120.       }
    121.     } else {
    122.       //错误消息
    123.       ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
    124.       out.writeUTF(res.getErrorMessage());
    125.       out.flushBuffer();
    126.       if (out instanceof Cleanable) {
    127.         ((Cleanable) out).cleanup();
    128.       }
    129.     }
    130.  
    131.     bos.flush();
    132.     bos.close();
    133.  
    134.     //写入的长度
    135.     int len = bos.writtenBytes();
    136.     //检查消息长度
    137.     checkPayload(channel, len);
    138.     Bytes.int2bytes(len, header, 12);
    139.     //重置写入位置
    140.     buffer.writerIndex(savedWriteIndex);
    141.     //写入消息头
    142.     buffer.writeBytes(header);
    143.     //buffer写出去的位置从writeIndex开始 加上header长度 数据长度
    144.     buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
    145.   } catch (Throwable t) {
    146.     // clear buffer
    147.     buffer.writerIndex(savedWriteIndex);
    148.     // send error message to Consumer, otherwise, Consumer will wait till timeout.
    149.     if (!res.isEvent() && res.getStatus() != Response.BAD_RESPONSE) {
    150.       Response r = new Response(res.getId(), res.getVersion());
    151.       r.setStatus(Response.BAD_RESPONSE);
    152.  
    153.       if (t instanceof ExceedPayloadLimitException) {
    154.         logger.warn(t.getMessage(), t);
    155.         try {
    156.           r.setErrorMessage(t.getMessage());
    157.           channel.send(r);
    158.           return;
    159.         } catch (RemotingException e) {
    160.           logger.warn("Failed to send bad_response info back: " + t.getMessage() + ", cause: " + e.getMessage(), e);
    161.         }
    162.       } else {
    163.         // FIXME log error message in Codec and handle in caught() of IoHanndler?
    164.         logger.warn("Fail to encode response: " + res + ", send bad_response info instead, cause: " + t.getMessage(), t);
    165.         try {
    166.           r.setErrorMessage("Failed to send response: " + res + ", cause: " + StringUtils.toString(t));
    167.           channel.send(r);
    168.           return;
    169.         } catch (RemotingException e) {
    170.           logger.warn("Failed to send bad_response info back: " + res + ", cause: " + e.getMessage(), e);
    171.         }
    172.       }
    173.     }
    174.  
    175.     // Rethrow exception
    176.     if (t instanceof IOException) {
    177.       throw (IOException) t;
    178.     } else if (t instanceof RuntimeException) {
    179.       throw (RuntimeException) t;
    180.     } else if (t instanceof Error) {
    181.       throw (Error) t;
    182.     } else {
    183.       throw new RuntimeException(t.getMessage(), t);
    184.     }
    185.   }
    186. }

    ExchangeCodec的decode方法是encode方法的逆过程，会先检查魔数，然后读取协议头和后续消息的长度，最后根据协议头中的各个标志位构造相应的对象，以及反序列化数据。

    DubboCodec

    在ExchangeCodecencode的encode方法中，不论是encodeRequest还是encodeResponse都调用encodeRequestData方法，该方法会对Boby内容进行编码，该方法实现是在DubboCodec中，因此DubboCodec是对消息体的编解码，接下来我们来看下encodeRequestData和encodeResponseData方法的实现，

    1. protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
    2.   RpcInvocation inv = (RpcInvocation) data;
    3.  
    4.   //dubbo服务版本
    5.   out.writeUTF(version);
    6.   // https://github.com/apache/dubbo/issues/6138
    7.   String serviceName = inv.getAttachment(INTERFACE_KEY);
    8.   if (serviceName == null) {
    9.     //服务path
    10.     serviceName = inv.getAttachment(PATH_KEY);
    11.   }
    12.   //服务名
    13.   out.writeUTF(serviceName);
    14.   //版本号
    15.   out.writeUTF(inv.getAttachment(VERSION_KEY));
    16.   //方法名
    17.   out.writeUTF(inv.getMethodName());
    18.   //方法类型描述
    19.   out.writeUTF(inv.getParameterTypesDesc());
    20.   Object[] args = inv.getArguments();
    21.   if (args != null) {
    22.     for (int i = 0; i < args.length; i++) {
    23.       //参数值
    24.       out.writeObject(encodeInvocationArgument(channel, inv, i));
    25.     }
    26.   }
    27.   //附加属性
    28.   out.writeAttachments(inv.getObjectAttachments());
    29. }
    30. @Override
    31.   protected void encodeResponseData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
    32.   Result result = (Result) data;
    33.   //检验版本
    34.   boolean attach = Version.isSupportResponseAttachment(version);
    35.   Throwable th = result.getException();
    36.   if (th == null) {
    37.     Object ret = result.getValue();
    38.     if (ret == null) {
    39.       //空结果
    40.       out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE);
    41.     } else {
    42.       //正常写入
    43.       out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE);
    44.       out.writeObject(ret);
    45.     }
    46.   } else {
    47.     //异常
    48.     out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION);
    49.     out.writeThrowable(th);
    50.   }
    51.  
    52.   if (attach) {
    53.     //Dubbo版本号
    54.     result.getObjectAttachments().put(DUBBO_VERSION_KEY, Version.getProtocolVersion());
    55.     out.writeAttachments(result.getObjectAttachments());
    56.   }
    57. }

    结束

    欢迎大家点点关注，点点赞!

  • 相关阅读:
    关于#单片机#的问题：第七行打印完他就会获取到温湿度，单片机的连接也正常，reset也没用单片机中烧写的完整代码如下图
    软件过程与管理学习之：项目计划（Project Schedule）
    一个非常简单的变分量子分类器 （VQC）
    C++之STL容器(2/3)
    shiro学习之HashedCredentialsMatcher密码匹配过程
    分享一个在linux中运行通义千问的方法
    Java 入门指南：JVM（Java虚拟机）——类的生命周期与加载过程
    微服务使用SockJs+Stomp实现Websocket 前后端实例 | Vuex形式断开重连、跨域等等问题踩坑（一）
    小程序源码：网课查题微信小程序源码下载,题库资源丰富自动采集，支持语音拍照识别
    从Oracle迁移数据到Hadoop
  • 原文地址：https://blog.csdn.net/weixin_38592881/article/details/127953674