2.3 服务提供方接收请求

前面说过，默认情况下 Dubbo 使用 Netty 作为底层的通信框架。Netty 检测到有数据入站后，首先会通过解码器对数据进行解码，并将解码后的数据传递给下一个入站处理器的指定方法。所以在进行后续的分析之前，我们先来看一下数据解码过程。

2.3.1 请求解码

这里直接分析请求数据的解码逻辑，忽略中间过程，如下：

  1. public class ExchangeCodec extends TelnetCodec {
  3.     @Override
  4.     public Object decode(Channel channel, ChannelBuffer buffer) throws IOException {
  5.         int readable = buffer.readableBytes();
  6.         // 创建消息头字节数组
  7.         byte[] header = new byte[Math.min(readable, HEADER_LENGTH)];
  8.         // 读取消息头数据
  9.         buffer.readBytes(header);
  10.         // 调用重载方法进行后续解码工作
  11.         return decode(channel, buffer, readable, header);
  12.     }
  14.     @Override
  15.     protected Object decode(Channel channel, ChannelBuffer buffer, int readable, byte[] header) throws IOException {
  16.         // 检查魔数是否相等
  17.         if (readable > 0 && header[0] != MAGIC_HIGH
  18.                 || readable > 1 && header[1] != MAGIC_LOW) {
  19.             int length = header.length;
  20.             if (header.length < readable) {
  21.                 header = Bytes.copyOf(header, readable);
  22.                 buffer.readBytes(header, length, readable - length);
  23.             }
  24.             for (int i = 1; i < header.length - 1; i++) {
  25.                 if (header[i] == MAGIC_HIGH && header[i + 1] == MAGIC_LOW) {
  26.                     buffer.readerIndex(buffer.readerIndex() - header.length + i);
  27.                     header = Bytes.copyOf(header, i);
  28.                     break;
  29.                 }
  30.             }
  31.             // 通过 telnet 命令行发送的数据包不包含消息头，所以这里
  32.             // 调用 TelnetCodec 的 decode 方法对数据包进行解码
  33.             return super.decode(channel, buffer, readable, header);
  34.         }
  36.         // 检测可读数据量是否少于消息头长度，若小于则立即返回 DecodeResult.NEED_MORE_INPUT
  37.         if (readable < HEADER_LENGTH) {
  38.             return DecodeResult.NEED_MORE_INPUT;
  39.         }
  41.         // 从消息头中获取消息体长度
  42.         int len = Bytes.bytes2int(header, 12);
  43.         // 检测消息体长度是否超出限制，超出则抛出异常
  44.         checkPayload(channel, len);
  46.         int tt = len + HEADER_LENGTH;
  47.         // 检测可读的字节数是否小于实际的字节数
  48.         if (readable < tt) {
  49.             return DecodeResult.NEED_MORE_INPUT;
  50.         }
  52.         ChannelBufferInputStream is = new ChannelBufferInputStream(buffer, len);
  54.         try {
  55.             // 继续进行解码工作
  56.             return decodeBody(channel, is, header);
  57.         } finally {
  58.             if (is.available() > 0) {
  59.                 try {
  60.                     StreamUtils.skipUnusedStream(is);
  61.                 } catch (IOException e) {
  62.                     logger.warn(e.getMessage(), e);
  63.                 }
  64.             }
  65.         }
  66.     }
  67. }

