暂无图片
暂无图片
暂无图片
暂无图片
暂无图片
首页 / dubbo编解码原理分析

dubbo编解码原理分析

shysheng 2019-01-27
457

在dubbo的一次请求、响应过程中，客户端和服务端之间的数据交换都涉及对数据的编解码工作，而这都是通过codec2接口来处理的，默认使用dubbo协议。本文将主要回答以下3个问题： 1.dubbo协议头格式 2.dubbo采用异步通信方式，那请求和响应是怎么关联的 3.编解码过程

1.Codec2接口

Codec2接口比较简单，除了负责编解码的2个方法之外，只有一个枚举DecodeResult用来对解码结果进行标识。

  • NEEDMOREINPUT：收到的字节流不是一个完整数据包，需要等待更多数据到达。

  • SKIPSOMEINPUT：忽略掉一部分数据包。

  1. @SPI
  2. public interface Codec2 {

  4.     @Adaptive({Constants.CODEC_KEY})
  5.     void encode(Channel channel, ChannelBuffer buffer, Object message) throws IOException;

  7.     @Adaptive({Constants.CODEC_KEY})
  8.     Object decode(Channel channel, ChannelBuffer buffer) throws IOException;

  10.     enum DecodeResult {
  11.         NEED_MORE_INPUT, SKIP_SOME_INPUT
  12.     }

  14. }
复制
2. ExchangeCodec核心属性

dubbo中编解码入口在DubboCountCodec中，最终都会调到ExchangeCodec，其核心属性简单介绍如下：

  1. // 协议头长度，16个字节
  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. protected static final byte FLAG_TWOWAY = (byte) 0x40;
  10. protected static final byte FLAG_EVENT = (byte) 0x20;
  11. // 5位序列化掩码，因此dubbo最多支持32种序列化协议
  12. protected static final int SERIALIZATION_MASK = 0x1f;
复制
3.dubbo协议头格式

dubbo官网使用下面这个图对dubbo协议头格式进行了介绍：

可以看到，dubbo协议头一共包含16个字节：

  • 2字节的魔数

  • 5位的序列化id

  • 1位事件标识

  • 1位单双向标识

  • 1位请求响应标识

  • 1个字节状态位

  • 8字节请求id

  • 4字节的消息体长度

4.编码源码解读

从DubboCountCodec#encode方法开始，最终会调到ExchangeCodec#encodeRequest方法，我们就从这里开始看。

4.1 客户端请求编码
  1. protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
  2.     // 获取序列化协议，默认为hessian2
  3.     Serialization serialization = getSerialization(channel);
  4.     // 协议头总长度为16字节
  5.     byte[] header = new byte[HEADER_LENGTH];
  6.     // 设置魔数
  7.     Bytes.short2bytes(MAGIC, header);

  9.     // 设置消息标记位，总长度为1个字节，前3位分别表示请求响应标识、单双向标识和事件标识，后5位标识序列化id
  10.     header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());

  12.     if (req.isTwoWay()) {
  13.         header[2] |= FLAG_TWOWAY;
  14.     }
  15.     if (req.isEvent()) {
  16.         header[2] |= FLAG_EVENT;
  17.     }

  19.     // 设置请求id
  20.     Bytes.long2bytes(req.getId(), header, 4);

  22.     // encode request data.
  23.     int savedWriteIndex = buffer.writerIndex();
  24.     buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
  25.     ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
  26.     ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
  27.     if (req.isEvent()) {
  28.         // 心跳事件
  29.         encodeEventData(channel, out, req.getData());
  30.     } else {
  31.         // 正常请求
  32.         encodeRequestData(channel, out, req.getData(), req.getVersion());
  33.     }
  34.     out.flushBuffer();
  35.     if (out instanceof Cleanable) {
  36.         ((Cleanable) out).cleanup();
  37.     }
  38.     bos.flush();
  39.     bos.close();
  40.     int len = bos.writtenBytes();
  41.     // 检查请求大小是否超过限制，最大8M
  42.     checkPayload(channel, len);
  43.     Bytes.int2bytes(len, header, 12);

  45.     // write
  46.     buffer.writerIndex(savedWriteIndex);
  47.     buffer.writeBytes(header); // write header.
  48.     buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
  49. }
复制

解读源码发现，对请求消息进行编码时，没有第4个字节的状态位，同时官网中关于dubbo协议头的结构对于第3字节的描述其实反了，正确的dubbo协议头格式如下：

