优秀的编程知识分享平台

网站首页 > 技术文章 正文

spark2.1.0之源码分析——RPC客户端TransportClient详解

nanyue 2024-08-03 18:12:42 技术文章 5 ℃

代码清单1 TransportClient的构造器


  2. public TransportClient(Channel channel, TransportResponseHandler handler) {
  3. this.channel = Preconditions.checkNotNull(channel);
  4. this.handler = Preconditions.checkNotNull(handler);
  5. this.timedOut = false;
  6. }

TransportClient一共有五个方法用于发送请求，分别为：

  1. fetchChunk：从远端协商好的流中请求单个块；
  2. stream：使用流的ID，从远端获取流数据；
  3. sendRpc：向服务端发送RPC的请求，通过At least Once Delivery原则保证请求不会丢失；
  4. sendRpcSync：向服务端发送异步的RPC的请求，并根据指定的超时时间等待响应；
  5. send：向服务端发送RPC的请求，但是并不期望能获取响应，因而不能保证投递的可靠性；

本节只选择最常用的sendRpc和fetchChunk进行分析，其余实现都可以触类旁通。

发送RPC请求

sendRpc方法的实现见代码清单2。

代码清单2 sendRpc的实现


  2. public long sendRpc(ByteBuffer message, final RpcResponseCallback callback) {
  3. final long startTime = System.currentTimeMillis();
  4. if (logger.isTraceEnabled()) {
  5. logger.trace("Sending RPC to {}", getRemoteAddress(channel));
  6. }
  7. // 使用UUID生成请求主键requestId
  8. final long requestId = Math.abs(UUID.randomUUID().getLeastSignificantBits());
  9. handler.addRpcRequest(requestId, callback);// 添加requestId与RpcResponseCallback的引用之间的关系
  10. // 发送RPC请求
  11. channel.writeAndFlush(new RpcRequest(requestId, new NioManagedBuffer(message))).addListener(
  12. new ChannelFutureListener() {
  13. @Override
  14. public void operationComplete(ChannelFuture future) throws Exception {
  15. if (future.isSuccess()) {
  16. long timeTaken = System.currentTimeMillis() - startTime;
  17. if (logger.isTraceEnabled()) {
  18. logger.trace("Sending request {} to {} took {} ms", requestId,
  19. getRemoteAddress(channel), timeTaken);
  20. }
  21. } else {
  22. String errorMsg = String.format("Failed to send RPC %s to %s: %s", requestId,
  23. getRemoteAddress(channel), future.cause());
  24. logger.error(errorMsg, future.cause());
  25. handler.removeRpcRequest(requestId);
  26. channel.close();
  27. try {
  28. callback.onFailure(new IOException(errorMsg, future.cause()));
  29. } catch (Exception e) {
  30. logger.error("Uncaught exception in RPC response callback handler!", e);
  31. }
  32. }
  33. }
  34. });
  36. return requestId;
  37. }

结合代码清单2，我们知道sendRpc方法的实现步骤如下：

  1. 使用UUID生成请求主键requestId；
  2. 调用addRpcRequest向handler（特别提醒下读者这里的handler不是RpcHandler,而是通过TransportClient构造器传入的TransportResponseHandler）添加requestId与回调类RpcResponseCallback的引用之间的关系。TransportResponseHandler的addRpcRequest方法（见代码清单3）将更新最后一次请求的时间为当前系统时间，然后将requestId与RpcResponseCallback之间的映射加入到outstandingRpcs缓存中。outstandingRpcs专门用于缓存发出的RPC请求信息。
  3. 调用Channel的writeAndFlush方法将RPC请求发送出去，这和在《spark2.1.0之源码分析——服务端RPC处理器RpcHandler详解》一文列出的代码清单7中服务端调用的respond方法响应客户端的一样，都是使用Channel的writeAndFlush方法。当发送成功或者失败时会回调ChannelFutureListener的operationComplete方法。如果发送成功，那么只会打印requestId、远端地址及花费时间的日志，如果发送失败，除了打印错误日志外，还要调用TransportResponseHandler的removeRpcRequest方法（见代码清单4）将此次请求从outstandingRpcs缓存中移除。

