2.2 服务消费方发送请求

2.2.1 发送请求

本节我们来看一下同步调用模式下，服务消费方是如何发送调用请求的。在深入分析源码前，我们先来看一张图。

2.2 服务消费方发送请求 - 图1

这张图展示了服务消费方发送请求过程的部分调用栈，略为复杂。从上图可以看出，经过多次调用后，才将请求数据送至 Netty NioClientSocketChannel。这样做的原因是通过 Exchange 层为框架引入 Request 和 Response 语义，这一点会在接下来的源码分析过程中会看到。其他的就不多说了，下面开始进行分析。首先分析 ReferenceCountExchangeClient 的源码。

  1. final class ReferenceCountExchangeClient implements ExchangeClient {
  3.     private final URL url;
  4.     private final AtomicInteger referenceCount = new AtomicInteger(0);
  6.     public ReferenceCountExchangeClient(ExchangeClient client, ConcurrentMap<String, LazyConnectExchangeClient> ghostClientMap) {
  7.         this.client = client;
  8.         // 引用计数自增
  9.         referenceCount.incrementAndGet();
  10.         this.url = client.getUrl();
  12.         // ...
  13.     }
  15.     @Override
  16.     public ResponseFuture request(Object request) throws RemotingException {
  17.         // 直接调用被装饰对象的同签名方法
  18.         return client.request(request);
  19.     }
  21.     @Override
  22.     public ResponseFuture request(Object request, int timeout) throws RemotingException {
  23.         // 直接调用被装饰对象的同签名方法
  24.         return client.request(request, timeout);
  25.     }
  27.     /** 引用计数自增，该方法由外部调用 */
  28.     public void incrementAndGetCount() {
  29.         // referenceCount 自增
  30.         referenceCount.incrementAndGet();
  31.     }
  33.         @Override
  34.     public void close(int timeout) {
  35.         // referenceCount 自减
  36.         if (referenceCount.decrementAndGet() <= 0) {
  37.             if (timeout == 0) {
  38.                 client.close();
  39.             } else {
  40.                 client.close(timeout);
  41.             }
  42.             client = replaceWithLazyClient();
  43.         }
  44.     }
  46.     // 省略部分方法
  47. }

ReferenceCountExchangeClient 内部定义了一个引用计数变量 referenceCount，每当该对象被引用一次 referenceCount 都会进行自增。每当 close 方法被调用时，referenceCount 进行自减。ReferenceCountExchangeClient 内部仅实现了一个引用计数的功能，其他方法并无复杂逻辑，均是直接调用被装饰对象的相关方法。所以这里就不多说了，继续向下分析，这次是 HeaderExchangeClient。

  1. public class HeaderExchangeClient implements ExchangeClient {
  3.     private static final ScheduledThreadPoolExecutor scheduled = new ScheduledThreadPoolExecutor(2, new NamedThreadFactory("dubbo-remoting-client-heartbeat", true));
  4.     private final Client client;
  5.     private final ExchangeChannel channel;
  6.     private ScheduledFuture<?> heartbeatTimer;
  7.     private int heartbeat;
  8.     private int heartbeatTimeout;
  10.     public HeaderExchangeClient(Client client, boolean needHeartbeat) {
  11.         if (client == null) {
  12.             throw new IllegalArgumentException("client == null");
  13.         }
  14.         this.client = client;
  16.         // 创建 HeaderExchangeChannel 对象
  17.         this.channel = new HeaderExchangeChannel(client);
  19.         // 以下代码均与心跳检测逻辑有关
  20.         String dubbo = client.getUrl().getParameter(Constants.DUBBO_VERSION_KEY);
  21.         this.heartbeat = client.getUrl().getParameter(Constants.HEARTBEAT_KEY, dubbo != null && dubbo.startsWith("1.0.") ? Constants.DEFAULT_HEARTBEAT : 0);
  22.         this.heartbeatTimeout = client.getUrl().getParameter(Constants.HEARTBEAT_TIMEOUT_KEY, heartbeat * 3);
  23.         if (heartbeatTimeout < heartbeat * 2) {
  24.             throw new IllegalStateException("heartbeatTimeout < heartbeatInterval * 2");
  25.         }
  26.         if (needHeartbeat) {
  27.             // 开启心跳检测定时器
  28.             startHeartbeatTimer();
  29.         }
  30.     }
  32.     @Override
  33.     public ResponseFuture request(Object request) throws RemotingException {
  34.         // 直接 HeaderExchangeChannel 对象的同签名方法
  35.         return channel.request(request);
  36.     }
  38.     @Override
  39.     public ResponseFuture request(Object request, int timeout) throws RemotingException {
  40.         // 直接 HeaderExchangeChannel 对象的同签名方法
  41.         return channel.request(request, timeout);
  42.     }
  44.     @Override
  45.     public void close() {
  46.         doClose();
  47.         channel.close();
  48.     }
  50.     private void doClose() {
  51.         // 停止心跳检测定时器
  52.         stopHeartbeatTimer();
  53.     }
  55.     private void startHeartbeatTimer() {
  56.         stopHeartbeatTimer();
  57.         if (heartbeat > 0) {
  58.             heartbeatTimer = scheduled.scheduleWithFixedDelay(
  59.                     new HeartBeatTask(new HeartBeatTask.ChannelProvider() {
  60.                         @Override
  61.                         public Collection<Channel> getChannels() {
  62.                             return Collections.<Channel>singletonList(HeaderExchangeClient.this);
  63.                         }
  64.                     }, heartbeat, heartbeatTimeout),
  65.                     heartbeat, heartbeat, TimeUnit.MILLISECONDS);
  66.         }
  67.     }
  69.     private void stopHeartbeatTimer() {
  70.         if (heartbeatTimer != null && !heartbeatTimer.isCancelled()) {
  71.             try {
  72.                 heartbeatTimer.cancel(true);
  73.                 scheduled.purge();
  74.             } catch (Throwable e) {
  75.                 if (logger.isWarnEnabled()) {
  76.                     logger.warn(e.getMessage(), e);
  77.                 }
  78.             }
  79.         }
  80.         heartbeatTimer = null;
  81.     }
  83.     // 省略部分方法
  84. }

