一、概述

dubbo服务的底层通信实现默认是netty，netty的通道接收到数据之后还要经过层层的包装类处理，是一个调用链，最终才会调用具体的invoker反射调用对应的接口。这个调用链是怎么形成的呢？在DubboProtocol类发布服务export时，会开启通信服务，在这个过程中构建这个调用链

二、时序图

image.png

三、响应过程源码解析

1、NettyServerHandler

  1. public class NettyServerHandler extends ChannelDuplexHandler {
  2.      public void channelActive(ChannelHandlerContext ctx) throws Exception {
  3.         NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
  4.         try {
  5.             if (channel != null) {
  6.                 channels.put(NetUtils.toAddressString((InetSocketAddress) ctx.channel().remoteAddress()), channel);
  7.             }
  8.             handler.connected(channel);
  9.         } finally {
  10.             NettyChannel.removeChannelIfDisconnected(ctx.channel());
  11.         }
  12.     }
  14.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  15.         NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
  16.         try {
  17.             //handler指向NettyServer
  18.             handler.received(channel, msg);
  19.         } finally {
  20.             NettyChannel.removeChannelIfDisconnected(ctx.channel());
  21.         }
  22.     }
  23. }

2、NettyServer

  1. public class NettyServer extends AbstractServer implements Server {
  2.     ...
  3. }
  5. public abstract class AbstractPeer implements Endpoint, ChannelHandler {
  6.       public void connected(Channel ch) throws RemotingException {
  7.         if (closed) {
  8.             return;
  9.         }
  10.         handler.connected(ch);
  11.     }
  13.     public void received(Channel ch, Object msg) throws RemotingException {
  14.         if (closed) {
  15.             return;
  16.         }
  17.         //handler指向MultiMessageHandler
  18.         handler.received(ch, msg);
  19.     }
  20. }

image.png

3、MultiMessageHandler

  1. public class MultiMessageHandler extends AbstractChannelHandlerDelegate {
  2.     public void received(Channel channel, Object message) throws RemotingException {
  3.         if (message instanceof MultiMessage) {
  4.             MultiMessage list = (MultiMessage) message;
  5.             for (Object obj : list) {
  6.                 handler.received(channel, obj);
  7.             }
  8.         } else {
  9.             //handler指向HeartbeatHandler
  10.             handler.received(channel, message);
  11.         }
  12.     }
  13. }

4、HeartbeatHandler

  1. public class HeartbeatHandler extends AbstractChannelHandlerDelegate {
  2.   public void received(Channel channel, Object message) throws RemotingException {
  3.         setReadTimestamp(channel);
  4.         //心跳检测
  5.         if (isHeartbeatRequest(message)) {
  6.             Request req = (Request) message;
  7.             if (req.isTwoWay()) {
  8.                 Response res = new Response(req.getId(), req.getVersion());
  9.                 res.setEvent(Response.HEARTBEAT_EVENT);
  10.                 channel.send(res);
  11.                 if (logger.isInfoEnabled()) {
  12.                     int heartbeat = channel.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
  13.                     if (logger.isDebugEnabled()) {
  14.                         logger.debug("Received heartbeat from remote channel " + channel.getRemoteAddress()
  15.                                 + ", cause: The channel has no data-transmission exceeds a heartbeat period"
  16.                                 + (heartbeat > 0 ? ": " + heartbeat + "ms" : ""));
  17.                     }
  18.                 }
  19.             }
  20.             return;
  21.         }
  22.         if (isHeartbeatResponse(message)) {
  23.             if (logger.isDebugEnabled()) {
  24.                 logger.debug("Receive heartbeat response in thread " + Thread.currentThread().getName());
  25.             }
  26.             return;
  27.         }
  28.         //handler指向的是AllChannelHandler
  29.         handler.received(channel, message);
  30.     }
  31. }

5、AllChannelHandler

  1. public class AllChannelHandler extends WrappedChannelHandler {
  2.     public void connected(Channel channel) throws RemotingException {
  3.         ExecutorService executor = getExecutorService();
  4.         try {
  5.             executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED));
  6.         } catch (Throwable t) {
  7.             throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t);
  8.         }
  9.     }
  11.     public void received(Channel channel, Object message) throws RemotingException {
  12.         ExecutorService executor = getExecutorService();
  13.         try {
  14.             //提交到线程池执行
  15.             executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
  16.         } catch (Throwable t) {
  17.             //TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring
  18.             //fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out
  19.             if(message instanceof Request && t instanceof RejectedExecutionException){
  20.                 Request request = (Request)message;
  21.                 if(request.isTwoWay()){
  22.                     String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") threadpool is exhausted ,detail msg:" + t.getMessage();
  23.                     Response response = new Response(request.getId(), request.getVersion());
  24.                     response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
  25.                     response.setErrorMessage(msg);
  26.                     channel.send(response);
  27.                     return;
  28.                 }
  29.             }
  30.             throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
  31.         }
  32.     }
  33. }

