Dubbo分析之Exchange层

系列文章

Dubbo分析Serialize层
Dubbo分析之Transport层
Dubbo分析之Exchange 层
Dubbo分析之Protocol层

前言

紧接着上文Dubbo分析之Transport层，本文继续介绍Exchange层，此层官方介绍为信息交换层：封装请求响应模式，同步转异步，以 Request, Response 为中心，扩展接口为 Exchanger, ExchangeChannel, ExchangeClient, ExchangeServer；下面分别进行介绍

Exchanger分析

Exchanger是此层的核心接口类，提供了connect()和bind()接口，分别返回ExchangeClient和ExchangeServer；dubbo提供了此接口的默认实现类HeaderExchanger，代码如下：

1public class HeaderExchanger implements Exchanger {
2 
3    public static final String NAME = "header";
4 
5    @Override
6    public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
7        return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
8    }
9 
10    @Override
11    public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
12        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
13    }
14 
15}

在实现类中在connect和bind中分别实例化了HeaderExchangeClient和HeaderExchangeServer，传入的参数是Transporters，可以认为这里就是Transport层的入口类；这里的ExchangeClient/ExchangeServer其实就是对Client/Server的包装，同时传入了自己的ChannelHandler；ChannelHandler已经在Transport层介绍过了，提供了连接建立，连接端口，发送请求，接受请求等接口；已默认使用的Netty为例，这里就是对NettyClient和NettyServer的包装，同时传入DecodeHandler，在NettyHandler中被调用；

ExchangeClient分析

ExchangeClient本身也继承于Client，同时也继承于ExchangeChannel：

1public interface ExchangeClient extends Client, ExchangeChannel {
2 
3}
4 
5public interface ExchangeChannel extends Channel {
6 
7    ResponseFuture request(Object request) throws RemotingException;
8 
9    ResponseFuture request(Object request, int timeout) throws RemotingException;
10 
11    ExchangeHandler getExchangeHandler();
12 
13    @Override
14    void close(int timeout);
15 
16}

ExchangeChannel负责将上层的data包装成Request，然后发送给Transport层；具体的逻辑在HeaderExchangeChannel中：

1public ResponseFuture request(Object request, int timeout) throws RemotingException {
2       if (closed) {
3           throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
4       }
5       // create request.
6       Request req = new Request();
7       req.setVersion(Version.getProtocolVersion());
8       req.setTwoWay(true);
9       req.setData(request);
10       DefaultFuture future = new DefaultFuture(channel, req, timeout);
11       try {
12           channel.send(req);
13       } catch (RemotingException e) {
14           future.cancel();
15           throw e;
16       }
17       return future;
18   }

创建了一个Request，在构造器中同时会产生一个RequestId；设置了协议版本，是否双向通信，最后设置了真实的业务数据；接下来实例化了一个DefaultFuture类，此类实现了同步转异步的方式，channel调用send发送请求之后，不需要等待结果，直接将DefaultFuture返回给上层，上层可以通过调用DefaultFuture的get方法来获取响应，get方法会阻塞等待获取服务器的响应才会返回；Client接收消息在handler里面，比如Netty在NettyHandler里面messageReceived方法介绍响应消息，NettyHandler最终会调用上面传入的DecodeHandler，DecodeHandler会先判断一下是否已经解码，如果解码就直接调用HeaderExchangeHandler，默认已经设置了编码解码器，所以会直接调用HeaderExchangeHandler里面的received方法：

1public void received(Channel channel, Object message) throws RemotingException {
2       channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
3       ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
4       try {
5           if (message instanceof Request) {
6               // handle request.
7               Request request = (Request) message;
8               if (request.isEvent()) {
9                   handlerEvent(channel, request);
10               } else {
11                   if (request.isTwoWay()) {
12                       Response response = handleRequest(exchangeChannel, request);
13                       channel.send(response);
14                   } else {
15                       handler.received(exchangeChannel, request.getData());
16                   }
17               }
18           } else if (message instanceof Response) {
19               handleResponse(channel, (Response) message);
20           } else if (message instanceof String) {
21               if (isClientSide(channel)) {
22                   Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
23                   logger.error(e.getMessage(), e);
24               } else {
25                   String echo = handler.telnet(channel, (String) message);
26                   if (echo != null && echo.length() > 0) {
27                       channel.send(echo);
28                   }
29               }
30           } else {
31               handler.received(exchangeChannel, message);
32           }
33       } finally {
34           HeaderExchangeChannel.removeChannelIfDisconnected(channel);
35       }
36   }