HeaderExchangeClient 中很多方法只有一行代码，即调用 HeaderExchangeChannel 对象的同签名方法。那 HeaderExchangeClient 有什么用处呢？答案是封装了一些关于心跳检测的逻辑。心跳检测并非本文所关注的点，因此就不多说了，继续向下看。

  1. final class HeaderExchangeChannel implements ExchangeChannel {
  3.     private final Channel channel;
  5.     HeaderExchangeChannel(Channel channel) {
  6.         if (channel == null) {
  7.             throw new IllegalArgumentException("channel == null");
  8.         }
  10.         // 这里的 channel 指向的是 NettyClient
  11.         this.channel = channel;
  12.     }
  14.     @Override
  15.     public ResponseFuture request(Object request) throws RemotingException {
  16.         return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
  17.     }
  19.     @Override
  20.     public ResponseFuture request(Object request, int timeout) throws RemotingException {
  21.         if (closed) {
  22.             throw new RemotingException(..., "Failed to send request ...);
  23.         }
  24.         // 创建 Request 对象
  25.         Request req = new Request();
  26.         req.setVersion(Version.getProtocolVersion());
  27.         // 设置双向通信标志为 true
  28.         req.setTwoWay(true);
  29.         // 这里的 request 变量类型为 RpcInvocation
  30.         req.setData(request);
  32.         // 创建 DefaultFuture 对象
  33.         DefaultFuture future = new DefaultFuture(channel, req, timeout);
  34.         try {
  35.             // 调用 NettyClient 的 send 方法发送请求
  36.             channel.send(req);
  37.         } catch (RemotingException e) {
  38.             future.cancel();
  39.             throw e;
  40.         }
  41.         // 返回 DefaultFuture 对象
  42.         return future;
  43.     }
  44. }

到这里大家终于看到了 Request 语义了，上面的方法首先定义了一个 Request 对象，然后再将该对象传给 NettyClient 的 send 方法，进行后续的调用。需要说明的是，NettyClient 中并未实现 send 方法，该方法继承自父类 AbstractPeer，下面直接分析 AbstractPeer 的代码。

  1. public abstract class AbstractPeer implements Endpoint, ChannelHandler {
  3.     @Override
  4.     public void send(Object message) throws RemotingException {
  5.         // 该方法由 AbstractClient 类实现
  6.         send(message, url.getParameter(Constants.SENT_KEY, false));
  7.     }
  9.     // 省略其他方法
  10. }
  12. public abstract class AbstractClient extends AbstractEndpoint implements Client {
  14.     @Override
  15.     public void send(Object message, boolean sent) throws RemotingException {
  16.         if (send_reconnect && !isConnected()) {
  17.             connect();
  18.         }
  20.         // 获取 Channel，getChannel 是一个抽象方法，具体由子类实现
  21.         Channel channel = getChannel();
  22.         if (channel == null || !channel.isConnected()) {
  23.             throw new RemotingException(this, "message can not send ...");
  24.         }
  26.         // 继续向下调用
  27.         channel.send(message, sent);
  28.     }
  30.     protected abstract Channel getChannel();
  32.     // 省略其他方法
  33. }

默认情况下，Dubbo 使用 Netty 作为底层的通信框架，因此下面我们到 NettyClient 类中看一下 getChannel 方法的实现逻辑。

  2. public class NettyClient extends AbstractClient {
  4.     // 这里的 Channel 全限定名称为 org.jboss.netty.channel.Channel
  5.     private volatile Channel channel;
  7.     @Override
  8.     protected com.alibaba.dubbo.remoting.Channel getChannel() {
  9.         Channel c = channel;
  10.         if (c == null || !c.isConnected())
  11.             return null;
  12.         // 获取一个 NettyChannel 类型对象
  13.         return NettyChannel.getOrAddChannel(c, getUrl(), this);
  14.     }
  15. }
  17. final class NettyChannel extends AbstractChannel {
  19.     private static final ConcurrentMap<org.jboss.netty.channel.Channel, NettyChannel> channelMap = 
  20.         new ConcurrentHashMap<org.jboss.netty.channel.Channel, NettyChannel>();
  22.     private final org.jboss.netty.channel.Channel channel;
  24.     /** 私有构造方法 */
  25.     private NettyChannel(org.jboss.netty.channel.Channel channel, URL url, ChannelHandler handler) {
  26.         super(url, handler);
  27.         if (channel == null) {
  28.             throw new IllegalArgumentException("netty channel == null;");
  29.         }
  30.         this.channel = channel;
  31.     }
  33.     static NettyChannel getOrAddChannel(org.jboss.netty.channel.Channel ch, URL url, ChannelHandler handler) {
  34.         if (ch == null) {
  35.             return null;
  36.         }
  38.         // 尝试从集合中获取 NettyChannel 实例
  39.         NettyChannel ret = channelMap.get(ch);
  40.         if (ret == null) {
  41.             // 如果 ret = null，则创建一个新的 NettyChannel 实例
  42.             NettyChannel nc = new NettyChannel(ch, url, handler);
  43.             if (ch.isConnected()) {
  44.                 // 将 <Channel, NettyChannel> 键值对存入 channelMap 集合中
  45.                 ret = channelMap.putIfAbsent(ch, nc);
  46.             }
  47.             if (ret == null) {
  48.                 ret = nc;
  49.             }
  50.         }
  51.         return ret;
  52.     }
  53. }

获取到 NettyChannel 实例后，即可进行后续的调用。下面看一下 NettyChannel 的 send 方法。

  1. public void send(Object message, boolean sent) throws RemotingException {
  2.     super.send(message, sent);
  4.     boolean success = true;
  5.     int timeout = 0;
  6.     try {
  7.         // 发送消息(包含请求和响应消息)
  8.         ChannelFuture future = channel.write(message);
  10.         // sent 的值源于 &lt;dubbo:method sent="true/false" /> 中 sent 的配置值，有两种配置值：
  11.         //   1. true: 等待消息发出，消息发送失败将抛出异常
  12.         //   2. false: 不等待消息发出，将消息放入 IO 队列，即刻返回
  13.         // 默认情况下 sent = false；
  14.         if (sent) {
  15.             timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
  16.             // 等待消息发出，若在规定时间没能发出，success 会被置为 false
  17.             success = future.await(timeout);
  18.         }
  19.         Throwable cause = future.getCause();
  20.         if (cause != null) {
  21.             throw cause;
  22.         }
  23.     } catch (Throwable e) {
  24.         throw new RemotingException(this, "Failed to send message ...");
  25.     }
  27.     // 若 success 为 false，这里抛出异常
  28.     if (!success) {
  29.         throw new RemotingException(this, "Failed to send message ...");
  30.     }
  31. }

经历多次调用，到这里请求数据的发送过程就结束了，过程漫长。为了便于大家阅读代码，这里以 DemoService 为例，将 sayHello 方法的整个调用路径贴出来。

  1. proxy0#sayHello(String)
  2.   —> InvokerInvocationHandler#invoke(Object, Method, Object[])
  3.     —> MockClusterInvoker#invoke(Invocation)
  4.       —> AbstractClusterInvoker#invoke(Invocation)
  5.         —> FailoverClusterInvoker#doInvoke(Invocation, List<Invoker<T>>, LoadBalance)
  6.           —> Filter#invoke(Invoker, Invocation)  // 包含多个 Filter 调用
  7.             —> ListenerInvokerWrapper#invoke(Invocation) 
  8.               —> AbstractInvoker#invoke(Invocation) 
  9.                 —> DubboInvoker#doInvoke(Invocation)
  10.                   —> ReferenceCountExchangeClient#request(Object, int)
  11.                     —> HeaderExchangeClient#request(Object, int)
  12.                       —> HeaderExchangeChannel#request(Object, int)
  13.                         —> AbstractPeer#send(Object)
  14.                           —> AbstractClient#send(Object, boolean)
  15.                             —> NettyChannel#send(Object, boolean)
  16.                               —> NioClientSocketChannel#write(Object)

在 Netty 中，出站数据在发出之前还需要进行编码操作，接下来我们来分析一下请求数据的编码逻辑。

2.2.2 请求编码

在分析请求编码逻辑之前，我们先来看一下 Dubbo 数据包结构。

2.2 服务消费方发送请求 - 图2

Dubbo 数据包分为消息头和消息体，消息头用于存储一些元信息，比如魔数（Magic），数据包类型（Request/Response），消息体长度（Data Length）等。消息体中用于存储具体的调用消息，比如方法名称，参数列表等。下面简单列举一下消息头的内容。

偏移量(Bit)字段取值
0 ~ 7魔数高位0xda00
8 ~ 15魔数低位0xbb
16数据包类型0 - Response, 1 - Request
17调用方式仅在第16位被设为1的情况下有效，0 - 单向调用，1 - 双向调用
18事件标识0 - 当前数据包是请求或响应包，1 - 当前数据包是心跳包
19 ~ 23序列化器编号2 - Hessian2Serialization
3 - JavaSerialization
4 - CompactedJavaSerialization
6 - FastJsonSerialization
7 - NativeJavaSerialization
8 - KryoSerialization
9 - FstSerialization
24 ~ 31状态20 - OK
30 - CLIENT_TIMEOUT
31 - SERVER_TIMEOUT
40 - BAD_REQUEST
50 - BAD_RESPONSE
……
32 ~ 95请求编号共8字节，运行时生成
96 ~ 127消息体长度运行时计算

了解了 Dubbo 数据包格式，接下来我们就可以探索编码过程了。这次我们开门见山，直接分析编码逻辑所在类。如下：

  1. public class ExchangeCodec extends TelnetCodec {
  3.     // 消息头长度
  4.     protected static final int HEADER_LENGTH = 16;
  5.     // 魔数内容
  6.     protected static final short MAGIC = (short) 0xdabb;
  7.     protected static final byte MAGIC_HIGH = Bytes.short2bytes(MAGIC)[0];
  8.     protected static final byte MAGIC_LOW = Bytes.short2bytes(MAGIC)[1];
  9.     protected static final byte FLAG_REQUEST = (byte) 0x80;
  10.     protected static final byte FLAG_TWOWAY = (byte) 0x40;
  11.     protected static final byte FLAG_EVENT = (byte) 0x20;
  12.     protected static final int SERIALIZATION_MASK = 0x1f;
  13.     private static final Logger logger = LoggerFactory.getLogger(ExchangeCodec.class);
  15.     public Short getMagicCode() {
  16.         return MAGIC;
  17.     }
  19.     @Override
  20.     public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException {
  21.         if (msg instanceof Request) {
  22.             // 对 Request 对象进行编码
  23.             encodeRequest(channel, buffer, (Request) msg);
  24.         } else if (msg instanceof Response) {
  25.             // 对 Response 对象进行编码，后面分析
  26.             encodeResponse(channel, buffer, (Response) msg);
  27.         } else {
  28.             super.encode(channel, buffer, msg);
  29.         }
  30.     }
  32.     protected void encodeRequest(Channel channel, ChannelBuffer buffer, Request req) throws IOException {
  33.         Serialization serialization = getSerialization(channel);
  35.         // 创建消息头字节数组，长度为 16
  36.         byte[] header = new byte[HEADER_LENGTH];
  38.         // 设置魔数
  39.         Bytes.short2bytes(MAGIC, header);
  41.         // 设置数据包类型（Request/Response）和序列化器编号
  42.         header[2] = (byte) (FLAG_REQUEST | serialization.getContentTypeId());
  44.         // 设置通信方式(单向/双向)
  45.         if (req.isTwoWay()) {
  46.             header[2] |= FLAG_TWOWAY;
  47.         }
  49.         // 设置事件标识
  50.         if (req.isEvent()) {
  51.             header[2] |= FLAG_EVENT;
  52.         }
  54.         // 设置请求编号，8个字节，从第4个字节开始设置
  55.         Bytes.long2bytes(req.getId(), header, 4);
  57.         // 获取 buffer 当前的写位置
  58.         int savedWriteIndex = buffer.writerIndex();
  59.         // 更新 writerIndex，为消息头预留 16 个字节的空间
  60.         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH);
  61.         ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer);
  62.         // 创建序列化器，比如 Hessian2ObjectOutput
  63.         ObjectOutput out = serialization.serialize(channel.getUrl(), bos);
  64.         if (req.isEvent()) {
  65.             // 对事件数据进行序列化操作
  66.             encodeEventData(channel, out, req.getData());
  67.         } else {
  68.             // 对请求数据进行序列化操作
  69.             encodeRequestData(channel, out, req.getData(), req.getVersion());
  70.         }
  71.         out.flushBuffer();
  72.         if (out instanceof Cleanable) {
  73.             ((Cleanable) out).cleanup();
  74.         }
  75.         bos.flush();
  76.         bos.close();
  78.         // 获取写入的字节数，也就是消息体长度
  79.         int len = bos.writtenBytes();
  80.         checkPayload(channel, len);
  82.         // 将消息体长度写入到消息头中
  83.         Bytes.int2bytes(len, header, 12);
  85.         // 将 buffer 指针移动到 savedWriteIndex，为写消息头做准备
  86.         buffer.writerIndex(savedWriteIndex);
  87.         // 从 savedWriteIndex 下标处写入消息头
  88.         buffer.writeBytes(header);
  89.         // 设置新的 writerIndex，writerIndex = 原写下标 + 消息头长度 + 消息体长度
  90.         buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len);
  91.     }
  93.     // 省略其他方法
  94. }

