零. RpcService服务概述

RpcService负责创建和启动Flink集群环境中RpcEndpoint组件的RpcServer，且RpcService在启动集群时会提前创建好。AkkaRpcService作为RpcService的唯一实现类，基于Akka的ActorSystem进行封装，为不同的RpcEndpoint创建相应的ActorRef实例。

 

RpcService主要包含如下两个重要方法。

1. startServer()：用于启动RpcEndpoint中的RpcServer。RpcServer实际上就是对Actor进行封装，启动完成后，RpcEndpoint中的RpcServer就能够对外提供服务了。
2. connect()：用于连接远端RpcEndpoint并返回给调用方RpcGateway接口的方法，建立连接后RPC客户端就能像本地一样调用RpcServer提供的RpcGateway接口了。

例如在JobMaster组件中创建与ResourceManager组件之间的RPC连接时。此时可以通过Akka发送消息到ResourceManager的RpcServer中，这样就使得JobMaster像调用本地方法一样在ResourceManager中执行请求任务。

 

1. AkkaRpcService的创建和初始化

在创建和启动ClusterEntrypoint及TaskManagerRunner的过程中，会调用AkkaRpcServiceUtils.createRpcService()方法创建默认的AkkaRpcService，接着启动RpcServer。

例如管理节点中会使用AkkaRpcService实例创建并启动ResourceManager、Dispatcher以及JobMaster等RPC服务。

创建AkkaRpcService主要包括如下步骤。

1. 在ClusterEntrypoint中创建RpcService。
2. 启动ActorSystem服务。
3. 创建RobustActorSystem。RobustActorSystem实际上是对Akka的ActorSystem进行了封装和拓展，相比于原生Akka
   ActorSystem，RobustActorSystem包含了UncaughtExceptionHandler组件，能够对ActorSystem抛出的异常进行处理。
4. 使用RobustActorSystem创建AkkaRpcService实例。
5. 将AkkaRpcService返回到ClusterEntrypoint中，用于启动集群中各个RpcEndpoint组件服务

 

2.通过AkkaRpcService初始化RpcServer

在集群运行时中创建了共用的AkkaRpcService服务，相当于创建了Akka系统中的ActorSystem，接下来就是使用AkkaRpcService启动各个RpcEndpoint中的RpcServer实例。（AkkaRpcService服务作为共用的rpc服务，启动其他各个组件的RpcServer实例？）

 
这里先看通过AkkaRpcService初始化RpcEndpoint对应的RpcServer的逻辑。如下在org.apache.flink.runtime.rpc.RpcEndpoint的构造器中，执行了RpcServer的初始化

protected RpcEndpoint(final RpcService rpcService, final String endpointId) {this.rpcService = checkNotNull(rpcService, "rpcService");this.endpointId = checkNotNull(endpointId, "endpointId");// 初始化RpcEndpoint中的RpcServerthis.rpcServer = rpcService.startServer(this);this.mainThreadExecutor = new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
}

具体看下rpcService.startServer(this) 启动rpcServer的逻辑

1. ActorSystem创建相应Actor的ActorRef引用类。创建完毕后会将RpcEndpoint和ActorRef信息存储在Actor键值对集合中。
2. 启动RpcEndpoint对应的RPC服务，首先获取当前RpcEndpoint实现的RpcGateways接口。 RpcGateway接口最终通过RpcUtils.extractImplementedRpcGateways()方法从类定义抽取出来，例如JobMaster组件会抽取JobMasterGateway接口定义。
3. 创建InvocationHandler代理类，根据InvocationHandler代理类提供的invoke()方法实现被代理类的具体方法。
4. 根据RpcEndpoint是否为FencedRpcEndpoint，InvocationHandler分为FencedAkkaInvocationHandler和AkkaInvocationHandler两种类型。

FencedMainThreadExecutable代理的接口主要有FencedMainThreadExecutable和FencedRpcGateway两种。
AkkaInvocationHandler主要代理实现AkkaBasedEndpoint、RpcGateway、StartStoppable、MainThreadExecutable、RpcServer等接口。

5. 创建好InvocationHandler代理类后，通过反射的方式（Proxy.newProxyInstance()）创建代理类。创建的代理类会被转换为RpcServer实例，再返回给RpcEndpoint使用。