服务端和客户端都会使用此方法，这里是客户端接受的是Response，直接调用handleResponse方法：

1static void handleResponse(Channel channel, Response response) throws RemotingException {
2    if (response != null && !response.isHeartbeat()) {
3        DefaultFuture.received(channel, response);
4    }
5}

接收到响应之后，再去告诉DefaultFuture已经收到响应，DefaultFuture本身存放了requestId对应DefaultFuture的一个ConcurrentHashMap；具体怎么映射过去，Response也包含一个responseId，此responseId和requestId是相同的；

1private final Lock lock = new ReentrantLock();
2private final Condition done = lock.newCondition();
3   
4public static void received(Channel channel, Response response) {
5      try {
6          DefaultFuture future = FUTURES.remove(response.getId());
7          if (future != null) {
8              future.doReceived(response);
9          } else {
10              logger.warn("The timeout response finally returned at "
11                      + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
12                      + ", response " + response
13                      + (channel == null ? "" : ", channel: " + channel.getLocalAddress()
14                      + " -> " + channel.getRemoteAddress()));
15          }
16      } finally {
17          CHANNELS.remove(response.getId());
18      }
19  }
20   
21  private void doReceived(Response res) {
22      lock.lock();
23      try {
24          response = res;
25          if (done != null) {
26              done.signal();
27          }
28      } finally {
29          lock.unlock();
30      }
31      if (callback != null) {
32          invokeCallback(callback);
33      }
34  }

通过responseId获取了之前请求时创建的DefaultFuture，然后再更新DefaultFuture内部的response对象，更新完之后在调用Condition的signal方法，用户唤起通过DefaultFuture的get方法获取响应的阻塞线程：

1public Object get(int timeout) throws RemotingException {
2        if (timeout <= 0) {
3            timeout = Constants.DEFAULT_TIMEOUT;
4        }
5        if (!isDone()) {
6            long start = System.currentTimeMillis();
7            lock.lock();
8            try {
9                while (!isDone()) {
10                    done.await(timeout, TimeUnit.MILLISECONDS);
11                    if (isDone() || System.currentTimeMillis() - start > timeout) {
12                        break;
13                    }
14                }
15            } catch (InterruptedException e) {
16                throw new RuntimeException(e);
17            } finally {
18                lock.unlock();
19            }
20            if (!isDone()) {
21                throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
22            }
23        }
24        return returnFromResponse();
25    }

可以发现阻塞要么被获取被signal方法唤醒，要么等待超时；以上大致是客户端发送获取响应的流程，下面看看服务器端流程

ExchangeServer分析

ExchangeServer继承于Server，同时提供了两个包装服务端Channel的方法

1public interface ExchangeServer extends Server {
2 
3    Collection<ExchangeChannel> getExchangeChannels();
4 
5    ExchangeChannel getExchangeChannel(InetSocketAddress remoteAddress);
6}

服务器端主要用于接收Request消息，然后处理消息，最后把响应发送给客户端，相关接收消息已经在上面介绍过了，同样是在HeaderExchangeHandler里面的received方法中，只不过这里的消息类型为Request；

1Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
2      Response res = new Response(req.getId(), req.getVersion());
3      if (req.isBroken()) {
4          Object data = req.getData();
5 
6          String msg;
7          if (data == null) msg = null;
8          else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
9          else msg = data.toString();
10          res.setErrorMessage("Fail to decode request due to: " + msg);
11          res.setStatus(Response.BAD_REQUEST);
12 
13          return res;
14      }
15      // find handler by message class.
16      Object msg = req.getData();
17      try {
18          // handle data.
19          Object result = handler.reply(channel, msg);
20          res.setStatus(Response.OK);
21          res.setResult(result);
22      } catch (Throwable e) {
23          res.setStatus(Response.SERVICE_ERROR);
24          res.setErrorMessage(StringUtils.toString(e));
25      }
26      return res;
27  }

首先创建了一个Response，并且指定responseId为requestId，方便在客户端定位到具体的DefaultFuture；然后调用handler的reply方法处理消息，返回结果，如何处理的将在后面的protocol层介绍，大致就是通过Request的信息，反射调用Server端的服务，然后返回结果，然后将结果放入Response对象中，通过channel将消息发送客户端；

总结

本文介绍了Exchange层的大体流程，围绕Exchanger，ExchangeClient和ExchangeServer展开；请求封装成Request，响应封装成Response，客户端通过异步的方式接收服务器请求；

示例代码地址

https://github.com/ksfzhaohui...
https://gitee.com/OutOfMemory...