上面方法通过检测消息头中的魔数是否与规定的魔数相等，提前拦截掉非常规数据包，比如通过 telnet 命令行发出的数据包。接着再对消息体长度，以及可读字节数进行检测。最后调用 decodeBody 方法进行后续的解码工作，ExchangeCodec 中实现了 decodeBody 方法，但因其子类 DubboCodec 覆写了该方法，所以在运行时 DubboCodec 中的 decodeBody 方法会被调用。下面我们来看一下该方法的代码。

  1. public class DubboCodec extends ExchangeCodec implements Codec2 {
  3.     @Override
  4.     protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException {
  5.         // 获取消息头中的第三个字节，并通过逻辑与运算得到序列化器编号
  6.         byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK);
  7.         Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto);
  8.         // 获取调用编号
  9.         long id = Bytes.bytes2long(header, 4);
  10.         // 通过逻辑与运算得到调用类型，0 - Response，1 - Request
  11.         if ((flag & FLAG_REQUEST) == 0) {
  12.             // 对响应结果进行解码，得到 Response 对象。这个非本节内容，后面再分析
  13.             // ...
  14.         } else {
  15.             // 创建 Request 对象
  16.             Request req = new Request(id);
  17.             req.setVersion(Version.getProtocolVersion());
  18.             // 通过逻辑与运算得到通信方式，并设置到 Request 对象中
  19.             req.setTwoWay((flag & FLAG_TWOWAY) != 0);
  21.             // 通过位运算检测数据包是否为事件类型
  22.             if ((flag & FLAG_EVENT) != 0) {
  23.                 // 设置心跳事件到 Request 对象中
  24.                 req.setEvent(Request.HEARTBEAT_EVENT);
  25.             }
  26.             try {
  27.                 Object data;
  28.                 if (req.isHeartbeat()) {
  29.                     // 对心跳包进行解码，该方法已被标注为废弃
  30.                     data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is));
  31.                 } else if (req.isEvent()) {
  32.                     // 对事件数据进行解码
  33.                     data = decodeEventData(channel, deserialize(s, channel.getUrl(), is));
  34.                 } else {
  35.                     DecodeableRpcInvocation inv;
  36.                     // 根据 url 参数判断是否在 IO 线程上对消息体进行解码
  37.                     if (channel.getUrl().getParameter(
  38.                             Constants.DECODE_IN_IO_THREAD_KEY,
  39.                             Constants.DEFAULT_DECODE_IN_IO_THREAD)) {
  40.                         inv = new DecodeableRpcInvocation(channel, req, is, proto);
  41.                         // 在当前线程，也就是 IO 线程上进行后续的解码工作。此工作完成后，可将
  42.                         // 调用方法名、attachment、以及调用参数解析出来
  43.                         inv.decode();
  44.                     } else {
  45.                         // 仅创建 DecodeableRpcInvocation 对象，但不在当前线程上执行解码逻辑
  46.                         inv = new DecodeableRpcInvocation(channel, req,
  47.                                 new UnsafeByteArrayInputStream(readMessageData(is)), proto);
  48.                     }
  49.                     data = inv;
  50.                 }
  52.                 // 设置 data 到 Request 对象中
  53.                 req.setData(data);
  54.             } catch (Throwable t) {
  55.                 // 若解码过程中出现异常，则将 broken 字段设为 true，
  56.                 // 并将异常对象设置到 Reqeust 对象中
  57.                 req.setBroken(true);
  58.                 req.setData(t);
  59.             }
  60.             return req;
  61.         }
  62.     }
  63. }

如上，decodeBody 对部分字段进行了解码，并将解码得到的字段封装到 Request 中。随后会调用 DecodeableRpcInvocation 的 decode 方法进行后续的解码工作。此工作完成后，可将调用方法名、attachment、以及调用参数解析出来。下面我们来看一下 DecodeableRpcInvocation 的 decode 方法逻辑。

  1. public class DecodeableRpcInvocation extends RpcInvocation implements Codec, Decodeable {
  3.     @Override
  4.     public Object decode(Channel channel, InputStream input) throws IOException {
  5.         ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType)
  6.                 .deserialize(channel.getUrl(), input);
  8.         // 通过反序列化得到 dubbo version，并保存到 attachments 变量中
  9.         String dubboVersion = in.readUTF();
  10.         request.setVersion(dubboVersion);
  11.         setAttachment(Constants.DUBBO_VERSION_KEY, dubboVersion);
  13.         // 通过反序列化得到 path，version，并保存到 attachments 变量中
  14.         setAttachment(Constants.PATH_KEY, in.readUTF());
  15.         setAttachment(Constants.VERSION_KEY, in.readUTF());
  17.         // 通过反序列化得到调用方法名
  18.         setMethodName(in.readUTF());
  19.         try {
  20.             Object[] args;
  21.             Class<?>[] pts;
  22.             // 通过反序列化得到参数类型字符串，比如 Ljava/lang/String;
  23.             String desc = in.readUTF();
  24.             if (desc.length() == 0) {
  25.                 pts = DubboCodec.EMPTY_CLASS_ARRAY;
  26.                 args = DubboCodec.EMPTY_OBJECT_ARRAY;
  27.             } else {
  28.                 // 将 desc 解析为参数类型数组
  29.                 pts = ReflectUtils.desc2classArray(desc);
  30.                 args = new Object[pts.length];
  31.                 for (int i = 0; i < args.length; i++) {
  32.                     try {
  33.                         // 解析运行时参数
  34.                         args[i] = in.readObject(pts[i]);
  35.                     } catch (Exception e) {
  36.                         if (log.isWarnEnabled()) {
  37.                             log.warn("Decode argument failed: " + e.getMessage(), e);
  38.                         }
  39.                     }
  40.                 }
  41.             }
  43.             // 设置参数类型数组
  44.             setParameterTypes(pts);
  46.             // 通过反序列化得到原 attachment 的内容
  47.             Map<String, String> map = (Map<String, String>) in.readObject(Map.class);
  48.             if (map != null && map.size() > 0) {
  49.                 Map<String, String> attachment = getAttachments();
  50.                 if (attachment == null) {
  51.                     attachment = new HashMap<String, String>();
  52.                 }
  53.                 // 将 map 与当前对象中的 attachment 集合进行融合
  54.                 attachment.putAll(map);
  55.                 setAttachments(attachment);
  56.             }
  58.             // 对 callback 类型的参数进行处理
  59.             for (int i = 0; i < args.length; i++) {
  60.                 args[i] = decodeInvocationArgument(channel, this, pts, i, args[i]);
  61.             }
  63.             // 设置参数列表
  64.             setArguments(args);
  66.         } catch (ClassNotFoundException e) {
  67.             throw new IOException(StringUtils.toString("Read invocation data failed.", e));
  68.         } finally {
  69.             if (in instanceof Cleanable) {
  70.                 ((Cleanable) in).cleanup();
  71.             }
  72.         }
  73.         return this;
  74.     }
  75. }