在RpcServer创建的过程中可以看出，实际上包含了创建RpcEndpoint中的Actor引用类ActorRef和AkkaInvocationHandler动态代理类。最后将动态代理类转换为RpcServer接口返回给RpcEndpoint实现类，此时实现的组件就能够获取到RpcServer服务，且通过RpcServer代理了所有的RpcGateways接口，提供了本地方法调用和远程方法调用两种模式。

@Override  
public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {  checkNotNull(rpcEndpoint, "rpc endpoint");  final SupervisorActor.ActorRegistration actorRegistration =  registerAkkaRpcActor(rpcEndpoint);  final ActorRef actorRef = actorRegistration.getActorRef();  final CompletableFuture<Void> actorTerminationFuture =  actorRegistration.getTerminationFuture();  //启动RpcEndpoint对应的RPC服务LOG.info(  "Starting RPC endpoint for {} at {} .",  rpcEndpoint.getClass().getName(),  actorRef.path());  final String akkaAddress = AkkaUtils.getAkkaURL(actorSystem, actorRef);  final String hostname;  Option<String> host = actorRef.path().address().host();  if (host.isEmpty()) {  hostname = "localhost";  } else {  hostname = host.get();  }  //解析EpcEndpoint实现的所有RpcGateway接口Set<Class<?>> implementedRpcGateways =  new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass()));  //额外添加RpcServer和AkkaBasedEndpoint类implementedRpcGateways.add(RpcServer.class);  implementedRpcGateways.add(AkkaBasedEndpoint.class);  final InvocationHandler akkaInvocationHandler;  //根据是否是FencedRpcEndpoint创建不同的动态代理对象if (rpcEndpoint instanceof FencedRpcEndpoint) {  // a FencedRpcEndpoint needs a FencedAkkaInvocationHandler  akkaInvocationHandler =  new FencedAkkaInvocationHandler<>(  akkaAddress,  hostname,  actorRef,  configuration.getTimeout(),  configuration.getMaximumFramesize(),  actorTerminationFuture,  ((FencedRpcEndpoint<?>) rpcEndpoint)::getFencingToken,  captureAskCallstacks);  implementedRpcGateways.add(FencedMainThreadExecutable.class);  } else {  akkaInvocationHandler =  new AkkaInvocationHandler(  akkaAddress,  hostname,  actorRef,  configuration.getTimeout(),  configuration.getMaximumFramesize(),  actorTerminationFuture,  captureAskCallstacks);  }  // Rather than using the System ClassLoader directly, we derive the ClassLoader  // from this class . That works better in cases where Flink runs embedded and all Flink    // code is loaded dynamically (for example from an OSGI bundle) through a custom ClassLoader    ClassLoader classLoader = getClass().getClassLoader();  @SuppressWarnings("unchecked")  RpcServer server =  (RpcServer)  Proxy.newProxyInstance(  classLoader,  implementedRpcGateways.toArray(  new Class<?>[implementedRpcGateways.size()]),  akkaInvocationHandler);  return server;  
}

 

3. ResourceManager中RPC服务的启动

RpcServer在RpcEndpoint的构造器中完成初始化后，接下来就是启动RpcEndpoint和RpcServer，这里以ResourceManager为例进行说明。

在启动集群时，看下如何启动ResourceManager的RPC服务的。如下调用链

ClusterEntrypoint.startCluster->runCluster
->dispatcherResourceManagerComponentFactory.create
->resourceManager.start();
=>
public final void start() {  rpcServer.start();  
}

继续探索RPC服务是如何启动的

首先在DefaultDispatcherResourceManagerComponentFactory中调用ResourceManager.start()方法启动ResourceManager实例，此时在ResourceManager.start()方法中会同步调用RpcServer.start()方法，启动ResourceManager所在RpcEndpoint中的RpcServer，如下。

1. 调用ResourceManager.start()方法，此时会调用RpcEndpoint.start()父方法，启动ResourceManager组件的RpcServer。
2. 通过动态代理AkkaInvocationHandler.invoke()方法执行流程，发现调用的是StartStoppable.start()方法，此时会直接调用AkkaInvocationHandler.start()本地方法。
3. 在AkkaInvocationHandler.start()方法中，实际上会调用rpcEndpoint.tell(ControlMessages.START,ActorRef.noSender())方法向ResourceManager对应的Actor发送控制消息，表明当前Actor实例可以正常启动并接收来自远端的RPC请求。
4. AkkaRpcActor调用handleControlMessage()方法处理ControlMessages.START控制消息。
5. 将AkkaRpcActor中的状态更新为StartedState，此时ResourceManager的RpcServer启动完成，ResourceManager组件能够接收来自其他组件的RPC请求。

