9.3 客户端接收响应信息(异步转同步的实现)

Posted 2020-10-14 赵计刚

一 总体流程

客户端接收响应消息
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     }