一 總體流程

客戶端接收響應消息
NettyHandler.messageReceived(ChannelHandlerContext ctx, MessageEvent e)
-->MultiMessageHandler.received(Channel channel, Object message)
  -->HeartbeatHandler.received(Channel channel, Object message)
    -->AllChannelHandler.received(Channel channel, Object message)
      -->ExecutorService cexecutor = getExecutorService()
      -->cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message))
        -->ChannelEventRunnable.run()
          -->DecodeHandler.received(Channel channel, Object message)
            -->decode(Object message)
            -->HeaderExchangeHandler.received(Channel channel, Object message)
              -->handleResponse(Channel channel, Response response)
                -->DefaultFuture.received(channel, response)
                  -->doReceived(Response res)//異步轉同步

二 源碼解析

在HeaderExchangeHandler.received(Channel channel, Object message)方法之前，與服務端接收請求消息一樣，不再贅述。

HeaderExchangeHandler.received(Channel channel, Object message)

 1     public void received(Channel channel, Object message) throws RemotingException {
 2         ...
 3         try {
 4             if
 
 (message instanceof Request) {
 5                 ...
 6             } else if (message instanceof Response) {
 7                 handleResponse(channel, (Response) message);
 8             } else if (message instanceof String) {
 9                 ...
10             } else {
11                 ...
 
12             }
13         } finally {
14             HeaderExchangeChannel.removeChannelIfDisconnected(channel);
15         }
16     }
17 
18     static void handleResponse(Channel channel, Response response) throws RemotingException {
19         if (response != null && !response.isHeartbeat()) {
20             DefaultFuture.received(channel, response);
21         }
22     }

DefaultFuture.received(Channel channel, Response response)

 1     private final long id;
 2     private final Request request;
 3     private final int timeout;
 4     private volatile Response response;
 5     private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
 6     private final Condition done = lock.newCondition();
 7 
 8     public static void received(Channel channel, Response response) {
 9         try {
10             DefaultFuture future = FUTURES.remove(response.getId());//刪除元素並返回key=response.getId()的DefaultFuture

11             if (future != null) {
12                 future.doReceived(response);
13             } else {
14                 logger.warn("The timeout response finally returned at "
15                         + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
16                         + ", response " + response
17                         + (channel == null ? "" : ", channel: " + channel.getLocalAddress()
18                         + " -> " + channel.getRemoteAddress()));
19             }
20         } finally {
21             CHANNELS.remove(response.getId());
22         }
23     }
24 
25     private void doReceived(Response res) {
26         lock.lock();
27         try {
28             //設置response
29             response = res;
30             if (done != null) {
31                 //喚醒阻塞的線程
32                 done.signal();
33             }
34         } finally {
35             lock.unlock();
36         }
37         if (callback != null) {
38             invokeCallback(callback);
39         }
40     }

這裏比較難懂，筆者再給出客戶端發出請求時的一段代碼：HeaderExchangeChannel.request(Object request, int timeout)

 1     public 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("2.0.0");
 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     }

netty是一個異步非阻塞的框架，所以當執行channel.send(req);的時候，當其內部執行到netty發送消息時，不會等待結果，直接返回。為了實現“異步轉為同步”，使用了DefaultFuture這個輔助類，

在HeaderExchangeChannel.request(Object request, int timeout)，在還沒有等到客戶端的響應回來的時候，就直接將future返回了。返回給誰？再來看HeaderExchangeChannel.request(Object request, int timeout)的調用者。

1                   -->DubboInvoker.doInvoke(final Invocation invocation)
2                     //獲取ExchangeClient進行消息的發送
3                     -->ReferenceCountExchangeClient.request(Object request, int timeout)
4                       -->HeaderExchangeClient.request(Object request, int timeout)
5                         -->HeaderExchangeChannel.request(Object request, int timeout)

DubboInvoker.doInvoke(final Invocation invocation)

 1 protected Result doInvoke(final Invocation invocation) throws Throwable {
 2         RpcInvocation inv = (RpcInvocation) invocation;
 3         final String methodName = RpcUtils.getMethodName(invocation);
 4         inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
 5         inv.setAttachment(Constants.VERSION_KEY, version);
 6 
 7         ExchangeClient currentClient;
 8         if (clients.length == 1) {
 9             currentClient = clients[0];
10         } else {
11             currentClient = clients[index.getAndIncrement() % clients.length];
12         }
13         try {
14             boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);//是否異步
15             boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);//是否沒有返回值
16             int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
17             if (isOneway) {
18                 boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
19                 currentClient.send(inv, isSent);
20                 RpcContext.getContext().setFuture(null);
21                 return new RpcResult();
22             } else if (isAsync) {
23                 ResponseFuture future = currentClient.request(inv, timeout);
24                 RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
25                 return new RpcResult();
26             } else {
27                 RpcContext.getContext().setFuture(null);
28                 return (Result) currentClient.request(inv, timeout).get();
29             }
30         } catch (TimeoutException e) {
31             throw new RpcException(...);
32         } catch (RemotingException e) {
33             throw new RpcException(...);
34         }
35     }