上面的方法通过反序列化将诸如 path、version、调用方法名、参数列表等信息依次解析出来，并设置到相应的字段中，最终得到一个具有完整调用信息的 DecodeableRpcInvocation 对象。

到这里，请求数据解码的过程就分析完了。此时我们得到了一个 Request 对象，这个对象会被传送到下一个入站处理器中，我们继续往下看。

2.3.2 调用服务

解码器将数据包解析成 Request 对象后，NettyHandler 的 messageReceived 方法紧接着会收到这个对象，并将这个对象继续向下传递。这期间该对象会被依次传递给 NettyServer、MultiMessageHandler、HeartbeatHandler 以及 AllChannelHandler。最后由 AllChannelHandler 将该对象封装到 Runnable 实现类对象中，并将 Runnable 放入线程池中执行后续的调用逻辑。整个调用栈如下：

  1. NettyHandler#messageReceived(ChannelHandlerContext, MessageEvent)
  2.   —> AbstractPeer#received(Channel, Object)
  3.     —> MultiMessageHandler#received(Channel, Object)
  4.       —> HeartbeatHandler#received(Channel, Object)
  5.         —> AllChannelHandler#received(Channel, Object)
  6.           —> ExecutorService#execute(Runnable)    // 由线程池执行后续的调用逻辑

考虑到篇幅，以及很多中间调用的逻辑并非十分重要，所以这里就不对调用栈中的每个方法都进行分析了。这里我们直接分析调用栈中的分析第一个和最后一个调用方法逻辑。如下：

  1. @Sharable
  2. public class NettyHandler extends SimpleChannelHandler {
  4.     private final Map<String, Channel> channels = new ConcurrentHashMap<String, Channel>();
  6.     private final URL url;
  8.     private final ChannelHandler handler;
  10.     public NettyHandler(URL url, ChannelHandler handler) {
  11.         if (url == null) {
  12.             throw new IllegalArgumentException("url == null");
  13.         }
  14.         if (handler == null) {
  15.             throw new IllegalArgumentException("handler == null");
  16.         }
  17.         this.url = url;
  19.         // 这里的 handler 类型为 NettyServer
  20.         this.handler = handler;
  21.     }
  23.     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
  24.         // 获取 NettyChannel
  25.         NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
  26.         try {
  27.             // 继续向下调用
  28.             handler.received(channel, e.getMessage());
  29.         } finally {
  30.             NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
  31.         }
  32.     }
  33. }

如上，NettyHandler 中的 messageReceived 逻辑比较简单。首先根据一些信息获取 NettyChannel 实例，然后将 NettyChannel 实例以及 Request 对象向下传递。下面再来看看 AllChannelHandler 的逻辑，在详细分析代码之前，我们先来了解一下 Dubbo 中的线程派发模型。

2.3.2.1 线程派发模型