在对请求进行编码时，会把版本号、服务路径、方法名、方法参数等信息都进行编码，其中参数类型是用JVM中的类型表示方法来编码的。

  1. @Override
  2. protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
  3.     RpcInvocation inv = (RpcInvocation) data;

  5.     out.writeUTF(version);
  6.     out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
  7.     out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));

  9.     out.writeUTF(inv.getMethodName());
  10.     out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
  11.     Object[] args = inv.getArguments();
  12.     if (args != null) {
  13.         for (int i = 0; i < args.length; i++) {
  14.             out.writeObject(encodeInvocationArgument(channel, inv, i));
  15.         }
  16.     }
  17.     out.writeObject(RpcUtils.getNecessaryAttachments(inv));
  18. }
复制
  1. public static String getDesc(Class<?> c) {
  2.     StringBuilder ret = new StringBuilder();

  4.     while (c.isArray()) {
  5.         ret.append('[');
  6.         c = c.getComponentType();
  7.     }

  9.     if (c.isPrimitive()) {
  10.         String t = c.getName();
  11.         if ("void".equals(t)) {
  12.             ret.append(JVM_VOID);
  13.         } else if ("boolean".equals(t)) {
  14.             ret.append(JVM_BOOLEAN);
  15.         } else if ("byte".equals(t)) {
  16.             ret.append(JVM_BYTE);
  17.         } else if ("char".equals(t)) {
  18.             ret.append(JVM_CHAR);
  19.         } else if ("double".equals(t)) {
  20.             ret.append(JVM_DOUBLE);
  21.         } else if ("float".equals(t)) {
  22.             ret.append(JVM_FLOAT);
  23.         } else if ("int".equals(t)) {
  24.             ret.append(JVM_INT);
  25.         } else if ("long".equals(t)) {
  26.             ret.append(JVM_LONG);
  27.         } else if ("short".equals(t)) {
  28.             ret.append(JVM_SHORT);
  29.         }
  30.     } else {
  31.         ret.append('L');
  32.         ret.append(c.getName().replace('.', '/'));
  33.         ret.append(';');
  34.     }
  35.     return ret.toString();
  36. }
复制
4.2 服务端响应编码

服务端对响应进行编码的逻辑与客户端类似，具体见代码注释：

  1. protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
  2.     int savedWriteIndex = buffer.writerIndex();
  3.     try {
  4.         // 获取序列化协议、初始化等等
  5.         Serialization serialization = getSerialization(channel);
  6.         byte[] header = new byte[HEADER_LENGTH];
  7.         Bytes.short2bytes(MAGIC, header);
  8.         header[2] = serialization.getContentTypeId();
  9.         if (res.isHeartbeat()) {
  10.             header[2] |= FLAG_EVENT;
  11.         }
  12.         // 对响应进行编码，会有第3字节的响应状态位
  13.         byte status = res.getStatus();
  14.         header[3] = status;
  15.         Bytes.long2bytes(res.getId(), header, 4);

  17.         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
  18.         ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
  19.         ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
  20.         // encode response data or error message.
  21.         if (status == Response.OK) {
  22.             if (res.isHeartbeat()) {
  23.                 encodeHeartbeatData(channel, out, res.getResult());
  24.             } else {
  25.                 encodeResponseData(channel, out, res.getResult(), res.getVersion());
  26.             }
  27.         } else {
  28.             out.writeUTF(res.getErrorMessage());
  29.         }
  30.         ...
  31. }
复制
  1. @Override
  2. protected void encodeResponseData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
  3.     Result result = (Result) data;
  4.     // 2.0.2之后版本的响应中，才会存在附加属性，这里会根据版本号来判断是否需要将附加属性进行编码
  5.     boolean attach = Version.isSupportResponseAttatchment(version);
  6.     Throwable th = result.getException();
  7.     if (th == null) {
  8.         Object ret = result.getValue();
  9.         if (ret == null) {
  10.             out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE);
  11.         } else {
  12.             out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE);
  13.             out.writeObject(ret);
  14.         }
  15.     } else {
  16.         out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION);
  17.         out.writeObject(th);
  18.     }

  20.     if (attach) {
  21.         // returns current version of Response to consumer side.
  22.         result.getAttachments().put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());
  23.         out.writeObject(result.getAttachments());
  24.     }
  25. }
复制
5.解码源码解读

