文章目录1. 集群内部通讯方法概述2. TaskManager向ResourceManager注册RPC服务3. JobMaster向ResourceManager申请Slot计算资源现在我们已经知道Flink中RPC通信框架的底层设计与实现接下来通过具体的实例了解集群运行时中组件如何基于RPC通信框架构建相互之间的调用关系。1. 集群内部通讯方法概述通过RegisteredRpcConnection进行连接注册与通讯维护心跳等当TaskExecutor启动后会立即向ResourceManager中注册当前TaskManager的信息。同样JobMaster组件启动后也立即会向ResourceManager注册JobMaster的信息。这些注册操作实际上就是在构建集群中各个组件之间的RPC连接这里的注册连接在Flink中被称为RegisteredRpcConnection集群组件之间的RPC通信都会通过创建RegisteredRpcConnection进行例如获取RpcEndpoint对应的RpcGateway接口以及维护组件之间的心跳连接等。如下图集群运行时中各组件的注册连接主要如下三种RegisteredRpcConnection的实现。JobManagerRegisteredRpcConnection用于管理TaskManager中与JobManager之间的RPC连接。ResourceManagerConnection用于管理JobManager中与ResourceManager之间的RPC连接。TaskExecutorToResourceManagerConnection用于管理TaskExecutor中与ResourceManager之间的RPC连接。如下图再有RegisteredRpcConnection提供了generateRegistration()抽象方法主要用于生成组件之间的RPC连接每次调用RegisteredRpcConnection.start()方法启动RegisteredRpcConnection时都会创建新的RetryingRegistration。不同RegisteredRpcConnection创建的RetryingRegistration也会有所不同例如在TaskExecutorToResourceManagerConnection中就会创建ResourceManagerRegistration实例。调用rpcService.connect(targetAddress, targetType) 返回RpcGateway的代理对象通过RpcGateway连接到目标RpcEndpoint上。在RetryingRegistration中会提供invokeRegistration()抽象方法用于实现子类的RPC注册操作。例如在ResourceManagerRegistration中会实现invokeRegistration()方法在方法中调用resourceManager.registerTaskExecutor()将TaskExecutor信息注册到ResourceManager中这里的ResourceManager就是ResourceManagerGateway接口代理类。调用onRegistrationSuccess()方法继续后续操作例如在JobManagerRegisteredRpcConnection中会向jobLeaderListener添加当前的jobId等信息。如果当前组件没有成功到注册至目标组件时会调用onRegistrationFailure()抽象方法继续后续操作包括连接重连或停止整个RpcEndpoint对应的服务。接着以TaskManager向ResourceManager注册RPC服务为例介绍整个RPC连接的注册过程。2. TaskManager向ResourceManager注册RPC服务TaskManager向ResourceManager注册RPC服务的过程如图所示。TaskExecutor节点正常启动后调用RpcEndpoint.onStart()方法初始化并启动TaskExecutor组件的内部服务。创建监听服务TaskExecutor调用resourceManagerLeaderRetriever.start()方法启动ResourceManager组件领导节点的监听服务并传入ResourceManagerLeaderListener用于监听ResourceManager的领导节点的变化情况。当ResourceManagerLeaderListener接收到来自ResourceManager的leaderAddress以及leaderSessionID的信息后调用notifyOfNewResourceManagerLeader()方法通知TaskExecutor和新的ResourceManagerLeader建立RPC连接。创建与ResourceManager组件的RPC网络连接a. 调用TaskExecutor.reconnectToResourceManager()内部方法创建与ResourceManager组件之间的RPC网络连接。b. 在reconnectToResourceManager()方法中会事先调用closeResourceManagerConnection()方法关闭之前的ResourceManager连接然后依次调用tryConnectToResourceManager()和connectToResourceManager()方法创建与ResourceManager节点的RPC连接。创建TaskExecutorRegistration对象在connectToResourceManager()方法中会创建TaskExecutorRegistration对象用于存储TaskManager的注册信息其中包括taskExecutorAddress、resourceId以及dataPort等连接信息同时还包含hardwareDescription、defaultSlotResourceProfile以及totalResourceProfile等资源描述信息。正式建立网络连接创建TaskExecutorToResourceManagerConnection实例正式与ResourceManager建立RPC网络连接同时调用TaskExecutorToResourceManagerConnection.start()方法启动RPC连接。实际上调用的是RegisteredRpcConnection.start()方法。创建新的创建新的Registration与其他组件的RPC连接在RegisteredRpcConnection中会调用内部方法createNewRegistration()创建新的Registration。而在createNewRegistration()方法中会调用generateRegistration()子类方法创建与其他组件之间的RPC连接。这里主要调用的是TaskExecutorToResourceManagerConnection.generateRegistration()方法。调用RetryingRegistration.startRegistration()方法注册具体的RPC连接实际上会调用AkkaRpcService.connect()方法创建并获取ResourceManager对应的RpcGateway接口。调用ResourceManagerGateway.registerTaskExecutor()方法最终完成在ResourceManager中注册TaskManager的操作。创建的TaskExecutorRegistration同时会传递给ResourceManager。当ResourceManager接收到TaskManager的注册信息后会在本地维护TaskManager的注册信息并建立与TaskManager组件之间的心跳连接至此完成了TaskManager启动后向ResourceManager进行RPC网络连接注册的全部流程。如代码所示TaskExecutor.connectToResourceManager()方法中首先会创建TaskExecutorRegistration注册信息和TaskExecutorToResourceManagerConnection对象。然后调用TaskExecutorToResourceManagerConnection.start()方法启动TaskManager和ResourceManager之间的RPC注册连接。privatevoidconnectToResourceManager(){assert(resourceManagerAddress!null);assert(establishedResourceManagerConnectionnull);assert(resourceManagerConnectionnull);log.info(Connecting to ResourceManager {}.,resourceManagerAddress);// TaskExecutor注册信息finalTaskExecutorRegistrationtaskExecutorRegistrationnewTaskExecutorRegistration(getAddress(),getResourceID(),taskManagerLocation.dataPort(),hardwareDescription,taskManagerConfiguration.getDefaultSlotResourceProfile(),taskManagerConfiguration.getTotalResourceProfile());resourceManagerConnectionnewTaskExecutorToResourceManagerConnection(log,getRpcService(),taskManagerConfiguration.getRetryingRegistrationConfiguration(),resourceManagerAddress.getAddress(),resourceManagerAddress.getResourceManagerId(),getMainThreadExecutor(),newResourceManagerRegistrationListener(),taskExecutorRegistration);resourceManagerConnection.start();}接着看RegisteredRpcConnection.start()的代码逻辑如代码所示。publicvoidstart(){checkState(!closed,The RPC connection is already closed);checkState(!isConnected()pendingRegistrationnull,The RPC connection is already started);// 创建RetryingRegistrationfinalRetryingRegistrationF,G,SnewRegistrationcreateNewRegistration();// 启动RetryingRegistrationif(REGISTRATION_UPDATER.compareAndSet(this,null,newRegistration)){newRegistration.startRegistration();}else{// 并行启动后直接取消当前RegistrationnewRegistration.cancel();}}关注RetryingRegistration.startRegistration()逻辑。根据不同的targetType选择创建FencedRpcGateway还是普通的RpcGateway。创建RpcGateway代理类后就可以连接到指定的RpcEndpoint了。对于rpcService.connect()方法的定义我们已经在RPC底层原理中介绍过。创建RPC连接后尝试注册操作。如果注册失败则进行Retry操作除非接收到取消操作的消息。publicvoidstartRegistration(){if(canceled){return;}try{finalCompletableFutureGrpcGatewayFuture;// 根据不同的targetType选择创建FencedRpcGateway还是普通的RpcGatewayif(FencedRpcGateway.class.isAssignableFrom(targetType)){rpcGatewayFuture(CompletableFutureG)rpcService.connect(targetAddress,fencingToken,targetType.asSubclass(FencedRpcGateway.class));}else{rpcGatewayFuturerpcService.connect(targetAddress,targetType);}// 成功获取网关后尝试注册操作CompletableFutureVoidrpcGatewayAcceptFuturerpcGatewayFuture.thenAcceptAsync((GrpcGateway)-{log.info(Resolved {} address, beginning registration,targetName);register(rpcGateway,1,retryingRegistrationConfiguration.getInitialRegistrationTimeoutMillis());},rpcService.getExecutor());// 如果注册失败则进行Retry操作除非取消操作rpcGatewayAcceptFuture.whenCompleteAsync((Voidv,Throwablefailure)-{if(failure!null!canceled){finalThrowablestrippedFailureExceptionUtils.stripCompletionException(failure);// 间隔指定时间后再次注册startRegistrationLater(retryingRegistrationConfiguration.getErrorDelayMillis());}},rpcService.getExecutor());}catch(Throwablet){completionFuture.completeExceptionally(t);cancel();}}继续了解ResourceManagerRegistration.invokeRegistration()的具体实现。该方法会创建和ResourceManagerGateway之间的连接以及注册操作resourceManager会接收来自TaskExecutor的注册信息并根据taskExecutorRegistration提供的注册信息将TaskExecutor信息记录在ResourceManager的本地存储中然后开启TaskExecutor之间的心跳连接。至此TaskManager能和ResourceManager进行正常的RPC通信了。protectedCompletableFutureRegistrationResponseinvokeRegistration(ResourceManagerGatewayresourceManager,ResourceManagerIdfencingToken,longtimeoutMillis)throwsException{TimetimeoutTime.milliseconds(timeoutMillis);returnresourceManager.registerTaskExecutor(taskExecutorRegistration,timeout);}对于其他组件之间的RpcConnection注册操作例如TaskManager与JobMaster之间的RPC连接注册基本上和ResourceManagerRegistration一样这里暂不介绍。接下来我们看JobMaster是如何向ResourceManager申请Slot计算资源的。3. JobMaster向ResourceManager申请Slot计算资源当JobMaster组件启动后会调用JobMaster.startJobMasterServices()启动JobMaster中的内部服务其中就包括了SlotPool组件。同时会创建和启动JobMaster与ResourceManager之间的RPC连接ResourceManagerConnection。创建成功后会执行包括向ResourceManager发送申请Slot计算资源的RPC请求等后续操作。如代码所示//从SlotPoolImpl.connectToResourceManager()可以看出方法中分别遍历//waitingForResourceManager集合中的PendingRequest//然后就每个PendingRequest调用requestSlotFromResourceManager()方法向//ResourceManager申请PendingRequest中指定的Slot计算资源。publicvoidconnectToResourceManager(NonnullResourceManagerGatewayresourceManagerGateway){this.resourceManagerGatewaycheckNotNull(resourceManagerGateway);for(PendingRequestpendingRequest:waitingForResourceManager.values()){requestSlotFromResourceManager(resourceManagerGateway,pendingRequest);}waitingForResourceManager.clear();}继续看SlotPoolImpl.requestSlotFromResourceManager()方法的实现如下代码所示。创建AllocationID并将pendingRequest和AllocationID存储在pendingRequests集合中。判断pendingRequest是否出现异常或已经分配了其他AllocationID如果出现异常或已分配则取消当前pendingRequest中的资源分配请求。调用resourceManagerGateway.requestSlot()远程RPC方法向ResourceManager申请Slot计算资源此时会在方法中创建SlotRequest对象指定申请Slot计算资源的具体参数。ResourceManager接收到SlotPool发送的SlotRequest请求后会将SlotRequest转发给SlotManager进行处理此时如果能正常分配到Slot资源则会返回Acknowledge信息。调用FutureUtils.whenCompleteAsyncIfNotDone()方法执行返回rmResponse CompletableFuture的对象此时如果Slot资源申请过程出现异常则调用slotRequestToResourceManager-Failed()方法进行处理。privatevoidrequestSlotFromResourceManager(finalResourceManagerGatewayresourceManagerGateway,finalPendingRequestpendingRequest){checkNotNull(resourceManagerGateway);checkNotNull(pendingRequest);log.info(Requesting new slot [{}] and profile {} from resource manager.,pendingRequest.getSlotRequestId(),pendingRequest.getResourceProfile());finalAllocationIDallocationIdnewAllocationID();pendingRequests.put(pendingRequest.getSlotRequestId(),allocationId,pendingRequest);pendingRequest.getAllocatedSlotFuture().whenComplete((AllocatedSlotallocatedSlot,Throwablethrowable)-{if(throwable!null||!allocationId.equals(allocatedSlot.getAllocationId())){resourceManagerGateway.cancelSlotRequest(allocationId);}});CompletableFutureAcknowledgermResponseresourceManagerGateway.requestSlot(jobMasterId,newSlotRequest(jobId,allocationId,pendingRequest.getResourceProfile(),jobManagerAddress),rpcTimeout);FutureUtils.whenCompleteAsyncIfNotDone(rmResponse,componentMainThreadExecutor,(Acknowledgeignored,Throwablefailure)-{if(failure!null){slotRequestToResourceManagerFailed(pendingRequest.getSlotRequestId(),failure);}});}从以上实例可以看出集群运行时中各个组件之间都是基于RPC通信框架相互访问的。RpcEndpoint组件会创建与其他RpcEndpoint之间的RegisteredRpcConnection并通过RpcGateway接口的动态代理类与其他组件进行通信。需要注意的是Flink把Akka作为RPC底层的通信框架但没有使用Akka其他丰富的监督功能并且未来有去掉Akka依赖的可能。参考《Flink设计与实现核心原理与源码解析》–张利兵