Dubbo 将底层通信框架中接收请求的线程称为 IO 线程。如果一些事件处理逻辑可以很快执行完，比如只在内存打一个标记，此时直接在 IO 线程上执行该段逻辑即可。但如果事件的处理逻辑比较耗时，比如该段逻辑会发起数据库查询或者 HTTP 请求。此时我们就不应该让事件处理逻辑在 IO 线程上执行，而是应该派发到线程池中去执行。原因也很简单，IO 线程主要用于接收请求，如果 IO 线程被占满，将导致它不能接收新的请求。

以上就是线程派发的背景，下面我们再来通过 Dubbo 调用图，看一下线程派发器所处的位置。

2.3 服务提供方接收请求 - 图1

如上图，红框中的 Dispatcher 就是线程派发器。需要说明的是，Dispatcher 真实的职责创建具有线程派发能力的 ChannelHandler，比如 AllChannelHandler、MessageOnlyChannelHandler 和 ExecutionChannelHandler 等，其本身并不具备线程派发能力。Dubbo 支持 5 种不同的线程派发策略，下面通过一个表格列举一下。

策略用途
all所有消息都派发到线程池，包括请求，响应，连接事件，断开事件等
direct所有消息都不派发到线程池，全部在 IO 线程上直接执行
message只有请求和响应消息派发到线程池，其它消息均在 IO 线程上执行
execution只有请求消息派发到线程池，不含响应。其它消息均在 IO 线程上执行
connection在 IO 线程上，将连接断开事件放入队列，有序逐个执行，其它消息派发到线程池

默认配置下，Dubbo 使用 all 派发策略，即将所有的消息都派发到线程池中。下面我们来分析一下 AllChannelHandler 的代码。

  1. public class AllChannelHandler extends WrappedChannelHandler {
  3.     public AllChannelHandler(ChannelHandler handler, URL url) {
  4.         super(handler, url);
  5.     }
  7.     /** 处理连接事件 */
  8.     @Override
  9.     public void connected(Channel channel) throws RemotingException {
  10.         // 获取线程池
  11.         ExecutorService cexecutor = getExecutorService();
  12.         try {
  13.             // 将连接事件派发到线程池中处理
  14.             cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
  15.         } catch (Throwable t) {
  16.             throw new ExecutionException(..., " error when process connected event .", t);
  17.         }
  18.     }
  20.     /** 处理断开事件 */
  21.     @Override
  22.     public void disconnected(Channel channel) throws RemotingException {
  23.         ExecutorService cexecutor = getExecutorService();
  24.         try {
  25.             cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.DISCONNECTED));
  26.         } catch (Throwable t) {
  27.             throw new ExecutionException(..., "error when process disconnected event .", t);
  28.         }
  29.     }
  31.     /** 处理请求和响应消息，这里的 message 变量类型可能是 Request，也可能是 Response */
  32.     @Override
  33.     public void received(Channel channel, Object message) throws RemotingException {
  34.         ExecutorService cexecutor = getExecutorService();
  35.         try {
  36.             // 将请求和响应消息派发到线程池中处理
  37.             cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
  38.         } catch (Throwable t) {
  39.             if(message instanceof Request && t instanceof RejectedExecutionException){
  40.                 Request request = (Request)message;
  41.                 // 如果通信方式为双向通信，此时将 Server side ... threadpool is exhausted 
  42.                 // 错误信息封装到 Response 中，并返回给服务消费方。
  43.                 if(request.isTwoWay()){
  44.                     String msg = "Server side(" + url.getIp() + "," + url.getPort() 
  45.                         + ") threadpool is exhausted ,detail msg:" + t.getMessage();
  46.                     Response response = new Response(request.getId(), request.getVersion());
  47.                     response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
  48.                     response.setErrorMessage(msg);
  49.                     // 返回包含错误信息的 Response 对象
  50.                     channel.send(response);
  51.                     return;
  52.                 }
  53.             }
  54.             throw new ExecutionException(..., " error when process received event .", t);
  55.         }
  56.     }
  58.     /** 处理异常信息 */
  59.     @Override
  60.     public void caught(Channel channel, Throwable exception) throws RemotingException {
  61.         ExecutorService cexecutor = getExecutorService();
  62.         try {
  63.             cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CAUGHT, exception));
  64.         } catch (Throwable t) {
  65.             throw new ExecutionException(..., "error when process caught event ...");
  66.         }
  67.     }
  68. }