解码逻辑从DubboCountCodec#decode方法开始，最终会调到ExchangeCodec#decode方法，从这里一步步往下看。

  1. @Override
  2. protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
  3.     // 检查魔数，如果不是dubbo协议，则使用父类的解码方法
  4.     if (readable > 0 && header[0] != MAGIC_HIGH
  5.             || readable > 1 && header[1] != MAGIC_LOW) {
  6.         int length = header.length;
  7.         if (header.length < readable) {
  8.             header = Bytes.copyOf(header, readable);
  9.             buffer.readBytes(header, length, readable - length);
  10.         }
  11.         for (int i = 1; i < header.length - 1; i++) {
  12.             if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
  13.                 buffer.readerIndex(buffer.readerIndex() - header.length + i);
  14.                 header = Bytes.copyOf(header, i);
  15.                 break;
  16.             }
  17.         }
  18.         return super.decode(channel, buffer, readable, header);
  19.     }
  20.     // 长度不够，放弃本次解码，等待读取更多字节
  21.     if (readable < HEADER_LENGTH) {
  22.         return DecodeResult.NEED_MORE_INPUT;
  23.     }

  25.     // 读取具体编码数据
  26.     int len = Bytes.bytes2int(header, 12);
  27.     checkPayload(channel, len);

  29.     // 长度不够，放弃本次解码，等待读取更多字节
  30.     int tt = len + HEADER_LENGTH;
  31.     if (readable < tt) {
  32.         return DecodeResult.NEED_MORE_INPUT;
  33.     }

  35.     // limit input stream.
  36.     ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);

  38.     try {
  39.         return decodeBody(channel, is, header);
  40.     } finally {
  41.     ...
  42.     }
  43. }
复制

decodeBody分为两部分，客户端对响应的解码和服务端队请求的解码，具体是请求还是响应就是根据协议头中的请求响应标志位来区分的，源码如下：

  1. @Override
  2. protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
  3.     // 读取消息标记为和协议类型
  4.     byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
  5.     // 读取请求id
  6.     long id = Bytes.bytes2long(header, 4);
  7.     if ((flag & FLAG_REQUEST) == 0) {
  8.     // 客户端对响应进行解码
  9.     }else {
  10.     // 服务端对请求进行解码
  11.     }
  12. }
复制
5.1 客户端响应解码
  1. // 客户端对响应进行解码
  2. Response res = new Response(id);
  3. if ((flag & FLAG_EVENT) != 0) {
  4.     res.setEvent(Response.HEARTBEAT_EVENT);
  5. }
  6. // 读取响应状态
  7. byte status = header[3];
  8. res.setStatus(status);
  9. try {
  10.     // 获取序列化协议，对输入数据进行解码，这里会根据序列化协议id和名称来校验序列化协议是否合法
  11.     ObjectInput in = CodecSupport.deserialize(channel.getUrl(), is, proto);
  12.     if (status == Response.OK) {
  13.         Object data;
  14.         if (res.isHeartbeat()) {
  15.             data = decodeHeartbeatData(channel, in);
  16.         } else if (res.isEvent()) {
  17.             data = decodeEventData(channel, in);
  18.         } else {
  19.             // 根据decode.in.io的配置决定何时进行解码
  20.             DecodeableRpcResult result;
  21.             if (channel.getUrl().getParameter(
  22.                     Constants.DECODE_IN_IO_THREAD_KEY,
  23.                     Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
  24.                 result = new DecodeableRpcResult(channel, res, is,
  25.                         (Invocation) getRequestData(id), proto);
  26.                 result.decode();
  27.             } else {
  28.                 result = new DecodeableRpcResult(channel, res,
  29.                         new UnsafeByteArrayInputStream(readMessageData(is)),
  30.                         (Invocation) getRequestData(id), proto);
  31.             }
  32.             data = result;
  33.         }
  34.     ...
  35. }
  36. return res;
复制
5.2 服务端请求解码

解码过程与客户端对响应的解码基本一样，这里就不再赘述。


数据库
文章转载自shysheng，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

评论

相关阅读

数据库国产化替代深化：DBA的机遇与挑战
代晓磊
1288次阅读
2025-04-27 16:53:22
2025年4月国产数据库中标情况一览：4个千万元级项目，GaussDB与OceanBase大放异彩！
通讯员
771次阅读
2025-04-30 15:24:06
国产数据库需要扩大场景覆盖面才能在竞争中更有优势
白鳝的洞穴
627次阅读
2025-04-14 09:40:20
【活动】分享你的压箱底干货文档，三篇解锁进阶奖励！
墨天轮编辑部
529次阅读
2025-04-17 17:02:24
一页概览：Oracle GoldenGate
甲骨文云技术
493次阅读
2025-04-30 12:17:56
GoldenDB数据库v7.2焕新发布，助力全行业数据库平滑替代
GoldenDB分布式数据库
477次阅读
2025-04-30 12:17:50
优炫数据库成功入围新疆维吾尔自治区行政事业单位数据库2025年框架协议采购！
优炫软件
365次阅读
2025-04-18 10:01:22
给准备学习国产数据库的朋友几点建议
白鳝的洞穴
343次阅读
2025-05-07 10:06:14
XCOPS广州站：从开源自研之争到AI驱动的下一代数据库架构探索
韩锋频道
305次阅读
2025-04-29 10:35:54
MySQL 30 周年庆！MySQL 8.4 认证免费考！这次是认真的。。。
数据库运维之道
284次阅读
2025-04-28 11:01:25