6、ChannelEventRunnable

  1. public class ChannelEventRunnable implements Runnable {
  2. public void run() {
  3.         if (state == ChannelState.RECEIVED) {
  4.             try {
  5.                 //handler指向的是DecodeHandler
  6.                 handler.received(channel, message);
  7.             } catch (Exception e) {
  8.                 logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
  9.                         + ", message is " + message, e);
  10.             }
  11.         } else {
  12.             switch (state) {
  13.             case CONNECTED:
  14.                 try {
  15.                     handler.connected(channel);
  16.                 } catch (Exception e) {
  17.                     logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
  18.                 }
  19.                 break;
  20.             case DISCONNECTED:
  21.                 try {
  22.                     handler.disconnected(channel);
  23.                 } catch (Exception e) {
  24.                     logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
  25.                 }
  26.                 break;
  27.             case SENT:
  28.                 try {
  29.                     handler.sent(channel, message);
  30.                 } catch (Exception e) {
  31.                     logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
  32.                             + ", message is " + message, e);
  33.                 }
  34.                 break;
  35.             case CAUGHT:
  36.                 try {
  37.                     handler.caught(channel, exception);
  38.                 } catch (Exception e) {
  39.                     logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
  40.                             + ", message is: " + message + ", exception is " + exception, e);
  41.                 }
  42.                 break;
  43.             default:
  44.                 logger.warn("unknown state: " + state + ", message is " + message);
  45.             }
  46.         }
  47.     }
  48. }

7、DecodeHandler

  1. public class DecodeHandler extends AbstractChannelHandlerDelegate {
  2.      public void received(Channel channel, Object message) throws RemotingException {
  3.         if (message instanceof Decodeable) {
  4.             decode(message);
  5.         }
  6.         if (message instanceof Request) {
  7.             decode(((Request) message).getData());
  8.         }
  9.         if (message instanceof Response) {
  10.             decode(((Response) message).getResult());
  11.         }
  12.         //handler指向HeaderExchangeHandler 
  13.         handler.received(channel, message);
  14.     }
  15. }

8、HeaderExchangeHandler

  1. public class HeaderExchangeHandler implements ChannelHandlerDelegate {
  3.      public void received(Channel channel, Object message) throws RemotingException {
  4.         channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
  5.         final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
  6.         try {
  7.             if (message instanceof Request) {
  8.                 // handle request.
  9.                 Request request = (Request) message;
  10.                 if (request.isEvent()) {
  11.                     handlerEvent(channel, request);
  12.                 } else {
  13.                     if (request.isTwoWay()) {
  14.                         //代码会走到这一步
  15.                         handleRequest(exchangeChannel, request);
  16.                     } else {
  17.                         handler.received(exchangeChannel, request.getData());
  18.                     }
  19.                 }
  20.             } else if (message instanceof Response) {
  21.                 handleResponse(channel, (Response) message);
  22.             } else if (message instanceof String) {
  23.                 if (isClientSide(channel)) {
  24.                     Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
  25.                     logger.error(e.getMessage(), e);
  26.                 } else {
  27.                     String echo = handler.telnet(channel, (String) message);
  28.                     if (echo != null && echo.length() > 0) {
  29.                         channel.send(echo);
  30.                     }
  31.                 }
  32.             } else {
  33.                 handler.received(exchangeChannel, message);
  34.             }
  35.         } finally {
  36.             HeaderExchangeChannel.removeChannelIfDisconnected(channel);
  37.         }
  38.     }
  40.     void handleRequest(final ExchangeChannel channel, Request req) throws RemotingException {
  41.         Response res = new Response(req.getId(), req.getVersion());
  42.         if (req.isBroken()) {
  43.             Object data = req.getData();
  44.             String msg;
  45.             if (data == null) {
  46.                 msg = null;
  47.             } else if (data instanceof Throwable) {
  48.                 msg = StringUtils.toString((Throwable) data);
  49.             } else {
  50.                 msg = data.toString();
  51.             }
  52.             res.setErrorMessage("Fail to decode request due to: " + msg);
  53.             res.setStatus(Response.BAD_REQUEST);
  54.             channel.send(res);
  55.             return;
  56.         }
  57.         // find handler by message class.
  58.         Object msg = req.getData();
  59.         try {
  60.             //handler指向的是ExchangeHandlerAdapter的实例，但ExchangeHandlerAdapter是抽象类，
  61.             //它的具体实现在DubboProtocol类中
  62.             CompletionStage<Object> future = handler.reply(channel, msg);
  63.             future.whenComplete((appResult, t) -> {
  64.                 try {
  65.                     if (t == null) {
  66.                         res.setStatus(Response.OK);
  67.                         res.setResult(appResult);
  68.                     } else {
  69.                         res.setStatus(Response.SERVICE_ERROR);
  70.                         res.setErrorMessage(StringUtils.toString(t));
  71.                     }
  72.                     channel.send(res);
  73.                 } catch (RemotingException e) {
  74.                     logger.warn("Send result to consumer failed, channel is " + channel + ", msg is " + e);
  75.                 } finally {
  76.                     // HeaderExchangeChannel.removeChannelIfDisconnected(channel);
  77.                 }
  78.             });
  79.         } catch (Throwable e) {
  80.             res.setStatus(Response.SERVICE_ERROR);
  81.             res.setErrorMessage(StringUtils.toString(e));
  82.             channel.send(res);
  83.         }
  84.     }
  85. }