如上，请求对象会被封装 ChannelEventRunnable 中，ChannelEventRunnable 将会是服务调用过程的新起点。所以接下来我们以 ChannelEventRunnable 为起点向下探索。

2.3.2.2 调用服务

本小节，我们从 ChannelEventRunnable 开始分析，该类的主要代码如下：

  1. public class ChannelEventRunnable implements Runnable {
  3.     private final ChannelHandler handler;
  4.     private final Channel channel;
  5.     private final ChannelState state;
  6.     private final Throwable exception;
  7.     private final Object message;
  9.     @Override
  10.     public void run() {
  11.         // 检测通道状态，对于请求或响应消息，此时 state = RECEIVED
  12.         if (state == ChannelState.RECEIVED) {
  13.             try {
  14.                 // 将 channel 和 message 传给 ChannelHandler 对象，进行后续的调用
  15.                 handler.received(channel, message);
  16.             } catch (Exception e) {
  17.                 logger.warn("... operation error, channel is ... message is ...");
  18.             }
  19.         } 
  21.         // 其他消息类型通过 switch 进行处理
  22.         else {
  23.             switch (state) {
  24.             case CONNECTED:
  25.                 try {
  26.                     handler.connected(channel);
  27.                 } catch (Exception e) {
  28.                     logger.warn("... operation error, channel is ...");
  29.                 }
  30.                 break;
  31.             case DISCONNECTED:
  32.                 // ...
  33.             case SENT:
  34.                 // ...
  35.             case CAUGHT:
  36.                 // ...
  37.             default:
  38.                 logger.warn("unknown state: " + state + ", message is " + message);
  39.             }
  40.         }
  42.     }
  43. }

如上，请求和响应消息出现频率明显比其他类型消息高，所以这里对该类型的消息进行了针对性判断。ChannelEventRunnable 仅是一个中转站，它的 run 方法中并不包含具体的调用逻辑，仅用于将参数传给其他 ChannelHandler 对象进行处理，该对象类型为 DecodeHandler。

  1. public class DecodeHandler extends AbstractChannelHandlerDelegate {
  3.     public DecodeHandler(ChannelHandler handler) {
  4.         super(handler);
  5.     }
  7.     @Override
  8.     public void received(Channel channel, Object message) throws RemotingException {
  9.         if (message instanceof Decodeable) {
  10.             // 对 Decodeable 接口实现类对象进行解码
  11.             decode(message);
  12.         }
  14.         if (message instanceof Request) {
  15.             // 对 Request 的 data 字段进行解码
  16.             decode(((Request) message).getData());
  17.         }
  19.         if (message instanceof Response) {
  20.             // 对 Request 的 result 字段进行解码
  21.             decode(((Response) message).getResult());
  22.         }
  24.         // 执行后续逻辑
  25.         handler.received(channel, message);
  26.     }
  28.     private void decode(Object message) {
  29.         // Decodeable 接口目前有两个实现类，
  30.         // 分别为 DecodeableRpcInvocation 和 DecodeableRpcResult
  31.         if (message != null && message instanceof Decodeable) {
  32.             try {
  33.                 // 执行解码逻辑
  34.                 ((Decodeable) message).decode();
  35.             } catch (Throwable e) {
  36.                 if (log.isWarnEnabled()) {
  37.                     log.warn("Call Decodeable.decode failed: " + e.getMessage(), e);
  38.                 }
  39.             }
  40.         }
  41.     }
  42. }