在flink1.12中省略了AkkaInvocationHandler的干预。

经过以上步骤，指定组件的RpcEndpoint节点就正常启动，此时RpcServer会作为独立的线程运行在JobManager或TaskManager进程中，处理本地和远程提交的RPC请求。

 

4. 实现相互通讯能力

当AkkaRpcService启动RpcEndpoint中的RpcServer后，RpcEndpoint组件仅能对外提供处理RPC请求的能力，RpcEndpoint组件需要在启动后向其他组件注册自己的RpcEndpoint信息，并完成组件之间的RpcConnection注册，才能相互访问和通信。而创建RPC连接需要调用RpcService.connect()方法。

如代码所示，在AkkaRpcService.connect()方法中，完成了RpcConnection对象的创建。

@Override  
public <C extends RpcGateway> CompletableFuture<C> connect(  final String address, final Class<C> clazz) {  return connectInternal(  address,  clazz,  (ActorRef actorRef) -> {  Tuple2<String, String> addressHostname = extractAddressHostname(actorRef);  return new AkkaInvocationHandler(  addressHostname.f0,  addressHostname.f1,  actorRef,  configuration.getTimeout(),  configuration.getMaximumFramesize(),  null,  captureAskCallstacks);  });  
}

具体看AkkaRpcService.connectInternal()方法逻辑。

1. 获取ActorRef引用对象。
2. 调用Patterns.ask()方法，向actorRef对应的RpcEndpoint节点发送RemoteHandshakeMessage消息，确保连接的RpcEndpoint节点正常，如果成功，则RpcEndpoint会返回HandshakeSuccessMessage消息。
3. 调用invocationHandlerFactory创建invocationHandler动态代理类，此时可以看到传递的接口列表为new Class<?>[]{clazz}，也就是当前RpcEndpoint需要访问的RpcGateway接口。例如JobMaster访问ResourceManager时，这里就是ResourceManagerGateway接口。

private <C extends RpcGateway> CompletableFuture<C> connectInternal(  final String address,  final Class<C> clazz,  Function<ActorRef, InvocationHandler> invocationHandlerFactory) {  checkState(!stopped, "RpcService is stopped");  LOG.debug(  "Try to connect to remote RPC endpoint with address {}. Returning a {} gateway.",  address,  clazz.getName());  //获取actorRef实例  final CompletableFuture<ActorRef> actorRefFuture = resolveActorAddress(address);  //进行handshake操作，确保需要连接的RpcEndpoint节点正常  final CompletableFuture<HandshakeSuccessMessage> handshakeFuture =  actorRefFuture.thenCompose(  (ActorRef actorRef) ->  FutureUtils.toJava( //调用Patterns.ask()方法，向actorRef对应的//RpcEndpoint节点发送RemoteHandshakeMessage消息，//确保连接的RpcEndpoint节点正常，如果成功，则//RpcEndpoint会返回HandshakeSuccessMessage消息。 Patterns.ask(  actorRef,  new RemoteHandshakeMessage(  clazz, getVersion()),  configuration.getTimeout().toMilliseconds())  .<HandshakeSuccessMessage>mapTo(  ClassTag$.MODULE$  .<HandshakeSuccessMessage>apply(  HandshakeSuccessMessage  .class))));  //创建RPC动态代理类  return actorRefFuture.thenCombineAsync(  handshakeFuture,  (ActorRef actorRef, HandshakeSuccessMessage ignored) -> {  InvocationHandler invocationHandler = invocationHandlerFactory.apply(actorRef);  // Rather than using the System ClassLoader directly, we derive the ClassLoader  // from this class . That works better in cases where Flink runs embedded and                // all Flink                // code is loaded dynamically (for example from an OSGI bundle) through a custom                // ClassLoader                ClassLoader classLoader = getClass().getClassLoader();  @SuppressWarnings("unchecked")  C proxy =  (C)  Proxy.newProxyInstance(  classLoader, new Class<?>[] {clazz}, invocationHandler);  return proxy;  },  actorSystem.dispatcher());  
}

经过以上步骤，实现了创建RpcEndpoint组件之间的RPC连接，此时集群RPC组件之间可以进行相互访问，例如JobMaster可以向ResourceManager发送Slot资源请求。
RPC 服务启动的 Akka actor 能接收来自RpcGateway RPC 调用。

 

参考：《Flink设计与实现：核心原理与源码解析》–张利兵