其中currentClient.request(inv, timeout)返回值是ResponseFuture，DefaultFuture是ResponseFuture的實現類，實際上這裏返回的就是DefaultFuture實例，而該實例就是HeaderExchangeChannel.request(Object request, int timeout)返回的那個future實例。之後調用DefaultFuture.get()。

 1     public Object get() throws RemotingException {
 2         return get(timeout);
 3     }
 4 
 5     public Object get(int timeout) throws RemotingException {
 6         if (timeout <= 0) {
 7             timeout = Constants.DEFAULT_TIMEOUT;
 8         }
 9         if (!isDone()) {
10             long start = System.currentTimeMillis();
11             lock.lock();
12             try {
13                 while (!isDone()) {
14                     //Causes the current thread to wait until it is signalled or interrupted, or the specified waiting time elapses.
15                     done.await(timeout, TimeUnit.MILLISECONDS);
16                     if (isDone() || System.currentTimeMillis() - start > timeout) {
17                         break;
18                     }
19                 }
20             } catch (InterruptedException e) {
21                 throw new RuntimeException(e);
22             } finally {
23                 lock.unlock();
24             }
25             if (!isDone()) {
26                 throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
27             }
28         }
29         return returnFromResponse();
30     }
31 
32     public boolean isDone() {
33         return response != null;
34     }

此處我們看到當響應response沒有回來時，condition會執行await進行阻塞當前線程，直到被喚醒或被中斷或阻塞時間到時了。當客戶端接收到服務端的響應的時候，DefaultFuture.doReceived：

會先為response賦上返回值，之後執行condition的signal喚醒被阻塞的線程，get()方法就會釋放鎖，執行returnFromResponse()，返回值。

 1     private Object returnFromResponse() throws RemotingException {
 2         Response res = response;
 3         if (res == null) {
 4             throw new IllegalStateException("response cannot be null");
 5         }
 6         if (res.getStatus() == Response.OK) {
 7             return res.getResult();
 8         }
 9         if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
10             throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
11         }
12         throw new RemotingException(channel, res.getErrorMessage());
13     }

到現在其實還有一個問題？就是netty時異步非阻塞的，那麽假設現在我發了1w個Request，後來返回來1w個Response，那麽怎麽對應Request和Response呢？如果對應不上，最起碼的喚醒就會有問題。為了解決這個問題提，Request和Response中都有一個屬性id。

在HeaderExchangeChannel.request(Object request, int timeout)中：

 1         Request req = new Request();
 2         req.setVersion("2.0.0");
 3         req.setTwoWay(true);
 4         req.setData(request);
 5         DefaultFuture future = new DefaultFuture(channel, req, timeout);
 6         try {
 7             channel.send(req);
 8         } catch (RemotingException e) {
 9             future.cancel();
10             throw e;
11         }
12         return future;

看一下Request的構造器：

 1     private static final AtomicLong INVOKE_ID = new AtomicLong(0);
 2     private final long mId;
 3 
 4     public Request() {
 5         mId = newId();
 6     }
 7 
 8     private static long newId() {
 9         // getAndIncrement()增長到MAX_VALUE時，再增長會變為MIN_VALUE，負數也可以做為ID
10         return INVOKE_ID.getAndIncrement();
11     }

看一下DefaultFuture的構造器：

 1     private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
 2     private final long id;
 3     private final Request request;
 4     private volatile Response response;
 5 
 6     public DefaultFuture(Channel channel, Request request, int timeout) {
 7         ...
 8         this.request = request;
 9         this.id = request.getId();
10         ...
11         FUTURES.put(id, this);
12         ...
13     }

再來看一下響應。

HeaderExchangeHandler.handleRequest(ExchangeChannel channel, Request req)

 1     Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
 2         Response res = new Response(req.getId(), req.getVersion());
 3         ...
 4         Object msg = req.getData();
 5         try {
 6             // handle data.
 7             Object result = handler.reply(channel, msg);
 8             res.setStatus(Response.OK);
 9             res.setResult(result);
10         } catch (Throwable e) {
11             res.setStatus(Response.SERVICE_ERROR);
12             res.setErrorMessage(StringUtils.toString(e));
13         }
14         return res;
15     }

來看一下Response的構造器：

1     private long mId = 0;
2 
3     public Response(long id, String version) {
4         mId = id;
5         mVersion = version;
6     }

這裏response的id的值時request的id。最後來看一下服務端接收後的處理：

DefaultFuture.received(Channel channel, Response response)

 1     public static void received(Channel channel, Response response) {
 2         try {
 3             DefaultFuture future = FUTURES.remove(response.getId());//刪除元素並返回key=response.getId()的DefaultFuture
 4             if (future != null) {
 5                 future.doReceived(response);
 6             } else {
 7                ...
 8             }
 9         } finally {
10             CHANNELS.remove(response.getId());
11         }
12     }

9.3 客戶端接收響應信息(異步轉同步的實現)