DecodeHandler 主要是包含了一些解码逻辑。2.2.1 节分析请求解码时说过，请求解码可在 IO 线程上执行，也可在线程池中执行，这个取决于运行时配置。DecodeHandler 存在的意义就是保证请求或响应对象可在线程池中被解码。解码完毕后，完全解码后的 Request 对象会继续向后传递，下一站是 HeaderExchangeHandler。

  1. public class HeaderExchangeHandler implements ChannelHandlerDelegate {
  3.     private final ExchangeHandler handler;
  5.     public HeaderExchangeHandler(ExchangeHandler handler) {
  6.         if (handler == null) {
  7.             throw new IllegalArgumentException("handler == null");
  8.         }
  9.         this.handler = handler;
  10.     }
  12.     @Override
  13.     public void received(Channel channel, Object message) throws RemotingException {
  14.         channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
  15.         ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
  16.         try {
  17.             // 处理请求对象
  18.             if (message instanceof Request) {
  19.                 Request request = (Request) message;
  20.                 if (request.isEvent()) {
  21.                     // 处理事件
  22.                     handlerEvent(channel, request);
  23.                 } 
  24.                 // 处理普通的请求
  25.                 else {
  26.                     // 双向通信
  27.                     if (request.isTwoWay()) {
  28.                         // 向后调用服务，并得到调用结果
  29.                         Response response = handleRequest(exchangeChannel, request);
  30.                         // 将调用结果返回给服务消费端
  31.                         channel.send(response);
  32.                     } 
  33.                     // 如果是单向通信，仅向后调用指定服务即可，无需返回调用结果
  34.                     else {
  35.                         handler.received(exchangeChannel, request.getData());
  36.                     }
  37.                 }
  38.             }      
  39.             // 处理响应对象，服务消费方会执行此处逻辑，后面分析
  40.             else if (message instanceof Response) {
  41.                 handleResponse(channel, (Response) message);
  42.             } else if (message instanceof String) {
  43.                 // telnet 相关，忽略
  44.             } else {
  45.                 handler.received(exchangeChannel, message);
  46.             }
  47.         } finally {
  48.             HeaderExchangeChannel.removeChannelIfDisconnected(channel);
  49.         }
  50.     }
  52.     Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
  53.         Response res = new Response(req.getId(), req.getVersion());
  54.         // 检测请求是否合法，不合法则返回状态码为 BAD_REQUEST 的响应
  55.         if (req.isBroken()) {
  56.             Object data = req.getData();
  58.             String msg;
  59.             if (data == null)
  60.                 msg = null;
  61.             else if
  62.                 (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
  63.             else
  64.                 msg = data.toString();
  65.             res.setErrorMessage("Fail to decode request due to: " + msg);
  66.             // 设置 BAD_REQUEST 状态
  67.             res.setStatus(Response.BAD_REQUEST);
  69.             return res;
  70.         }
  72.         // 获取 data 字段值，也就是 RpcInvocation 对象
  73.         Object msg = req.getData();
  74.         try {
  75.             // 继续向下调用
  76.             Object result = handler.reply(channel, msg);
  77.             // 设置 OK 状态码
  78.             res.setStatus(Response.OK);
  79.             // 设置调用结果
  80.             res.setResult(result);
  81.         } catch (Throwable e) {
  82.             // 若调用过程出现异常，则设置 SERVICE_ERROR，表示服务端异常
  83.             res.setStatus(Response.SERVICE_ERROR);
  84.             res.setErrorMessage(StringUtils.toString(e));
  85.         }
  86.         return res;
  87.     }
  88. }

到这里，我们看到了比较清晰的请求和响应逻辑。对于双向通信，HeaderExchangeHandler 首先向后进行调用，得到调用结果。然后将调用结果封装到 Response 对象中，最后再将该对象返回给服务消费方。如果请求不合法，或者调用失败，则将错误信息封装到 Response 对象中，并返回给服务消费方。接下来我们继续向后分析，把剩余的调用过程分析完。下面分析定义在 DubboProtocol 类中的匿名类对象逻辑，如下：

  1. public class DubboProtocol extends AbstractProtocol {
  3.     public static final String NAME = "dubbo";
  5.     private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
  7.         @Override
  8.         public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
  9.             if (message instanceof Invocation) {
  10.                 Invocation inv = (Invocation) message;
  11.                 // 获取 Invoker 实例
  12.                 Invoker<?> invoker = getInvoker(channel, inv);
  13.                 if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
  14.                     // 回调相关，忽略
  15.                 }
  16.                 RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
  17.                 // 通过 Invoker 调用具体的服务
  18.                 return invoker.invoke(inv);
  19.             }
  20.             throw new RemotingException(channel, "Unsupported request: ...");
  21.         }
  23.         // 忽略其他方法
  24.     }
  26.     Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
  27.         // 忽略回调和本地存根相关逻辑
  28.         // ...
  30.         int port = channel.getLocalAddress().getPort();
  32.         // 计算 service key，格式为 groupName/serviceName:serviceVersion:port。比如：
  33.         //   dubbo/com.alibaba.dubbo.demo.DemoService:1.0.0:20880
  34.         String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));
  36.         // 从 exporterMap 查找与 serviceKey 相对应的 DubboExporter 对象，
  37.         // 服务导出过程中会将 <serviceKey, DubboExporter> 映射关系存储到 exporterMap 集合中
  38.         DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
  40.         if (exporter == null)
  41.             throw new RemotingException(channel, "Not found exported service ...");
  43.         // 获取 Invoker 对象，并返回
  44.         return exporter.getInvoker();
  45.     }
  47.     // 忽略其他方法
  48. }