以上就是请求对象的编码过程，该过程首先会通过位运算将消息头写入到 header 数组中。然后对 Request 对象的 data 字段执行序列化操作，序列化后的数据最终会存储到 ChannelBuffer 中。序列化操作执行完后，可得到数据序列化后的长度 len，紧接着将 len 写入到 header 指定位置处。最后再将消息头字节数组 header 写入到 ChannelBuffer 中，整个编码过程就结束了。本节的最后，我们再来看一下 Request 对象的 data 字段序列化过程，也就是 encodeRequestData 方法的逻辑，如下：

  1. public class DubboCodec extends ExchangeCodec implements Codec2 {
  3.     protected void encodeRequestData(Channel channel, ObjectOutput out, Object data, String version) throws IOException {
  4.         RpcInvocation inv = (RpcInvocation) data;
  6.         // 依次序列化 dubbo version、path、version
  7.         out.writeUTF(version);
  8.         out.writeUTF(inv.getAttachment(Constants.PATH_KEY));
  9.         out.writeUTF(inv.getAttachment(Constants.VERSION_KEY));
  11.         // 序列化调用方法名
  12.         out.writeUTF(inv.getMethodName());
  13.         // 将参数类型转换为字符串，并进行序列化
  14.         out.writeUTF(ReflectUtils.getDesc(inv.getParameterTypes()));
  15.         Object[] args = inv.getArguments();
  16.         if (args != null)
  17.             for (int i = 0; i < args.length; i++) {
  18.                 // 对运行时参数进行序列化
  19.                 out.writeObject(encodeInvocationArgument(channel, inv, i));
  20.             }
  22.         // 序列化 attachments
  23.         out.writeObject(inv.getAttachments());
  24.     }
  25. }

至此，关于服务消费方发送请求的过程就分析完了，接下来我们来看一下服务提供方是如何接收请求的。