【Flink集群RPC通讯机制（四）】集群组件（tm、jm与rm）之间的RPC通信

本文主要是介绍【Flink集群RPC通讯机制（四）】集群组件（tm、jm与rm）之间的RPC通信，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

文章目录

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注册连接。

private void connectToResourceManager() {assert(resourceManagerAddress != null);assert(establishedResourceManagerConnection == null);assert(resourceManagerConnection == null);log.info("Connecting to ResourceManager {}.", resourceManagerAddress);// TaskExecutor注册信息final TaskExecutorRegistration taskExecutorRegistration = new TaskExecutorRegistration(getAddress(),getResourceID(),taskManagerLocation.dataPort(),hardwareDescription,taskManagerConfiguration.getDefaultSlotResourceProfile(),taskManagerConfiguration.getTotalResourceProfile());resourceManagerConnection =new TaskExecutorToResourceManagerConnection(log,getRpcService(),taskManagerConfiguration.getRetryingRegistrationConfiguration(),resourceManagerAddress.getAddress(),resourceManagerAddress.getResourceManagerId(),getMainThreadExecutor(),new ResourceManagerRegistrationListener(),taskExecutorRegistration);resourceManagerConnection.start();
}

接着看RegisteredRpcConnection.start()的代码逻辑，如代码所示。

public void start() {checkState(!closed, "The RPC connection is already closed");checkState(!isConnected() && pendingRegistration == null, "The RPC connection is already started");// 创建RetryingRegistrationfinal RetryingRegistration<F, G, S> newRegistration = createNewRegistration();// 启动RetryingRegistrationif (REGISTRATION_UPDATER.compareAndSet(this, null, newRegistration)) {newRegistration.startRegistration();} else {// 并行启动后，直接取消当前RegistrationnewRegistration.cancel();}
}

关注：RetryingRegistration.startRegistration()逻辑。

根据不同的targetType，选择创建FencedRpcGateway还是普通的RpcGateway。
创建RpcGateway代理类后，就可以连接到指定的RpcEndpoint了。对于rpcService.connect()方法的定义，我们已经在RPC底层原理中介绍过。
创建RPC连接后，尝试注册操作。
如果注册失败，则进行Retry操作，除非接收到取消操作的消息。

public void startRegistration() {if (canceled) {return;}try {final CompletableFuture<G> rpcGatewayFuture;// 根据不同的targetType，选择创建FencedRpcGateway还是普通的RpcGatewayif (FencedRpcGateway.class.isAssignableFrom(targetType)) {rpcGatewayFuture = (CompletableFuture<G>) rpcService.connect(targetAddress,fencingToken,targetType.asSubclass(FencedRpcGateway.class));} else {rpcGatewayFuture = rpcService.connect(targetAddress, targetType);}// 成功获取网关后，尝试注册操作CompletableFuture<Void> rpcGatewayAcceptFuture = rpcGatewayFuture.thenAcceptAsync((G rpcGateway) -> {log.info("Resolved {} address, beginning registration", targetName);register(rpcGateway, 1, retryingRegistrationConfiguration.getInitialRegistrationTimeoutMillis());},rpcService.getExecutor());// 如果注册失败，则进行Retry操作，除非取消操作rpcGatewayAcceptFuture.whenCompleteAsync((Void v, Throwable failure) -> {if (failure != null && !canceled) {final Throwable strippedFailure =ExceptionUtils.stripCompletionException(failure);// 间隔指定时间后再次注册startRegistrationLater(retryingRegistrationConfiguration.getErrorDelayMillis());}},rpcService.getExecutor());}catch (Throwable t) {completionFuture.completeExceptionally(t);cancel();}}

继续了解ResourceManagerRegistration.invokeRegistration()的具体实现。

该方法会创建和ResourceManagerGateway之间的连接以及注册操作，
resourceManager会接收来自TaskExecutor的注册信息，并根据taskExecutorRegistration提供的注册信息，将TaskExecutor信息记录在ResourceManager的本地存储中，然后开启TaskExecutor之间的心跳连接。至此，TaskManager能和ResourceManager进行正常的RPC通信了。

protected CompletableFuture<RegistrationResponse> invokeRegistration(ResourceManagerGateway resourceManager, ResourceManagerId fencingToken, long timeoutMillis) throws Exception {Time timeout = Time.milliseconds(timeoutMillis);return resourceManager.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计算资源。
public void connectToResourceManager(@Nonnull ResourceManagerGateway resourceManagerGateway) {this.resourceManagerGateway = checkNotNull(resourceManagerGateway);for (PendingRequest pendingRequest : 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()方法进行处理。

private void requestSlotFromResourceManager(final ResourceManagerGateway resourceManagerGateway,final PendingRequest pendingRequest) {checkNotNull(resourceManagerGateway);checkNotNull(pendingRequest);log.info("Requesting new slot [{}] and profile {} from resource manager.", pendingRequest.getSlotRequestId(), pendingRequest.getResourceProfile());final AllocationID allocationId = new AllocationID();pendingRequests.put(pendingRequest.getSlotRequestId(), allocationId,pendingRequest);pendingRequest.getAllocatedSlotFuture().whenComplete((AllocatedSlot allocatedSlot, Throwable throwable) -> {if (throwable != null || !allocationId.equals(allocatedSlot.getAllocationId())) {resourceManagerGateway.cancelSlotRequest(allocationId);}});CompletableFuture<Acknowledge> rmResponse = resourceManagerGateway.requestSlot(jobMasterId,new SlotRequest(jobId, allocationId, pendingRequest.getResourceProfile(), jobManagerAddress),rpcTimeout);FutureUtils.whenCompleteAsyncIfNotDone(rmResponse,componentMainThreadExecutor,(Acknowledge ignored, Throwable failure) -> {if (failure != null) {slotRequestToResourceManagerFailed(pendingRequest.getSlotRequestId(), failure);}});
}