以上逻辑用于获取与指定服务对应的 Invoker 实例，并通过 Invoker 的 invoke 方法调用服务逻辑。invoke 方法定义在 AbstractProxyInvoker 中，代码如下。

  1. public abstract class AbstractProxyInvoker<T> implements Invoker<T> {
  3.     @Override
  4.     public Result invoke(Invocation invocation) throws RpcException {
  5.         try {
  6.             // 调用 doInvoke 执行后续的调用，并将调用结果封装到 RpcResult 中，并
  7.             return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
  8.         } catch (InvocationTargetException e) {
  9.             return new RpcResult(e.getTargetException());
  10.         } catch (Throwable e) {
  11.             throw new RpcException("Failed to invoke remote proxy method ...");
  12.         }
  13.     }
  15.     protected abstract Object doInvoke(T proxy, String methodName, Class<?>[] parameterTypes, Object[] arguments) throws Throwable;
  16. }

如上，doInvoke 是一个抽象方法，这个需要由具体的 Invoker 实例实现。Invoker 实例是在运行时通过 JavassistProxyFactory 创建的，创建逻辑如下：

  1. public class JavassistProxyFactory extends AbstractProxyFactory {
  3.     // 省略其他方法
  5.     @Override
  6.     public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
  7.         final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
  8.         // 创建匿名类对象
  9.         return new AbstractProxyInvoker<T>(proxy, type, url) {
  10.             @Override
  11.             protected Object doInvoke(T proxy, String methodName,
  12.                                       Class<?>[] parameterTypes,
  13.                                       Object[] arguments) throws Throwable {
  14.                 // 调用 invokeMethod 方法进行后续的调用
  15.                 return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
  16.             }
  17.         };
  18.     }
  19. }

Wrapper 是一个抽象类，其中 invokeMethod 是一个抽象方法。Dubbo 会在运行时通过 Javassist 框架为 Wrapper 生成实现类，并实现 invokeMethod 方法，该方法最终会根据调用信息调用具体的服务。以 DemoServiceImpl 为例，Javassist 为其生成的代理类如下。

  1. /** Wrapper0 是在运行时生成的，大家可使用 Arthas 进行反编译 */
  2. public class Wrapper0 extends Wrapper implements ClassGenerator.DC {
  3.     public static String[] pns;
  4.     public static Map pts;
  5.     public static String[] mns;
  6.     public static String[] dmns;
  7.     public static Class[] mts0;
  9.     // 省略其他方法
  11.     public Object invokeMethod(Object object, String string, Class[] arrclass, Object[] arrobject) throws InvocationTargetException {
  12.         DemoService demoService;
  13.         try {
  14.             // 类型转换
  15.             demoService = (DemoService)object;
  16.         }
  17.         catch (Throwable throwable) {
  18.             throw new IllegalArgumentException(throwable);
  19.         }
  20.         try {
  21.             // 根据方法名调用指定的方法
  22.             if ("sayHello".equals(string) && arrclass.length == 1) {
  23.                 return demoService.sayHello((String)arrobject[0]);
  24.             }
  25.         }
  26.         catch (Throwable throwable) {
  27.             throw new InvocationTargetException(throwable);
  28.         }
  29.         throw new NoSuchMethodException(new StringBuffer().append("Not found method \"").append(string).append("\" in class com.alibaba.dubbo.demo.DemoService.").toString());
  30.     }
  31. }

到这里，整个服务调用过程就分析完了。最后把调用过程贴出来，如下：

  1. ChannelEventRunnable#run()
  2.   —> DecodeHandler#received(Channel, Object)
  3.     —> HeaderExchangeHandler#received(Channel, Object)
  4.       —> HeaderExchangeHandler#handleRequest(ExchangeChannel, Request)
  5.         —> DubboProtocol.requestHandler#reply(ExchangeChannel, Object)
  6.           —> Filter#invoke(Invoker, Invocation)
  7.             —> AbstractProxyInvoker#invoke(Invocation)
  8.               —> Wrapper0#invokeMethod(Object, String, Class[], Object[])
  9.                 —> DemoServiceImpl#sayHello(String)