4 提供方返回调用结果
服务提供方调用指定服务后，会将调用结果封装到 Response 对象中，并将该对象返回给服务消费方。服务提供方也是通过 NettyChannel 的 send 方法将 Response 对象返回，这里就不在重复分析了。本节我们仅需关注 Response 对象的编码过程即可
public class ExchangeCodec extends TelnetCodec { public void encode(Channel channel, ChannelBuffer buffer, Object msg) throws IOException { if (msg instanceof Request) { encodeRequest(channel, buffer, (Request) msg); } else if (msg instanceof Response) { // 对响应对象进行编码 encodeResponse(channel, buffer, (Response) msg); } else { super.encode(channel, buffer, msg); } } protected void encodeResponse(Channel channel, ChannelBuffer buffer, Response res) throws IOException { int savedWriteIndex = buffer.writerIndex(); try { Serialization serialization = getSerialization(channel); // 创建消息头字节数组 byte[] header = new byte[HEADER_LENGTH]; // 设置魔数 Bytes.short2bytes(MAGIC, header); // 设置序列化器编号 header[2] = serialization.getContentTypeId(); if (res.isHeartbeat()) header[2] |= FLAG_EVENT; // 获取响应状态 byte status = res.getStatus(); // 设置响应状态 header[3] = status; // 设置请求编号 Bytes.long2bytes(res.getId(), header, 4); // 更新 writerIndex，为消息头预留 16 个字节的空间 buffer.writerIndex(savedWriteIndex + HEADER_LENGTH); ChannelBufferOutputStream bos = new ChannelBufferOutputStream(buffer); ObjectOutput out = serialization.serialize(channel.getUrl(), bos); if (status == Response.OK) { if (res.isHeartbeat()) { // 对心跳响应结果进行序列化，已废弃 encodeHeartbeatData(channel, out, res.getResult()); } else { // 对调用结果进行序列化 encodeResponseData(channel, out, res.getResult(), res.getVersion()); } } else { // 对错误信息进行序列化 out.writeUTF(res.getErrorMessage()) }; out.flushBuffer(); if (out instanceof Cleanable) { ((Cleanable) out).cleanup(); } bos.flush(); bos.close(); // 获取写入的字节数，也就是消息体长度 int len = bos.writtenBytes(); checkPayload(channel, len); // 将消息体长度写入到消息头中 Bytes.int2bytes(len, header, 12); // 将 buffer 指针移动到 savedWriteIndex，为写消息头做准备 buffer.writerIndex(savedWriteIndex); // 从 savedWriteIndex 下标处写入消息头 buffer.writeBytes(header); // 设置新的 writerIndex，writerIndex = 原写下标 + 消息头长度 + 消息体长度 buffer.writerIndex(savedWriteIndex + HEADER_LENGTH + len); } catch (Throwable t) { // 异常处理逻辑不是很难理解，但是代码略多，这里忽略了 } } } public class DubboCodec extends ExchangeCodec implements Codec2 { protected void encodeResponseData(Channel channel, ObjectOutput out, Object data, String version) throws IOException { Result result = (Result) data; // 检测当前协议版本是否支持带有 attachment 集合的 Response 对象 boolean attach = Version.isSupportResponseAttachment(version); Throwable th = result.getException(); // 异常信息为空 if (th == null) { Object ret = result.getValue(); // 调用结果为空 if (ret == null) { // 序列化响应类型 out.writeByte(attach ? RESPONSE_NULL_VALUE_WITH_ATTACHMENTS : RESPONSE_NULL_VALUE); } // 调用结果非空 else { // 序列化响应类型 out.writeByte(attach ? RESPONSE_VALUE_WITH_ATTACHMENTS : RESPONSE_VALUE); // 序列化调用结果 out.writeObject(ret); } } // 异常信息非空 else { // 序列化响应类型 out.writeByte(attach ? RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS : RESPONSE_WITH_EXCEPTION); // 序列化异常对象 out.writeObject(th); } if (attach) { // 记录 Dubbo 协议版本 result.getAttachments().put(Constants.DUBBO_VERSION_KEY, Version.getProtocolVersion()); // 序列化 attachments 集合 out.writeObject(result.getAttachments()); } } }
以上就是 Response 对象编码的过程，和前面分析的 Request 对象编码过程很相似。如果大家能看 Request 对象的编码逻辑，那么这里的 Response 对象的编码逻辑也不难理解，就不多说了。接下来我们再来分析双向通信的最后一环 —— 服务消费方接收调用结果。
5 消费方接收调用结果
服务消费方在收到响应数据后，首先要做的事情是对响应数据进行解码，得到 Response 对象。然后再将该对象传递给下一个入站处理器，这个入站处理器就是 NettyHandler。接下来 NettyHandler 会将这个对象继续向下传递，最后 AllChannelHandler 的 received 方法会收到这个对象，并将这个对象派发到线程池中。这个过程和服务提供方接收请求的过程是一样的，因此这里就不重复分析了
（1）响应数据解码
响应数据解码逻辑主要的逻辑封装在 DubboCodec 中，我们直接分析这个类的代码。如下：
public class DubboCodec extends ExchangeCodec implements Codec2 { @Override protected Object decodeBody(Channel channel, InputStream is, byte[] header) throws IOException { byte flag = header[2], proto = (byte) (flag & SERIALIZATION_MASK); Serialization s = CodecSupport.getSerialization(channel.getUrl(), proto); // 获取请求编号 long id = Bytes.bytes2long(header, 4); // 检测消息类型，若下面的条件成立，表明消息类型为 Response if ((flag & FLAG_REQUEST) == 0) { // 创建 Response 对象 Response res = new Response(id); // 检测事件标志位 if ((flag & FLAG_EVENT) != 0) { // 设置心跳事件 res.setEvent(Response.HEARTBEAT_EVENT); } // 获取响应状态 byte status = header[3]; // 设置响应状态 res.setStatus(status); // 如果响应状态为 OK，表明调用过程正常 if (status == Response.OK) { try { Object data; if (res.isHeartbeat()) { // 反序列化心跳数据，已废弃 data = decodeHeartbeatData(channel, deserialize(s, channel.getUrl(), is)); } else if (res.isEvent()) { // 反序列化事件数据 data = decodeEventData(channel, deserialize(s, channel.getUrl(), is)); } else { DecodeableRpcResult result; // 根据 url 参数决定是否在 IO 线程上执行解码逻辑 if (channel.getUrl().getParameter( Constants.DECODE_IN_IO_THREAD_KEY, Constants.DEFAULT_DECODE_IN_IO_THREAD)) { // 创建 DecodeableRpcResult 对象 result = new DecodeableRpcResult(channel, res, is, (Invocation) getRequestData(id), proto); // 进行后续的解码工作 result.decode(); } else { // 创建 DecodeableRpcResult 对象 result = new DecodeableRpcResult(channel, res, new UnsafeByteArrayInputStream(readMessageData(is)), (Invocation) getRequestData(id), proto); } data = result; } // 设置 DecodeableRpcResult 对象到 Response 对象中 res.setResult(data); } catch (Throwable t) { // 解码过程中出现了错误，此时设置 CLIENT_ERROR 状态码到 Response 对象中 res.setStatus(Response.CLIENT_ERROR); res.setErrorMessage(StringUtils.toString(t)); } } // 响应状态非 OK，表明调用过程出现了异常 else { // 反序列化异常信息，并设置到 Response 对象中 res.setErrorMessage(deserialize(s, channel.getUrl(), is).readUTF()); } return res; } else { // 对请求数据进行解码，前面已分析过，此处忽略 } } }
以上就是响应数据的解码过程，上面逻辑看起来是不是似曾相识。对的，我们在前面章节分析过 DubboCodec 的 decodeBody 方法中关于请求数据的解码过程，该过程和响应数据的解码过程很相似。下面，我们继续分析调用结果的反序列化过程
public class DecodeableRpcResult extends AppResponse implements Codec, Decodeable { private static final Logger log = LoggerFactory.getLogger(DecodeableRpcResult.class); private Channel channel; private byte serializationType; private InputStream inputStream; private Response response; private Invocation invocation; private volatile boolean hasDecoded; public DecodeableRpcResult(Channel channel, Response response, InputStream is, Invocation invocation, byte id) { Assert.notNull(channel, "channel == null"); Assert.notNull(response, "response == null"); Assert.notNull(is, "inputStream == null"); this.channel = channel; this.response = response; this.inputStream = is; this.invocation = invocation; this.serializationType = id; } @Override public void encode(Channel channel, OutputStream output, Object message) throws IOException { throw new UnsupportedOperationException(); } @Override public Object decode(Channel channel, InputStream input) throws IOException { ObjectInput in = CodecSupport.getSerialization(channel.getUrl(), serializationType) .deserialize(channel.getUrl(), input); // 反序列化响应类型 byte flag = in.readByte(); switch (flag) { case DubboCodec.RESPONSE_NULL_VALUE: break; case DubboCodec.RESPONSE_VALUE: handleValue(in); break; case DubboCodec.RESPONSE_WITH_EXCEPTION: handleException(in); break; // 返回值为空，且携带了 attachments 集合 case DubboCodec.RESPONSE_NULL_VALUE_WITH_ATTACHMENTS: handleAttachment(in); break; //返回值不为空，且携带了 attachments 集合 case DubboCodec.RESPONSE_VALUE_WITH_ATTACHMENTS: handleValue(in); handleAttachment(in); break; // 异常对象不为空，且携带了 attachments 集合 case DubboCodec.RESPONSE_WITH_EXCEPTION_WITH_ATTACHMENTS: handleException(in); handleAttachment(in); break; default: throw new IOException("Unknown result flag, expect '0' '1' '2' '3' '4' '5', but received: " + flag); } if (in instanceof Cleanable) { ((Cleanable) in).cleanup(); } return this; }
正常调用下，线程会进入 RESPONSE_VALUE_WITH_ATTACHMENTS 分支中。然后线程会从 invocation 变量（大家探索一下 invocation 变量的由来）中获取返回值类型，接着对调用结果进行反序列化，并将序列化后的结果存储起来。最后对 attachments 集合进行反序列化，并存到指定字段中
（2）获取调用结果
解码完成后，解码结果Response会进入NettyClientHandler，调用路径如下：
NettyServerHandler#channelRead(ChannelHandlerContext, MessageEvent) —> AbstractPeer#received(Channel, Object) —> MultiMessageHandler#received(Channel, Object) —> HeartbeatHandler#received(Channel, Object) —> AllChannelHandler#received(Channel, Object) —> ExecutorService#execute(Runnable) // 由线程池执行后续的调用逻辑 ChannelEventRunnable
跟服务提供者收到请求后的处理逻辑一样，接下来在ChannelEventRunnable中进行处理
最终在HeaderExchangeHandler.received中有处理响应结果的分支
public void received(Channel channel, Object message) throws RemotingException { channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis()); final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel); try { if (message instanceof Request) { // handle request. 处理请求 Request request = (Request) message; if (request.isEvent()) { handlerEvent(channel, request); } else { if (request.isTwoWay()) { /** * 真正处理请求，重点来看 */ handleRequest(exchangeChannel, request); } else { handler.received(exchangeChannel, request.getData()); } } } else if (message instanceof Response) { /** * 处理响应，重点来看 */ handleResponse(channel, (Response) message); } else if (message instanceof String) { if (isClientSide(channel)) { Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl()); logger.error(e.getMessage(), e); } else { String echo = handler.telnet(channel, (String) message); if (echo != null && echo.length() > 0) { channel.send(echo); } } } else { handler.received(exchangeChannel, message); } } finally { HeaderExchangeChannel.removeChannelIfDisconnected(channel); } } static void handleResponse(Channel channel, Response response) throws RemotingException { if (response != null && !response.isHeartbeat()) { DefaultFuture.received(channel, response); } }
然后在DefaultFuture中继续处理
public static void received(Channel channel, Response response) { received(channel, response, false); } public static void received(Channel channel, Response response, boolean timeout) { try { /** * Map<Long, DefaultFuture> FUTURES */ DefaultFuture future = FUTURES.remove(response.getId()); if (future != null) { Timeout t = future.timeoutCheckTask; if (!timeout) { // decrease Time t.cancel(); } future.doReceived(response); } else { logger.warn("The timeout response finally returned at " + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date())) + ", response " + response + (channel == null ? "" : ", channel: " + channel.getLocalAddress() + " -> " + channel.getRemoteAddress())); } } finally { CHANNELS.remove(response.getId()); } } private void doReceived(Response res) { if (res == null) { throw new IllegalStateException("response cannot be null"); } if (res.getStatus() == Response.OK) { /** * 通过`CompletableFuture#complete`方法来设置异步的返回结果 * CompletableFuture 是 jdk 提供的 */ this.complete(res.getResult()); } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) { this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage())); } else { this.completeExceptionally(new RemotingException(channel, res.getErrorMessage())); } }
设置完结果，在哪里获取呢？这得追溯到消费方代理方法的执行，在InvokerInvocationHandler中
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { String methodName = method.getName(); Class<?>[] parameterTypes = method.getParameterTypes(); if (method.getDeclaringClass() == Object.class) { return method.invoke(invoker, args); } if ("toString".equals(methodName) && parameterTypes.length == 0) { return invoker.toString(); } if ("hashCode".equals(methodName) && parameterTypes.length == 0) { return invoker.hashCode(); } if ("equals".equals(methodName) && parameterTypes.length == 1) { return invoker.equals(args[0]); } /** * Invocation 是会话域，它持有调用过程中的变量，比如方法名，参数等 * 将 method 和 args 封装到 RpcInvocation 中，并执行后续的调用 * * invoker: MockClusterInvoker 内部封装了服务降级逻辑 */ return invoker.invoke(new RpcInvocation(method, args)).recreate();// recreate获取结果，在AsyncRpcResult中 } public Object recreate() throws Throwable { RpcInvocation rpcInvocation = (RpcInvocation) invocation; FutureAdapter future = new FutureAdapter(this); RpcContext.getContext().setFuture(future); if (InvokeMode.FUTURE == rpcInvocation.getInvokeMode()) { return future; } // return getAppResponse().recreate(); //原代码 Result appResponse = getAppResponse(); return appResponse.recreate(); } public Result getAppResponse() { try { if (this.isDone()) { /** * this= AsyncRpcResult extends AbstractResult extends CompletableFuture 获取到结果了 */ return this.get(); } } catch (Exception e) { // This should never happen; logger.error("Got exception when trying to fetch the underlying result from AsyncRpcResult.", e); } return new AppResponse(); }
6 异步转同步
Dubbo发送数据至服务方后，在通信层面是异步的，通信线程并不会等待结果数据返回。而我们在使用Dubbo进行RPC调用缺省就是同步的，这其中就涉及到了异步转同步的操作。
而在2.7.x版本中，这种自实现的异步转同步操作进行了修改。新的DefaultFuture继承了CompletableFuture，新的doReceived(Response res)方法如下：
private void doReceived(Response res) { if (res == null) { throw new IllegalStateException("response cannot be null"); } if (res.getStatus() == Response.OK) { this.complete(res.getResult()); } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) { this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage())); } else { this.completeExceptionally(new RemotingException(channel, res.getErrorMessage())); } }
通过CompletableFuture#complete方法来设置异步的返回结果，且删除旧的get()方法，使用CompletableFuture#get()方法：
public T get() throws InterruptedException, ExecutionException { Object r; return reportGet((r = result) == null ? waitingGet(true) : r); }
使用CompletableFuture完成了异步转同步的操作。
7 异步多线程数据一致
这里简单说明一下。一般情况下，服务消费方会并发调用多个服务，每个用户线程发送请求后，会调用 get 方法进行等待。 一段时间后，服务消费方的线程池会收到多个响应对象。这个时候要考虑一个问题，如何将每个响应对象传递给相应的 Future 对象，不出错。答案是通过调用编号。Future 被创建时，会要求传入一个 Request 对象。此时 DefaultFuture 可从 Request 对象中获取调用编号，并将 <调用编号, DefaultFuture 对象> 映射关系存入到静态 Map 中，即 FUTURES。线程池中的线程在收到 Response 对象后，会根据 Response 对象中的调用编号到 FUTURES 集合中取出相应的 DefaultFuture 对象，然后再将 Response 对象设置到 DefaultFuture 对象中。这样用户线程即可从 DefaultFuture 对象中获取调用结果了。整个过程大致如下图：
private DefaultFuture(Channel channel, Request request, int timeout) { this.channel = channel; this.request = request; this.id = request.getId(); this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT); // put into waiting map. FUTURES.put(id, this); CHANNELS.put(id, channel); }
8 心跳检查
Dubbo采用双向心跳的方式检测Client端与Server端的连通性。
我们再来看看 Dubbo 是如何设计应用层心跳的。Dubbo 的心跳是双向心跳，客户端会给服务端发送心跳，反之，服务端也会向客户端发送心跳。
1. 创建定时器
public class HeaderExchangeClient implements ExchangeClient { private final Client client; private final ExchangeChannel channel; private static final HashedWheelTimer IDLE_CHECK_TIMER = new HashedWheelTimer( new NamedThreadFactory("dubbo-client-idleCheck", true), 1, TimeUnit.SECONDS, TICKS_PER_WHEEL); private HeartbeatTimerTask heartBeatTimerTask; private ReconnectTimerTask reconnectTimerTask; public HeaderExchangeClient(Client client, boolean startTimer) { Assert.notNull(client, "Client can't be null"); this.client = client; this.channel = new HeaderExchangeChannel(client); if (startTimer) { URL url = client.getUrl(); //开启心跳失败之后处理重连，断连的逻辑定时任务 startReconnectTask(url); //开启发送心跳请求定时任务 startHeartBeatTask(url); } }
Dubbo 在 HeaderExchangeClient初始化时开启了两个定时任务
startReconnectTask主要用于定时发送心跳请求startHeartBeatTask主要用于心跳失败之后处理重连，断连的逻辑
2. 发送心跳请求
详细解析下心跳检测定时任务的逻辑 HeartbeatTimerTask#doTask：
protected void doTask(Channel channel) { Long lastRead = lastRead(channel); Long lastWrite = lastWrite(channel); if ((lastRead != null && now() - lastRead > heartbeat) || (lastWrite != null && now() - lastWrite > heartbeat)) { Request req = new Request(); req.setVersion(Version.getProtocolVersion()); req.setTwoWay(true); req.setEvent(Request.HEARTBEAT_EVENT); channel.send(req); } }
前面已经介绍过，Dubbo 采取的是双向心跳设计，即服务端会向客户端发送心跳，客户端也会向服务端发送心跳，接收的一方更新 lastRead 字段，发送的一方更新 lastWrite 字段，超过心跳间隙的时间，便发送心跳请求给对端。这里的 lastRead/lastWrite 同样会被同一个通道上的普通调用更新，通过更新这两个字段，实现了只在连接空闲时才会真正发送空闲报文的机制，符合我们一开始科普的做法。
3. 处理重连和断连
继续研究下重连和断连定时器都实现了什么 ReconnectTimerTask#doTask。
protected void doTask(Channel channel) { Long lastRead = lastRead(channel); Long now = now(); if (!channel.isConnected()) { ((Client) channel).reconnect(); // check pong at client } else if (lastRead != null && now - lastRead > idleTimeout) { ((Client) channel).reconnect(); } }
第二个定时器则负责根据客户端、服务端类型来对连接做不同的处理，当超过设置的心跳总时间之后，客户端选择的是重新连接，服务端则是选择直接断开连接。这样的考虑是合理的，客户端调用是强依赖可用连接的，而服务端可以等待客户端重新建立连接。
Dubbo 对于建立的每一个连接，同时在客户端和服务端开启了 2 个定时器，一个用于定时发送心跳，一个用于定时重连、断连，执行的频率均为各自检测周期的 1/3。定时发送心跳的任务负责在连接空闲时，向对端发送心跳包。定时重连、断连的任务负责检测 lastRead 是否在超时周期内仍未被更新，如果判定为超时，客户端处理的逻辑是重连，服务端则采取断连的措施。