代码清单3 添加RPC请求到缓存


  2. public void addRpcRequest(long requestId, RpcResponseCallback callback) {
  3. updateTimeOfLastRequest();
  4. outstandingRpcs.put(requestId, callback);
  5. }

代码清单4 从缓存中删除RPC请求


  2. public void removeRpcRequest(long requestId) {
  3. outstandingRpcs.remove(requestId);
  4. }

请求发送成功后，客户端将等待接收服务端的响应。根据《spark2.1.0之源码分析——RPC管道初始化》一文的图1，返回的消息也会传递给TransportChannelHandler的channelRead方法（见《spark2.1.0之源码分析——RPC传输管道处理器详解》一文的代码清单1），根据之前的分析，消息的分析将最后交给TransportResponseHandler的handle方法来处理。TransportResponseHandler的handle方法分别对《spark2.1.0之源码分析——RPC传输管道处理器详解》一文的图2中的六种ResponseMessage进行处理，由于服务端使用processRpcRequest方法（见《spark2.1.0之源码分析——服务端RPC处理器RpcHandler详解》一文的代码清单4）处理RpcRequest类型的消息后返回给客户端的消息为RpcResponse或RpcFailure，所以我们来看看客户端的TransportResponseHandler的handle方法是如何处理RpcResponse和RpcFailure，见代码清单5。