9、ExchangeHandlerAdapter

  1. public class DubboProtocol extends AbstractProtocol {
  3.      private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
  5.         public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
  6.             if (!(message instanceof Invocation)) {
  7.                 throw new RemotingException(channel, "Unsupported request: "
  8.                         + (message == null ? null : (message.getClass().getName() + ": " + message))
  9.                         + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
  10.             }
  11.             Invocation inv = (Invocation) message;
  12.             //获取invoker实例，这个实例并不是接口的实现类，而是之前服务发布过程中被层层包装的
  13.             //Invoker实例，要经过包装类和filter责任链的层层调用才能最终调用到服务的实现类
  14.             Invoker<?> invoker = getInvoker(channel, inv);
  15.             // need to consider backward-compatibility if it's a callback
  16.             if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
  17.                 String methodsStr = invoker.getUrl().getParameters().get("methods");
  18.                 boolean hasMethod = false;
  19.                 if (methodsStr == null || !methodsStr.contains(",")) {
  20.                     hasMethod = inv.getMethodName().equals(methodsStr);
  21.                 } else {
  22.                     String[] methods = methodsStr.split(",");
  23.                     for (String method : methods) {
  24.                         if (inv.getMethodName().equals(method)) {
  25.                             hasMethod = true;
  26.                             break;
  27.                         }
  28.                     }
  29.                 }
  30.                 if (!hasMethod) {
  31.                     logger.warn(new IllegalStateException("The methodName " + inv.getMethodName()
  32.                             + " not found in callback service interface ,invoke will be ignored."
  33.                             + " please update the api interface. url is:"
  34.                             + invoker.getUrl()) + " ,invocation is :" + inv);
  35.                     return null;
  36.                 }
  37.             }
  38.             RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
  39.             Result result = invoker.invoke(inv);
  40.             return result.completionFuture().thenApply(Function.identity());
  41.         }
  43.         public void received(Channel channel, Object message) throws RemotingException {
  44.             if (message instanceof Invocation) {
  45.                 reply((ExchangeChannel) channel, message);
  47.             } else {
  48.                 super.received(channel, message);
  49.             }
  50.         }
  52.         public void connected(Channel channel) throws RemotingException {
  53.             invoke(channel, ON_CONNECT_KEY);
  54.         }
  56.         public void disconnected(Channel channel) throws RemotingException {
  57.             if (logger.isDebugEnabled()) {
  58.                 logger.debug("disconnected from " + channel.getRemoteAddress() + ",url:" + channel.getUrl());
  59.             }
  60.             invoke(channel, ON_DISCONNECT_KEY);
  61.         }
  63.         private void invoke(Channel channel, String methodKey) {
  64.             Invocation invocation = createInvocation(channel, channel.getUrl(), methodKey);
  65.             if (invocation != null) {
  66.                 try {
  67.                     received(channel, invocation);
  68.                 } catch (Throwable t) {
  69.                     logger.warn("Failed to invoke event method " + invocation.getMethodName() + "(), cause: " + t.getMessage(), t);
  70.                 }
  71.             }
  72.         }
  74.         private Invocation createInvocation(Channel channel, URL url, String methodKey) {
  75.             String method = url.getParameter(methodKey);
  76.             if (method == null || method.length() == 0) {
  77.                 return null;
  78.             }
  79.             RpcInvocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
  80.             invocation.setAttachment(PATH_KEY, url.getPath());
  81.             invocation.setAttachment(GROUP_KEY, url.getParameter(GROUP_KEY));
  82.             invocation.setAttachment(INTERFACE_KEY, url.getParameter(INTERFACE_KEY));
  83.             invocation.setAttachment(VERSION_KEY, url.getParameter(VERSION_KEY));
  84.             if (url.getParameter(STUB_EVENT_KEY, false)) {
  85.                 invocation.setAttachment(STUB_EVENT_KEY, Boolean.TRUE.toString());
  86.             }
  87.             return invocation;
  88.         }
  89.     };
  90. }

四、时序图中的调用链是如何构建的