2.4 服务提供方返回调用结果

服务提供方调用指定服务后，会将调用结果封装到 Response 对象中，并将该对象返回给服务消费方。服务提供方也是通过 NettyChannel 的 send 方法将 Response 对象返回，这个方法在 2.2.1 节分析过，这里就不在重复分析了。本节我们仅需关注 Response 对象的编码过程即可，这里仍然省略一些中间调用，直接分析具体的编码逻辑。

  1. public class ExchangeCodec extends TelnetCodec {
  2.     public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
  3.         if (msg instanceof Request) {
  4.             encodeRequest(channel, buffer, (Request) msg);
  5.         } else if (msg instanceof Response) {
  6.             // 对响应对象进行编码
  7.             encodeResponse(channel, buffer, (Response) msg);
  8.         } else {
  9.             super.encode(channel, buffer, msg);
  10.         }
  11.     }
  13.     protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException {
  14.         int savedWriteIndex = buffer.writerIndex();
  15.         try {
  16.             Serialization serialization = getSerialization(channel);
  17.             // 创建消息头字节数组
  18.             byte[] header = new byte[HEADER_LENGTH];
  19.             // 设置魔数
  20.             Bytes.short2bytes(MAGIC, header);
  21.             // 设置序列化器编号
  22.             header[2] = serialization.getContentTypeId();
  23.             if (res.isHeartbeat()) header[2] |= FLAG_EVENT;
  24.             // 获取响应状态
  25.             byte status = res.getStatus();
  26.             // 设置响应状态
  27.             header[3] = status;
  28.             // 设置请求编号
  29.             Bytes.long2bytes(res.getId(), header, 4);
  31.             // 更新 writerIndex，为消息头预留 16 个字节的空间
  32.             buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
  33.             ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
  34.             ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
  36.             if (status == Response.OK) {
  37.                 if (res.isHeartbeat()) {
  38.                     // 对心跳响应结果进行序列化，已废弃
  39.                     encodeHeartbeatData(channel, out, res.getResult());
  40.                 } else {
  41.                     // 对调用结果进行序列化
  42.                     encodeResponseData(channel, out, res.getResult(), res.getVersion());
  43.                 }
  44.             } else { 
  45.                 // 对错误信息进行序列化
  46.                 out.writeUTF(res.getErrorMessage())
  47.             };
  48.             out.flushBuffer();
  49.             if (out instanceof Cleanable) {
  50.                 ((Cleanable) out).cleanup();
  51.             }
  52.             bos.flush();
  53.             bos.close();
  55.             // 获取写入的字节数，也就是消息体长度
  56.             int len = bos.writtenBytes();
  57.             checkPayload(channel, len);
  59.             // 将消息体长度写入到消息头中
  60.             Bytes.int2bytes(len, header, 12);
  61.             // 将 buffer 指针移动到 savedWriteIndex，为写消息头做准备
  62.             buffer.writerIndex(savedWriteIndex);
  63.             // 从 savedWriteIndex 下标处写入消息头
  64.             buffer.writeBytes(header); 
  65.             // 设置新的 writerIndex，writerIndex = 原写下标 + 消息头长度 + 消息体长度
  66.             buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
  67.         } catch (Throwable t) {
  68.             // 异常处理逻辑不是很难理解，但是代码略多，这里忽略了
  69.         }
  70.     }
  71. }
  73. public class DubboCodec extends ExchangeCodec implements Codec2 {
  75.     protected void encodeResponseData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
  76.         Result result = (Result) data;
  77.         // 检测当前协议版本是否支持带有 attachment 集合的 Response 对象
  78.         boolean attach = Version.isSupportResponseAttachment(version);
  79.         Throwable th = result.getException();
  81.         // 异常信息为空
  82.         if (th == null) {
  83.             Object ret = result.getValue();
  84.             // 调用结果为空
  85.             if (ret == null) {
  86.                 // 序列化响应类型
  87.                 out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE);
  88.             } 
  89.             // 调用结果非空
  90.             else {
  91.                 // 序列化响应类型
  92.                 out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE);
  93.                 // 序列化调用结果
  94.                 out.writeObject(ret);
  95.             }
  96.         } 
  97.         // 异常信息非空
  98.         else {
  99.             // 序列化响应类型
  100.             out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION);
  101.             // 序列化异常对象
  102.             out.writeObject(th);
  103.         }
  105.         if (attach) {
  106.             // 记录 Dubbo 协议版本
  107.             result.getAttachments().put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion());
  108.             // 序列化 attachments 集合
  109.             out.writeObject(result.getAttachments());
  110.         }
  111.     }
  112. }

以上就是 Response 对象编码的过程，和前面分析的 Request 对象编码过程很相似。如果大家能看 Request 对象的编码逻辑，那么这里的 Response 对象的编码逻辑也不难理解，就不多说了。接下来我们再来分析双向通信的最后一环 —— 服务消费方接收调用结果。