Prerequisites
- Flink的RPC服务是基于Akka Remote实现的。一个简单的Akka Remoting ActorSystem的配置如下:
akka { actor { provider = remote } remote { enabled-transports = ["akka.remote.netty.tcp"] netty.tcp { hostname = "127.0.0.1" port = 2552 } } }- 从这份配置文件可以看出，要建立一个ActorSystem，首先需要提供ActorSystem运行的机器的地址和端口。
- 参考Akka Remoting的文档，获取远程节点的Actor有两条途径。
- 第一条，通过actorSelection(path)，在这儿需要知道远程节点的地址。获取到了ActorSelection就已经可以发送消息过去，也可以通过回信获取这个Actor的ActorRef。
- 第二条，通过配置，配置文件如下。通过这种方式，远程系统的daemon会被请求建立这个Actor，ActorRef可以直接通过system.actorOf(new Props(...)获取。
akka { actor { deployment { /sampleActor { remote = "akka.tcp://sampleActorSystem@127.0.0.1:2553" } } } }
- 第一条，通过actorSelection(path)，在这儿需要知道远程节点的地址。获取到了ActorSelection就已经可以发送消息过去，也可以通过回信获取这个Actor的ActorRef。
定义RPC协议
- RPC协议是客户端和服务端的通信接口。如下所示定义了一个BaseGateway的通信接口。
public interface BaseGateway extends RpcGateway { CompletableFuture<Integer> foobar(); }
- 在Flink中，RPC协议的定义通过实现RpcGateway.
public interface RpcGateway { /** * Returns the fully qualified address under which the associated rpc endpoint is reachable. * * @return Fully qualified (RPC) address under which the associated rpc endpoint is reachable */ String getAddress(); /** * Returns the fully qualified hostname under which the associated rpc endpoint is reachable. * * @return Fully qualified hostname under which the associated rpc endpoint is reachable */ String getHostname(); }
- 这个接口需要实现两个方法，分别是getAddress和getHostname。原因如下：
- 如上文所述，想要通过ActorSystem获取远程Actor，必须要有地址。而在Flink中，例如Yarn这种模式下，JobMaster会先建立ActorSystem，这时TaskExecutor的Container都还没有分配，自然无法在配置中指定远程Actor的地址，所以一个远程节点提供自己的地址是必须的。
实现RPC协议
- Flink的RPC协议一般定义为一个Java接口，服务端需要实现这个接口。如下是上面定义的BaseGateway的实现。
public static class BaseEndpoint extends RpcEndpoint implements BaseGateway { private final int foobarValue; protected BaseEndpoint(RpcService rpcService, int foobarValue) { super(rpcService); this.foobarValue = foobarValue; } @Override public CompletableFuture<Integer> foobar() { return CompletableFuture.completedFuture(foobarValue); } @Override public CompletableFuture<Void> postStop() { return CompletableFuture.completedFuture(null); } }
- RpcEndpoint是rpc请求的接收端的基类。RpcEndpoint是通过RpcService来启动的。
构造并启动RpcService
- RpcService会在每一个ClusterEntrypoint(JobMaster)和TaskManagerRunner(TaskExecutor)启动的过程中被初始化并启动。
- RpcService主要负责启动RpcEndpoint(也就是服务端)，连接到远程的RpcEndpoint并提供一个代理(也就是客户端)。
- 此外，为了防止状态的concurrent modification，RpcEndpoint上所有的Rpc调用都只会运行在主线程上，RpcService提供了运行在其它线程的方法。
构造并启动RpcEndpoint(服务端)
- 每一个RpcEndpoint在初始化阶段会通过该节点的RpcService的startServer方法来初始化服务。
- 在该方法中创建了一个Akka的Actor，这个Actor也是Rpc调用的实际接收者，Rpc的请求会在客户端被封装成RpcInvocation对象以Akka消息的形式发送。
@Override public <C extends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) { ... // 新建一个包含了rpcEndpoint的AkkaRpcActor，负责接收封装成Akka消息的rpc请求 akkaRpcActorProps = Props.create(AkkaRpcActor.class, rpcEndpoint, terminationFuture, getVersion()); ... actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId()); ... }- 接下来生成一个本地的InvocationHandler，用于将调用转换成消息发送到相应的RpcEndpoint(具体细节在下一节发送Rpc请求会详细介绍)
... // 获取这个Endpoint的所有Gateway，也就是所有RPC协议的接口 Set<Class<?>> implementedRpcGateways = new HashSet<>(RpcUtils.extractImplementedRpcGateways(rpcEndpoint.getClass())); ... // 新建一个InvocationHandler，用于将rpc请求包装成LocalRpcInvocation消息并发送给RpcServer(本地) final InvocationHandler akkaInvocationHandler; ... akkaInvocationHandler = new AkkaInvocationHandler(akkaAddress, hostname, actorRef, timeout, maximumFramesize, terminationFuture); ...- 通过Rpc接口和InvocationHandler构造一个代理对象，这个代理对象存在RpcEndpoint的RpcServer变量中，是给RpcEndpoint所在的JVM本地调用使用
// 生成一个包含这些接口的代理，将调用转发到InvocationHandler @SuppressWarnings("unchecked") RpcServer server = (RpcServer) Proxy.newProxyInstance(classLoader, implementedRpcGateways.toArray(new Class<?>[implementedRpcGateways.size()]), akkaInvocationHandler); return server; }
- 启动RpcEndpoint
- 实际上就是启动构造阶段生成的RpcServer的start方法，这个方法由AkkaInvocationHandler实现，实际上就是向绑定的RpcEndpoint的Actor发送一条START消息，通知它服务已启动。
构造Rpc客户端
- Rpc的客户端实际上是一个代理对象，构造这个代理对象，需要提供实现的接口和InvocationHandler，在Flink中有AkkaInvocationHandler的实现。
- 在构造RpcEndpoint的过程中实际上已经生成了一个供本地使用的Rpc客户端。并且通过RpcEndpoint的getSelfGateway方法可以直接获取这个代理对象。
- 而在远程调用时，则通过RpcService的connect方法获取远程RpcEndpoint的客户端(也是一个代理)。connect方法需要提供Actor的地址。(至于地址是如何获得的，可以通过LeaderRetrievalService，在这个部分不多做介绍。)
<C extends RpcGateway> CompletableFuture<C> connect( String address, Class<C> clazz);- 首先通过地址获取ActorSelection，在Prerequisite中也介绍了这是连接远程Actor的方法之一
private <C extends RpcGateway> CompletableFuture<C> connectInternal( final String address, final Class<C> clazz, Function<ActorRef, InvocationHandler> invocationHandlerFactory) { ... // 通过地址获取ActorSelection, 并获取ActorRef引用 final ActorSelection actorSel = actorSystem.actorSelection(address);- 通过ActorSelection获取ActorRef并发送握手消息
final Future<ActorIdentity> identify = Patterns .ask(actorSel, new Identify(42), timeout.toMilliseconds()) .<ActorIdentity>mapTo(ClassTag$.MODULE$.<ActorIdentity>apply(ActorIdentity.class)); final CompletableFuture<ActorIdentity> identifyFuture = FutureUtils.toJava(identify); final CompletableFuture<ActorRef> actorRefFuture = identifyFuture.thenApply( ... // 发送handshake消息 final CompletableFuture<HandshakeSuccessMessage> handshakeFuture = actorRefFuture.thenCompose( (ActorRef actorRef) -> FutureUtils.toJava( Patterns .ask(actorRef, new RemoteHandshakeMessage(clazz, getVersion()), timeout.toMilliseconds()) .<HandshakeSuccessMessage>mapTo(ClassTag$.MODULE$.<HandshakeSuccessMessage>apply(HandshakeSuccessMessage.class))));- 最后根据ActorRef，通过InvocationHandlerFactory生成AkkaInvocationHandler并构造代理
// 根据ActorRef引用生成InvocationHandler return actorRefFuture.thenCombineAsync( handshakeFuture, (ActorRef actorRef, HandshakeSuccessMessage ignored) -> { InvocationHandler invocationHandler = invocationHandlerFactory.apply(actorRef); ClassLoader classLoader = getClass().getClassLoader(); @SuppressWarnings("unchecked") C proxy = (C) Proxy.newProxyInstance( classLoader, new Class<?>[]{clazz}, invocationHandler); return proxy; }, actorSystem.dispatcher()); }- 而invocationHandlerFactory.apply方法如下
Tuple2<String, String> addressHostname = extractAddressHostname(actorRef); return new AkkaInvocationHandler(addressHostname.f0, addressHostname.f1, actorRef, timeout, maximumFramesize, null);
发送Rpc请求
- 上文中客户端会提供代理对象，而代理对象会调用AkkaInvocationHandler的invoke方法并传入Rpc调用的方法和参数信息。
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
-
而在AkkaInvocationHandler对该方法的实现中，会判断方法属于哪个类，如果是Rpc方法的话就会调用invokeRpc方法。
- 首先将方法封装成一个RpcInvocation，它有两种实现，一种是本地的LocalRpcInvocation，不需要序列化，另一种是远程的RemoteRpcInvocation。根据当前AkkaInvocationHandler和对应的RpcEndpoint是否在同一个JVM中来判断生成哪一个。
private Object invokeRpc(Method method, Object[] args) throws Exception { ... final RpcInvocation rpcInvocation = createRpcInvocationMessage(methodName, parameterTypes, args); - 根据返回类型判断使用tell还是ask的形式发送akka消息
Class<?> returnType = method.getReturnType(); final Object result; if (Objects.equals(returnType, Void.TYPE)) { tell(rpcInvocation); result = null; } else if (Objects.equals(returnType, CompletableFuture.class)) { // execute an asynchronous call result = ask(rpcInvocation, futureTimeout); } else { // execute a synchronous call CompletableFuture<?> futureResult = ask(rpcInvocation, futureTimeout); result = futureResult.get(futureTimeout.getSize(), futureTimeout.getUnit()); } return result; }
- 首先将方法封装成一个RpcInvocation，它有两种实现，一种是本地的LocalRpcInvocation，不需要序列化，另一种是远程的RemoteRpcInvocation。根据当前AkkaInvocationHandler和对应的RpcEndpoint是否在同一个JVM中来判断生成哪一个。
Rpc请求的处理
- 首先Rpc消息是通过RpcEndpoint所绑定的Actor的ActorRef发送的，所以接收到消息的就是RpcEndpoint构造期间生成的AkkRpcActor
akkaRpcActorProps = Props.create(AkkaRpcActor.class, rpcEndpoint, terminationFuture, getVersion()); actorRef = actorSystem.actorOf(akkaRpcActorProps, rpcEndpoint.getEndpointId());
- AkkaRpcActor接收到的消息总共有三种
- 一种是握手消息，如上文所述，在客户端构造时会通过ActorSelection发送过来。收到消息后会检查接口，版本，如果一致就返回成功
- 第二种是启停消息。例如在RpcEndpoint调用start方法后，就会向自身发送一条Processing.START消息，来转换当前Actor的状态为STARTED。STOP也类似。并且只有在Actor状态为STARTED时才会处理Rpc请求
- 第三种就是Rpc请求消息，通过解析RpcInvocation获取方法名和参数类型，并从RpcEndpoint类中找到Method对象，并通过反射调用该方法。如果有返回结果，会以Akka消息的形式发送回sender。