代码清单5 RpcResponse和RpcFailure消息的处理


  2. } else if (message instanceof RpcResponse) {
  3. RpcResponse resp = (RpcResponse) message;
  4. RpcResponseCallback listener = outstandingRpcs.get(resp.requestId);// 获取RpcResponseCallback
  5. if (listener == null) {
  6. logger.warn("Ignoring response for RPC {} from {} ({} bytes) since it is not outstanding",
  7. resp.requestId, getRemoteAddress(channel), resp.body().size());
  8. } else {
  9. outstandingRpcs.remove(resp.requestId);
  10. try {
  11. listener.onSuccess(resp.body().nioByteBuffer());
  12. } finally {
  13. resp.body().release();
  14. }
  15. }
  16. } else if (message instanceof RpcFailure) {
  17. RpcFailure resp = (RpcFailure) message;
  18. RpcResponseCallback listener = outstandingRpcs.get(resp.requestId); // 获取RpcResponseCallback
  19. if (listener == null) {
  20. logger.warn("Ignoring response for RPC {} from {} ({}) since it is not outstanding",
  21. resp.requestId, getRemoteAddress(channel), resp.errorString);
  22. } else {
  23. outstandingRpcs.remove(resp.requestId);
  24. listener.onFailure(new RuntimeException(resp.errorString));
  25. }

从代码清单5看到，处理RpcResponse的逻辑为：

  1. 使用RpcResponse对应的RpcRequest的主键requestId，从outstandingRpcs缓存中获取注册的RpcResponseCallback，此处的RpcResponseCallback即为代码清单2中传递给sendRpc方法的RpcResponseCallback；
  2. 移除outstandingRpcs缓存中requestId和RpcResponseCallback的注册信息；
  3. 调用RpcResponseCallback的onSuccess方法，处理成功响应后的具体逻辑。这里的RpcResponseCallback需要各个使用TransportClient的sendRpc方法的场景中分别实现；
  4. 最后释放RpcResponse的body，回收资源。

处理RpcFailure的逻辑为：

  1. 使用RpcFailure对应的RpcRequest的主键requestId，从outstandingRpcs缓存中获取注册的RpcResponseCallback，此处的RpcResponseCallback即为代码清单2中传递给sendRpc方法的RpcResponseCallback；
  2. 移除outstandingRpcs缓存中requestId和RpcResponseCallback的注册信息；
  3. 调用RpcResponseCallback的onFailure方法，处理失败响应后的具体逻辑。这里的RpcResponseCallback需要在使用TransportClient的sendRpc方法时指定或实现。

发送获取块请求

fetchChunk的实现见代码清单6。

代码清单6 fetchChunk的实现


  2. public void fetchChunk(
  3. long streamId,
  4. final int chunkIndex,
  5. final ChunkReceivedCallback callback) {
  6. final long startTime = System.currentTimeMillis();
  7. if (logger.isDebugEnabled()) {
  8. logger.debug("Sending fetch chunk request {} to {}", chunkIndex, getRemoteAddress(channel));
  9. }
  11. final StreamChunkId streamChunkId = new StreamChunkId(streamId, chunkIndex);// 创建StreamChunkId
  12. // 添加StreamChunkId与ChunkReceivedCallback之间的对应关系
  13. handler.addFetchRequest(streamChunkId, callback);
  14. // 发送块请求
  15. channel.writeAndFlush(new ChunkFetchRequest(streamChunkId)).addListener(
  16. new ChannelFutureListener() {
  17. @Override
  18. public void operationComplete(ChannelFuture future) throws Exception {
  19. if (future.isSuccess()) {
  20. long timeTaken = System.currentTimeMillis() - startTime;
  21. if (logger.isTraceEnabled()) {
  22. logger.trace("Sending request {} to {} took {} ms", streamChunkId,
  23. getRemoteAddress(channel), timeTaken);
  24. }
  25. } else {
  26. String errorMsg = String.format("Failed to send request %s to %s: %s", streamChunkId,
  27. getRemoteAddress(channel), future.cause());
  28. logger.error(errorMsg, future.cause());
  29. handler.removeFetchRequest(streamChunkId);
  30. channel.close();
  31. try {
  32. callback.onFailure(chunkIndex, new IOException(errorMsg, future.cause()));
  33. } catch (Exception e) {
  34. logger.error("Uncaught exception in RPC response callback handler!", e);
  35. }
  36. }
  37. }
  38. });
  39. }

结合代码清单6，我们知道fetchChunk方法的实现步骤如下：

  1. 使用流的标记streamId和块的索引chunkIndex创建StreamChunkId；
  2. 调用addFetchRequest向handler（特别提醒下读者这里的handler不是RpcHandler,而是通过TransportClient构造器传入的TransportResponseHandler）添加StreamChunkId与回调类ChunkReceivedCallback的引用之间的关系。TransportResponseHandler的addFetchRequest方法（见代码清单7）将更新最后一次请求的时间为当前系统时间，然后将StreamChunkId与ChunkReceivedCallback之间的映射加入到outstandingFetches缓存中。outstandingFetches专门用于缓存发出的块请求信息。
  3. 调用Channel的writeAndFlush方法将块请求发送出去，这和在《spark2.1.0之源码分析——服务端RPC处理器RpcHandler详解》一文列出的代码清单7中服务端调用的respond方法响应客户端的一样，都是使用Channel的writeAndFlush方法。当发送成功或者失败时会回调ChannelFutureListener的operationComplete方法。如果发送成功，那么只会打印StreamChunkId、远端地址及花费时间的日志，如果发送失败，除了打印错误日志外，还要调用TransportResponseHandler的removeFetchRequest方法（见代码清单8）将此次请求从outstandingFetches缓存中移除。

代码清单7 添加块请求到缓存


  2. public void addFetchRequest(StreamChunkId streamChunkId, ChunkReceivedCallback callback) {
  3. updateTimeOfLastRequest();
  4. outstandingFetches.put(streamChunkId, callback);
  5. }

代码清单8 从缓存中删除块请求


  2. public void removeFetchRequest(StreamChunkId streamChunkId) {
  3. outstandingFetches.remove(streamChunkId);
  4. }

请求发送成功后，客户端将等待接收服务端的响应。根据《spark2.1.0之源码分析——RPC管道初始化》一文的图1，返回的消息也会传递给TransportChannelHandler的channelRead方法（见《spark2.1.0之源码分析——RPC传输管道处理器详解》一文的代码清单1），根据之前的分析，消息的分析将最后交给TransportResponseHandler的handle方法来处理。TransportResponseHandler的handle方法分别对《spark2.1.0之源码分析——RPC传输管道处理器详解》一文的图2中的六种处理结果进行处理，由于服务端使用processFetchRequest方法（见《spark2.1.0之源码分析——服务端RPC处理器RpcHandler详解》一文的代码清单3）处理ChunkFetchRequest类型的消息后返回给客户端的消息为ChunkFetchSuccess或ChunkFetchFailure，所以我们来看看客户端的TransportResponseHandler的handle方法是如何处理ChunkFetchSuccess和ChunkFetchFailure，见代码清单9

代码清单9 ChunkFetchSuccess和ChunkFetchFailure消息的处理


  2. if (message instanceof ChunkFetchSuccess) {
  3. ChunkFetchSuccess resp = (ChunkFetchSuccess) message;
  4. ChunkReceivedCallback listener = outstandingFetches.get(resp.streamChunkId);
  5. if (listener == null) {
  6. logger.warn("Ignoring response for block {} from {} since it is not outstanding",
  7. resp.streamChunkId, getRemoteAddress(channel));
  8. resp.body().release();
  9. } else {
  10. outstandingFetches.remove(resp.streamChunkId);
  11. listener.onSuccess(resp.streamChunkId.chunkIndex, resp.body());
  12. resp.body().release();
  13. }
  14. } else if (message instanceof ChunkFetchFailure) {
  15. ChunkFetchFailure resp = (ChunkFetchFailure) message;
  16. ChunkReceivedCallback listener = outstandingFetches.get(resp.streamChunkId);
  17. if (listener == null) {
  18. logger.warn("Ignoring response for block {} from {} ({}) since it is not outstanding",
  19. resp.streamChunkId, getRemoteAddress(channel), resp.errorString);
  20. } else {
  21. outstandingFetches.remove(resp.streamChunkId);
  22. listener.onFailure(resp.streamChunkId.chunkIndex, new ChunkFetchFailureException(
  23. "Failure while fetching " + resp.streamChunkId + ": " + resp.errorString));
  24. }
  25. }

从代码清单9看到，处理ChunkFetchSuccess的逻辑为：

  1. 使用ChunkFetchSuccess对应的StreamChunkId，从outstandingFetches缓存中获取注册的ChunkReceivedCallback，此处的ChunkReceivedCallback即为代码清单6中传递给fetchChunk方法的ChunkReceivedCallback；
  2. 移除outstandingFetches缓存中StreamChunkId和ChunkReceivedCallback的注册信息；
  3. 调用ChunkReceivedCallback的onSuccess方法，处理成功响应后的具体逻辑。这里的ChunkReceivedCallback需要各个使用TransportClient的fetchChunk方法的场景中分别实现；
  4. 最后释放ChunkFetchSuccess的body，回收资源。

处理ChunkFetchFailure的逻辑为：

  1. 使用ChunkFetchFailure对应的StreamChunkId，从outstandingFetches缓存中获取注册的ChunkReceivedCallback，此处的ChunkReceivedCallback即为代码清单6中传递给fetchChunk方法的ChunkReceivedCallback；
  2. 移除outstandingFetches缓存中StreamChunkId和ChunkReceivedCallback的注册信息；
  3. 调用ChunkReceivedCallback的onFailure方法，处理失败响应后的具体逻辑。这里的ChunkReceivedCallback需要各个使用TransportClient的fetchChunk方法的场景中分别实现。

在详细介绍了TransportClient和TransportResponseHandler之后，对于客户端我们就可以扩展《spark2.1.0之源码分析——RPC管道初始化》一文的图1，把TransportResponseHandler及TransportClient的处理流程增加进来，如下图所示。

客户端请求、响应流程图

上图中的序号①表示调用TransportResponseHandler的addRpcRequest方法（或addFetchRequest方法）将更新最后一次请求的时间为当前系统时间，然后将requestId与RpcResponseCallback之间的映射加入到outstandingRpcs缓存中（或将StreamChunkId与ChunkReceivedCallback之间的映射加入到outstandingFetches缓存中）。②表示调用Channel的writeAndFlush方法将RPC请求发送出去。图中的虚线表示当TransportResponseHandler处理RpcResponse和RpcFailure时将从outstandingRpcs缓存中获取此请求对应的RpcResponseCallback（或处理ChunkFetchSuccess和ChunkFetchFailure时将从outstandingFetches缓存中获取StreamChunkId对应的ChunkReceivedCallback），并执行回调。此外，TransportClientBootstrap将可能存在于上图中任何两个组件的箭头连线中间。

Tags：

猜你喜欢

最近发表
标签列表