哈尔滨企业网站seo,网页怎么弄,做网站费,价格低廉的意思Spark的RPC通信二-初稿
Spark RPC的传输层
传输层主要还是借助netty框架进行实现。
TransportContext包含创建 TransportServer、TransportClientFactory 和使用 TransportChannelHandler 设置 Netty Channel 管道的上下文。TransportClient 提供两种通信协议#xff1a;co…Spark的RPC通信二-初稿
Spark RPC的传输层
传输层主要还是借助netty框架进行实现。
TransportContext包含创建 TransportServer、TransportClientFactory 和使用 TransportChannelHandler 设置 Netty Channel 管道的上下文。TransportClient 提供两种通信协议control-plane RPCs 和data-plane的 “chunk fetching”。RPC 的处理在 TransportContext 的范围之外进行即由用户提供的处理程序执行它负责设置流这些流可以使用零拷贝 IO 以块为单位通过数据平面进行流式传输。对消息的处理由RpcHandler处理。TransportServer 和 TransportClientFactory 都会为每个通道创建一个 TransportChannelHandler。由于每个 TransportChannelHandler 都包含一个 TransportClient因此服务器进程可以通过现有通道向客户端发送消息。 #mermaid-svg-RZKgMJJbCOL4j4rf {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-RZKgMJJbCOL4j4rf .error-icon{fill:#552222;}#mermaid-svg-RZKgMJJbCOL4j4rf .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-RZKgMJJbCOL4j4rf .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-RZKgMJJbCOL4j4rf .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-RZKgMJJbCOL4j4rf .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-RZKgMJJbCOL4j4rf .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-RZKgMJJbCOL4j4rf .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-RZKgMJJbCOL4j4rf .marker{fill:#333333;stroke:#333333;}#mermaid-svg-RZKgMJJbCOL4j4rf .marker.cross{stroke:#333333;}#mermaid-svg-RZKgMJJbCOL4j4rf svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-RZKgMJJbCOL4j4rf g.classGroup text{fill:#9370DB;fill:#131300;stroke:none;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:10px;}#mermaid-svg-RZKgMJJbCOL4j4rf g.classGroup text .title{font-weight:bolder;}#mermaid-svg-RZKgMJJbCOL4j4rf .nodeLabel,#mermaid-svg-RZKgMJJbCOL4j4rf .edgeLabel{color:#131300;}#mermaid-svg-RZKgMJJbCOL4j4rf .edgeLabel .label rect{fill:#ECECFF;}#mermaid-svg-RZKgMJJbCOL4j4rf .label text{fill:#131300;}#mermaid-svg-RZKgMJJbCOL4j4rf .edgeLabel .label span{background:#ECECFF;}#mermaid-svg-RZKgMJJbCOL4j4rf .classTitle{font-weight:bolder;}#mermaid-svg-RZKgMJJbCOL4j4rf .node rect,#mermaid-svg-RZKgMJJbCOL4j4rf .node circle,#mermaid-svg-RZKgMJJbCOL4j4rf .node ellipse,#mermaid-svg-RZKgMJJbCOL4j4rf .node polygon,#mermaid-svg-RZKgMJJbCOL4j4rf .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-RZKgMJJbCOL4j4rf .divider{stroke:#9370DB;stroke:1;}#mermaid-svg-RZKgMJJbCOL4j4rf g.clickable{cursor:pointer;}#mermaid-svg-RZKgMJJbCOL4j4rf g.classGroup rect{fill:#ECECFF;stroke:#9370DB;}#mermaid-svg-RZKgMJJbCOL4j4rf g.classGroup line{stroke:#9370DB;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf .classLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5;}#mermaid-svg-RZKgMJJbCOL4j4rf .classLabel .label{fill:#9370DB;font-size:10px;}#mermaid-svg-RZKgMJJbCOL4j4rf .relation{stroke:#333333;stroke-width:1;fill:none;}#mermaid-svg-RZKgMJJbCOL4j4rf .dashed-line{stroke-dasharray:3;}#mermaid-svg-RZKgMJJbCOL4j4rf #compositionStart,#mermaid-svg-RZKgMJJbCOL4j4rf .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #compositionEnd,#mermaid-svg-RZKgMJJbCOL4j4rf .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #dependencyStart,#mermaid-svg-RZKgMJJbCOL4j4rf .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #dependencyStart,#mermaid-svg-RZKgMJJbCOL4j4rf .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #extensionStart,#mermaid-svg-RZKgMJJbCOL4j4rf .extension{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #extensionEnd,#mermaid-svg-RZKgMJJbCOL4j4rf .extension{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #aggregationStart,#mermaid-svg-RZKgMJJbCOL4j4rf .aggregation{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf #aggregationEnd,#mermaid-svg-RZKgMJJbCOL4j4rf .aggregation{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-RZKgMJJbCOL4j4rf .edgeTerminals{font-size:11px;}#mermaid-svg-RZKgMJJbCOL4j4rf :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;} TransportContext -TransportConf conf -RpcHandler rpcHandler -boolean closeIdleConnections -boolean isClientOnly -MessageEncoder ENCODER -MessageEncoder DECODER -EventLoopGroup chunkFetchWorkers TransportClientFactory createClientFactory() TransportServer createServer() TransportChannelHandler createChannelHandler() TransportChannelHandler initializePipeline() ClientPool TransportClient[] clients Object[] lock TransportClientFactory -TransportContext context -TransportConf conf -List clientBootstraps -ConcurrentHashMap connectionPool -int numConnectionsPerPeer -final Class socketChannelClass -EventLoopGroup workerGroup -PooledByteBufAllocator pooledAllocator -NettyMemoryMetrics metrics «interface» TransportClientBootstrap void doBootstrap(TransportClient client, Channel channel) «interface» TransportServerBootstrap RpcHandler doBootstrap(Channel channel, RpcHandler rpcHandler) «abstract» MessageHandler abstract void handle(T message) abstract void channelActive() abstract void exceptionCaught(Throwable cause) abstract void channelInactive() «interface» Message «abstract» RpcHandler TransportRequestHandler RpcHandler rpcHandler StreamManager streamManager TransportResponseHandler TransportClient TransportServer TransportChannelHandler 传输上下文TransportContext
TransportContext的核心成员与核心方法
TransportConf conf传输的配置信息RpcHandler rpcHandler对接收的RPC消息进行处理EventLoopGroup chunkFetchWorkers处理 ChunkFetchRequest 的独立线程池。这有助于控制通过底层通道将 ChunkFetchRequest 信息写回客户端时阻塞的 TransportServer 工作线程的最大数量。createClientFactory()初始化 ClientFactory在返回新客户端之前运行给定的 TransportClientBootstraps。Bootstraps 将同步执行并且必须成功运行才能创建客户端。createServer()创建传输服务端TransportServer的实例initializePipeline()对TransportClientTransportRequestHandlerTransportResponseHandler进行初始化然后在用其构造TransportChannelHandler对象。借助Netty的API对管道进行配置。
TransportContext的createClientFactory方法创建传输客户端工厂TransportClientFactory的实例。在构造TransportClientFactory的实例时还会传递客户端引导程序TransportClientBootstrap的列表。TransportClientFactory内部维护每个Socket地址的连接池。通过调用TransportContext的createServer方法创建传输服务端TransportServer的实例。
核心类TransportClientFactory
用于使用 createClient方法 创建 TransportClients 的工厂。该工厂负责维护与其他主机的连接池并为同一远程主机返回相同的 TransportClient。它还为所有 TransportClients 共享一个工作线程池。只要有可能就会重复使用 TransportClients。在完成创建新的 TransportClient 之前将运行所有给定的 TransportClientBootstraps。
TransportClientFactory的核心成员和核心方法 静态内部类ClientPool一种简单的数据结构用于跟踪两个对等节点之间的客户端连接池保障其可以复用由于线程不安全所以增加了客户端对应的锁。 private static class ClientPool {TransportClient[] clients;Object[] locks;ClientPool(int size) {clients new TransportClient[size];locks new Object[size];for (int i 0; i size; i) {locks[i] new Object();}}}TransportContext contextTransportContext 的实例对象 TransportConf conf链接配置信息的实例对象 ListTransportClientBootstrap clientBootstraps客户端的引导程序主要是客户端在建立连接的时候进行一些初始化的准备操作。 ConcurrentHashMapSocketAddress, ClientPool connectionPool维护了连接地址上的客户端连接池的映射表。 createClient(String remoteHost, int remotePort) 首先根据远程地址确认客户端连接池connectionPool中是否存在关于这个地址的客户端池clientPool如果没有就新建一个客户端池放入连接池中。检查通道是否超时和客户端是否存活如果客户端失活则需要重建一个客户端。创建客户端的在createClient(InetSocketAddress address)方法中。 public TransportClient createClient(String remoteHost, int remotePort)throws IOException, InterruptedException {// 此处使用未解析地址以避免每次创建客户端时都进行 DNS 解析。final InetSocketAddress unresolvedAddress InetSocketAddress.createUnresolved(remoteHost, remotePort);// 如果clientPool不存在则新建.ClientPool clientPool connectionPool.get(unresolvedAddress);if (clientPool null) {connectionPool.putIfAbsent(unresolvedAddress, new ClientPool(numConnectionsPerPeer));clientPool connectionPool.get(unresolvedAddress);}int clientIndex rand.nextInt(numConnectionsPerPeer);TransportClient cachedClient clientPool.clients[clientIndex];if (cachedClient ! null cachedClient.isActive()) {// 更新处理程序的最后使用时间确保通道不会超时TransportChannelHandler handler cachedClient.getChannel().pipeline().get(TransportChannelHandler.class);synchronized (handler) {handler.getResponseHandler().updateTimeOfLastRequest();}// 然后检查客户端是否还活着以防在代码更新之前超时。if (cachedClient.isActive()) {logger.trace(Returning cached connection to {}: {},cachedClient.getSocketAddress(), cachedClient);return cachedClient;}}// 如果我们到达这里就没有打开现有连接尝试创建一个新连接。final long preResolveHost System.nanoTime();final InetSocketAddress resolvedAddress new InetSocketAddress(remoteHost, remotePort);final long hostResolveTimeMs (System.nanoTime() - preResolveHost) / 1000000;if (hostResolveTimeMs 2000) {logger.warn(DNS resolution for {} took {} ms, resolvedAddress, hostResolveTimeMs);} else {logger.trace(DNS resolution for {} took {} ms, resolvedAddress, hostResolveTimeMs);}// 多个线程可能会竞相在这里创建新连接。通过同步原语只保留其中一个处于活动状态。synchronized (clientPool.locks[clientIndex]) {cachedClient clientPool.clients[clientIndex];if (cachedClient ! null) {if (cachedClient.isActive()) {logger.trace(Returning cached connection to {}: {}, resolvedAddress, cachedClient);return cachedClient;} else {logger.info(Found inactive connection to {}, creating a new one., resolvedAddress);}}clientPool.clients[clientIndex] createClient(resolvedAddress);return clientPool.clients[clientIndex];}}createClient(InetSocketAddress address) 通过Netty的根引导程序进行初始化配置通过回调函数初始化bootstrap的Pipeline设置好客户端引用和管道引用。遍历客户端引导程序集clientBootstraps执行其初始化的内容 private TransportClient createClient(InetSocketAddress address)throws IOException, InterruptedException {logger.debug(Creating new connection to {}, address);// netty的连接创建的根引导程序Bootstrap bootstrap new Bootstrap();bootstrap.group(workerGroup).channel(socketChannelClass)// 禁用纳格尔算法因为我们不想让数据包等待.option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true).option(ChannelOption.CONNECT_TIMEOUT_MILLIS, conf.connectionTimeoutMs()).option(ChannelOption.ALLOCATOR, pooledAllocator);if (conf.receiveBuf() 0) {bootstrap.option(ChannelOption.SO_RCVBUF, conf.receiveBuf());}if (conf.sendBuf() 0) {bootstrap.option(ChannelOption.SO_SNDBUF, conf.sendBuf());}final AtomicReferenceTransportClient clientRef new AtomicReference();final AtomicReferenceChannel channelRef new AtomicReference();// 通过回调函数初始化bootstrap的Pipelinebootstrap.handler(new ChannelInitializerSocketChannel() {Overridepublic void initChannel(SocketChannel ch) {TransportChannelHandler clientHandler context.initializePipeline(ch);clientRef.set(clientHandler.getClient());channelRef.set(ch);}});// 连接远程服务器long preConnect System.nanoTime();ChannelFuture cf bootstrap.connect(address);if (!cf.await(conf.connectionTimeoutMs())) {throw new IOException(String.format(Connecting to %s timed out (%s ms), address, conf.connectionTimeoutMs()));} else if (cf.cause() ! null) {throw new IOException(String.format(Failed to connect to %s, address), cf.cause());}TransportClient client clientRef.get();Channel channel channelRef.get();assert client ! null : Channel future completed successfully with null client;// 在将客户端标记为成功之前同步执行任何客户端引导。long preBootstrap System.nanoTime();logger.debug(Connection to {} successful, running bootstraps..., address);try {// 遍历客户端引导程序集clientBootstraps执行其初始化的内容for (TransportClientBootstrap clientBootstrap : clientBootstraps) {clientBootstrap.doBootstrap(client, channel);}} catch (Exception e) { // catch non-RuntimeExceptions too as bootstrap may be written in Scalalong bootstrapTimeMs (System.nanoTime() - preBootstrap) / 1000000;logger.error(Exception while bootstrapping client after bootstrapTimeMs ms, e);client.close();throw Throwables.propagate(e);}long postBootstrap System.nanoTime();logger.info(Successfully created connection to {} after {} ms ({} ms spent in bootstraps),address, (postBootstrap - preConnect) / 1000000, (postBootstrap - preBootstrap) / 1000000);return client;}TransportClient
用于向server端发送rpc请求和从server 端获取流的chunk块旨在高效传输大量数据这些数据被分成大小从几百 KB 到几 MB 不等的数据块。
典型流程
// 打开远程文件
client.sendRPC(new OpenFile(/foo)) -- returns StreamId 100
// 获取远程文件的chunk
client.fetchChunk(streamId 100, chunkIndex 0, callback)
client.fetchChunk(streamId 100, chunkIndex 1, callback)
// 关闭远程文件
client.sendRPC(new CloseStream(100))用于获取预协商数据流中连续数据块的客户端处理的是从数据流即数据平面中获取数据块的过程但数据流的实际设置是在传输层范围之外完成的。提供 sendRPC 方便方法是为了在客户端和服务器之间进行控制平面通信以执行此设置。使用 TransportClientFactory 构建一个 TransportClient 实例。单个 TransportClient 可用于多个流但任何给定的流都必须仅限于单个客户端以避免响应顺序混乱。注意该类用于向服务器发出请求而 TransportResponseHandler 则负责处理来自服务器的响应。并发性线程安全可由多个线程调用。
TransportServer
服务器提供高效的底层流媒体服务。
消息的处理
消息处理类MessageHandler处理来自 Netty 的请求或响应信息。一个 MessageHandler 实例只与一个Netty通道相关联尽管同一通道上可能有多个客户端。以下是其定义的抽象方法。
abstract void handle(T message)对接收的单条信息的处理。abstract void channelActive()当该消息处理程序所在的频道处于活动状态时调用。abstract void exceptionCaught(Throwable cause)当通道上出现异常时调用。abstract void channelInactive()当此 MessageHandler 所处的通道处于非活动状态时调用。
MessageHandler有两个继承类TransportRequestHandler和TransportResponseHandler分别用来进行Server端处理Client的请求信息和Client端处理Server的响应信息。
TransportRequestHandler的handle(RequestMessage request)方法 public void handle(RequestMessage request) {if (request instanceof RpcRequest) {// 处理RPC请求依赖RpcHandler的receive()方法processRpcRequest((RpcRequest) request);} else if (request instanceof OneWayMessage) {// 处理无需回复的RPC请求依赖RpcHandler的receive()方法processOneWayMessage((OneWayMessage) request);} else if (request instanceof StreamRequest) {// 处理流请求依赖StreamManager的openStream()方法获取流数据并封装成ManagedBufferprocessStreamRequest((StreamRequest) request);} else {// 未知请求抛异常throw new IllegalArgumentException(Unknown request type: request);}}TransportResponseHandler的handle(ResponseMessage message)方法
在client端发送消息时根据发送消息的类型调用TransportResponseHandler中的方法注册回调函数回调函数和请求信息放入相应的缓存中。
待TransportResponseHandler收到server端的响应消息时再调用主要的工作方法handle()根据响应消息类型从对应缓存中取出回调函数并调用。 Overridepublic void handle(ResponseMessage message) throws Exception {if (message instanceof ChunkFetchSuccess) {ChunkFetchSuccess resp (ChunkFetchSuccess) message;ChunkReceivedCallback listener outstandingFetches.get(resp.streamChunkId);if (listener null) {logger.warn(Ignoring response for block {} from {} since it is not outstanding,resp.streamChunkId, getRemoteAddress(channel));resp.body().release();} else {outstandingFetches.remove(resp.streamChunkId);listener.onSuccess(resp.streamChunkId.chunkIndex, resp.body());resp.body().release();}} else if (message instanceof ChunkFetchFailure) {ChunkFetchFailure resp (ChunkFetchFailure) message;ChunkReceivedCallback listener outstandingFetches.get(resp.streamChunkId);if (listener null) {logger.warn(Ignoring response for block {} from {} ({}) since it is not outstanding,resp.streamChunkId, getRemoteAddress(channel), resp.errorString);} else {outstandingFetches.remove(resp.streamChunkId);listener.onFailure(resp.streamChunkId.chunkIndex, new ChunkFetchFailureException(Failure while fetching resp.streamChunkId : resp.errorString));}} else if (message instanceof RpcResponse) {RpcResponse resp (RpcResponse) message;RpcResponseCallback listener outstandingRpcs.get(resp.requestId);if (listener null) {logger.warn(Ignoring response for RPC {} from {} ({} bytes) since it is not outstanding,resp.requestId, getRemoteAddress(channel), resp.body().size());} else {outstandingRpcs.remove(resp.requestId);try {listener.onSuccess(resp.body().nioByteBuffer());} finally {resp.body().release();}}} else if (message instanceof RpcFailure) {RpcFailure resp (RpcFailure) message;RpcResponseCallback listener outstandingRpcs.get(resp.requestId);if (listener null) {logger.warn(Ignoring response for RPC {} from {} ({}) since it is not outstanding,resp.requestId, getRemoteAddress(channel), resp.errorString);} else {outstandingRpcs.remove(resp.requestId);listener.onFailure(new RuntimeException(resp.errorString));}} else if (message instanceof StreamResponse) {StreamResponse resp (StreamResponse) message;PairString, StreamCallback entry streamCallbacks.poll();if (entry ! null) {StreamCallback callback entry.getValue();if (resp.byteCount 0) {StreamInterceptor interceptor new StreamInterceptor(this, resp.streamId, resp.byteCount,callback);try {TransportFrameDecoder frameDecoder (TransportFrameDecoder)channel.pipeline().get(TransportFrameDecoder.HANDLER_NAME);frameDecoder.setInterceptor(interceptor);streamActive true;} catch (Exception e) {logger.error(Error installing stream handler., e);deactivateStream();}} else {try {callback.onComplete(resp.streamId);} catch (Exception e) {logger.warn(Error in stream handler onComplete()., e);}}} else {logger.error(Could not find callback for StreamResponse.);}} else if (message instanceof StreamFailure) {StreamFailure resp (StreamFailure) message;PairString, StreamCallback entry streamCallbacks.poll();if (entry ! null) {StreamCallback callback entry.getValue();try {callback.onFailure(resp.streamId, new RuntimeException(resp.error));} catch (IOException ioe) {logger.warn(Error in stream failure handler., ioe);}} else {logger.warn(Stream failure with unknown callback: {}, resp.error);}} else {throw new IllegalStateException(Unknown response type: message.type());}}消息的分类
MessageHandler用来处理的消息都是继承或实现自Message接口的。 #mermaid-svg-utrongvlch8I48n8 {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-utrongvlch8I48n8 .error-icon{fill:#552222;}#mermaid-svg-utrongvlch8I48n8 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-utrongvlch8I48n8 .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-utrongvlch8I48n8 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-utrongvlch8I48n8 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-utrongvlch8I48n8 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-utrongvlch8I48n8 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-utrongvlch8I48n8 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-utrongvlch8I48n8 .marker.cross{stroke:#333333;}#mermaid-svg-utrongvlch8I48n8 svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-utrongvlch8I48n8 g.classGroup text{fill:#9370DB;fill:#131300;stroke:none;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:10px;}#mermaid-svg-utrongvlch8I48n8 g.classGroup text .title{font-weight:bolder;}#mermaid-svg-utrongvlch8I48n8 .nodeLabel,#mermaid-svg-utrongvlch8I48n8 .edgeLabel{color:#131300;}#mermaid-svg-utrongvlch8I48n8 .edgeLabel .label rect{fill:#ECECFF;}#mermaid-svg-utrongvlch8I48n8 .label text{fill:#131300;}#mermaid-svg-utrongvlch8I48n8 .edgeLabel .label span{background:#ECECFF;}#mermaid-svg-utrongvlch8I48n8 .classTitle{font-weight:bolder;}#mermaid-svg-utrongvlch8I48n8 .node rect,#mermaid-svg-utrongvlch8I48n8 .node circle,#mermaid-svg-utrongvlch8I48n8 .node ellipse,#mermaid-svg-utrongvlch8I48n8 .node polygon,#mermaid-svg-utrongvlch8I48n8 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-utrongvlch8I48n8 .divider{stroke:#9370DB;stroke:1;}#mermaid-svg-utrongvlch8I48n8 g.clickable{cursor:pointer;}#mermaid-svg-utrongvlch8I48n8 g.classGroup rect{fill:#ECECFF;stroke:#9370DB;}#mermaid-svg-utrongvlch8I48n8 g.classGroup line{stroke:#9370DB;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 .classLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5;}#mermaid-svg-utrongvlch8I48n8 .classLabel .label{fill:#9370DB;font-size:10px;}#mermaid-svg-utrongvlch8I48n8 .relation{stroke:#333333;stroke-width:1;fill:none;}#mermaid-svg-utrongvlch8I48n8 .dashed-line{stroke-dasharray:3;}#mermaid-svg-utrongvlch8I48n8 #compositionStart,#mermaid-svg-utrongvlch8I48n8 .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #compositionEnd,#mermaid-svg-utrongvlch8I48n8 .composition{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #dependencyStart,#mermaid-svg-utrongvlch8I48n8 .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #dependencyStart,#mermaid-svg-utrongvlch8I48n8 .dependency{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #extensionStart,#mermaid-svg-utrongvlch8I48n8 .extension{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #extensionEnd,#mermaid-svg-utrongvlch8I48n8 .extension{fill:#333333!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #aggregationStart,#mermaid-svg-utrongvlch8I48n8 .aggregation{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 #aggregationEnd,#mermaid-svg-utrongvlch8I48n8 .aggregation{fill:#ECECFF!important;stroke:#333333!important;stroke-width:1;}#mermaid-svg-utrongvlch8I48n8 .edgeTerminals{font-size:11px;}#mermaid-svg-utrongvlch8I48n8 :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;} «interface» Message «interface» RequestMessage «interface» ResponseMessage «abstract» AbstractMessage «abstract» AbstractResponseMessage MessageHandler abstract void handle(T message) ChunkFetchRequest OneWayMessage RpcRequest StreamRequest ChunkFetchFailure RpcFailure StreamFailure ChunkFetchSuccess RpcResponse StreamResponse 根据上面的类图可以看出主要分类 AbstractMessage抽象类用于在单独的缓冲区中保存正文。其他消息类基本都继承该类。 RequestMessage定义了从客户端到服务端的消息接口 ChunkFetchRequest请求获取数据流中单个数据块的序列。这将对应一个响应信息成功或失败。RpcRequest由远程服务端 org.apache.spark.network.server.RpcHandler 处理的通用 RPC。这将对应一个响应信息成功或失败。OneWayMessage由远程服务端 org.apache.spark.network.server.RpcHandler 处理。不需要进行回复客户端。StreamRequest请求从远端流式传输数据。数据流 ID 是一个任意字符串需要两个端点协商后才能流式传输数据 ResponseMessage定义了从服务端到客户端的消息接口 AbstractResponseMessage响应信息的抽象类。 ChunkFetchSuccess处理ChunkFetchRequest成功后返回的消息。RpcResponse处理RpcRequest成功后返回的消息。StreamResponse处理StreamRequest成功后返回的消息。 ChunkFetchFailure处理ChunkFetchRequest失败后返回的消息。RpcFailure处理RpcRequest失败后返回的消息。StreamFailure处理StreamRequest失败后返回的消息。
client端请求和响应的流程 #mermaid-svg-tgJqsrIdtskNjh35 {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-tgJqsrIdtskNjh35 .error-icon{fill:#552222;}#mermaid-svg-tgJqsrIdtskNjh35 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-tgJqsrIdtskNjh35 .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-tgJqsrIdtskNjh35 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-tgJqsrIdtskNjh35 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-tgJqsrIdtskNjh35 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-tgJqsrIdtskNjh35 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-tgJqsrIdtskNjh35 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-tgJqsrIdtskNjh35 .marker.cross{stroke:#333333;}#mermaid-svg-tgJqsrIdtskNjh35 svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-tgJqsrIdtskNjh35 .label{font-family:"trebuchet ms",verdana,arial,sans-serif;color:#333;}#mermaid-svg-tgJqsrIdtskNjh35 .cluster-label text{fill:#333;}#mermaid-svg-tgJqsrIdtskNjh35 .cluster-label span{color:#333;}#mermaid-svg-tgJqsrIdtskNjh35 .label text,#mermaid-svg-tgJqsrIdtskNjh35 span{fill:#333;color:#333;}#mermaid-svg-tgJqsrIdtskNjh35 .node rect,#mermaid-svg-tgJqsrIdtskNjh35 .node circle,#mermaid-svg-tgJqsrIdtskNjh35 .node ellipse,#mermaid-svg-tgJqsrIdtskNjh35 .node polygon,#mermaid-svg-tgJqsrIdtskNjh35 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-tgJqsrIdtskNjh35 .node .label{text-align:center;}#mermaid-svg-tgJqsrIdtskNjh35 .node.clickable{cursor:pointer;}#mermaid-svg-tgJqsrIdtskNjh35 .arrowheadPath{fill:#333333;}#mermaid-svg-tgJqsrIdtskNjh35 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-tgJqsrIdtskNjh35 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-tgJqsrIdtskNjh35 .edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-tgJqsrIdtskNjh35 .edgeLabel rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-tgJqsrIdtskNjh35 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-tgJqsrIdtskNjh35 .cluster text{fill:#333;}#mermaid-svg-tgJqsrIdtskNjh35 .cluster span{color:#333;}#mermaid-svg-tgJqsrIdtskNjh35 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-tgJqsrIdtskNjh35 :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;} 传输层 1.addRpcRequest或addFetchRequest 2.WriteAndFlush IdleStateHandler MessageDecoder TransportFrameDecoder TransportResponseHandler TransportChannelHandler Netty MessageEncoder TransportClient Request Response server端处理请求和响应的流程 #mermaid-svg-f1APuzfNj2LgPlLA {font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;fill:#333;}#mermaid-svg-f1APuzfNj2LgPlLA .error-icon{fill:#552222;}#mermaid-svg-f1APuzfNj2LgPlLA .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-f1APuzfNj2LgPlLA .edge-thickness-normal{stroke-width:2px;}#mermaid-svg-f1APuzfNj2LgPlLA .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-f1APuzfNj2LgPlLA .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-f1APuzfNj2LgPlLA .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-f1APuzfNj2LgPlLA .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-f1APuzfNj2LgPlLA .marker{fill:#333333;stroke:#333333;}#mermaid-svg-f1APuzfNj2LgPlLA .marker.cross{stroke:#333333;}#mermaid-svg-f1APuzfNj2LgPlLA svg{font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-f1APuzfNj2LgPlLA .label{font-family:"trebuchet ms",verdana,arial,sans-serif;color:#333;}#mermaid-svg-f1APuzfNj2LgPlLA .cluster-label text{fill:#333;}#mermaid-svg-f1APuzfNj2LgPlLA .cluster-label span{color:#333;}#mermaid-svg-f1APuzfNj2LgPlLA .label text,#mermaid-svg-f1APuzfNj2LgPlLA span{fill:#333;color:#333;}#mermaid-svg-f1APuzfNj2LgPlLA .node rect,#mermaid-svg-f1APuzfNj2LgPlLA .node circle,#mermaid-svg-f1APuzfNj2LgPlLA .node ellipse,#mermaid-svg-f1APuzfNj2LgPlLA .node polygon,#mermaid-svg-f1APuzfNj2LgPlLA .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-f1APuzfNj2LgPlLA .node .label{text-align:center;}#mermaid-svg-f1APuzfNj2LgPlLA .node.clickable{cursor:pointer;}#mermaid-svg-f1APuzfNj2LgPlLA .arrowheadPath{fill:#333333;}#mermaid-svg-f1APuzfNj2LgPlLA .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-f1APuzfNj2LgPlLA .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-f1APuzfNj2LgPlLA .edgeLabel{background-color:#e8e8e8;text-align:center;}#mermaid-svg-f1APuzfNj2LgPlLA .edgeLabel rect{opacity:0.5;background-color:#e8e8e8;fill:#e8e8e8;}#mermaid-svg-f1APuzfNj2LgPlLA .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-f1APuzfNj2LgPlLA .cluster text{fill:#333;}#mermaid-svg-f1APuzfNj2LgPlLA .cluster span{color:#333;}#mermaid-svg-f1APuzfNj2LgPlLA div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:"trebuchet ms",verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-f1APuzfNj2LgPlLA :root{--mermaid-font-family:"trebuchet ms",verdana,arial,sans-serif;} 传输层 IdleStateHandler MessageDecoder TransportFrameDecoder TransportRequestHandler TransportChannelHandler StreamManager RpcHandler Netty MessageEncoder Request Response
