9.3 客戶端接收響應信息(異步轉同步的實現)
一 總體流程
客戶端接收響應消息 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 客戶端接收響應信息(異步轉同步的實現)