一 总体流程
客户端接收响应消息
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 }
时间： 2024-10-08 06:29:47