【callstack】详解Flink组件通信——RPC协议

Flink组件通信过程

通过Akka提供的tell/ask方法通信的本地/远程(RPC)调用。

Flink中RPC框架中涉及的主要类：

1、RpcGateway

Flink的RPC协议通过RpcGateway来定义，主要定义通信行为；用于远程调用RpcEndpoint的某些方法，可以理解为客服端代理。

若想与远端Actor通信，则必须提供地址（ip和port），如在Flink-on-Yarn模式下，JobMaster会先启动ActorSystem，此时TaskExecutor的Container还未分配，后面与TaskExecutor通信时，必须让其提供对应地址。

从类继承图可以看到基本上所有组件都实现了RpcGateway接口，其代码如下：

public interface RpcGateway {
/**
* Returns the fully qualified address underwhich the associated rpc endpoint is reachable.
*
* @return Fully qualified (RPC) address underwhich the associated rpc endpoint is reachable
*/
String getAddress();
/**
* Returns the fully qualified hostname underwhich the associated rpc endpoint is reachable.
*
* @return Fully qualified hostname under whichthe associated rpc endpoint is reachable
*/
String getHostname();
}

2、RpcEndpoint

RpcEndpoint是通信终端，提供RPC服务组件的生命周期管理(start、stop)。每个RpcEndpoint对应了一个路径（endpointId和actorSystem共同确定），每个路径对应一个Actor，其实现了RpcGateway接口，其构造函数如下：

protected RpcEndpoint(final RpcService rpcService, final String endpointId) {
// 保存rpcService和endpointId
=checkNotNull(rpcService, "rpcService");
=checkNotNull(endpointId, "endpointId");
// 通过RpcService启动RpcServer
=r(this);
// 主线程执行器，所有调用在主线程中串行执行
new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);
}

构造的时候调用r()启动RpcServer，进入可以接收处理请求的状态，最后将RpcServer绑定到主线程上真正执行起来。

在RpcEndpoint中还定义了一些方法如runAsync(Runnable)、callAsync(Callable, Time)方法来执行Rpc调用，值得注意的是在Flink的设计中，对于同一个Endpoint，所有的调用都运行在主线程，因此不会有并发问题，当启动RpcEndpoint/进行Rpc调用时，其会委托RcpServer进行处理。

3、RpcService和RpcServer

Akka有两种核心的异步通信方式：tell和ask。

RpcService 和 RpcServer是RpcEndPoint的成员变量。

1）RpcService 是Rpc服务的接口，其主要作用如下：

• 根据提供的RpcEndpoint来启动和停止RpcServer（Actor）；
• 根据提供的地址连接到RpcServer，并返回一个RpcGateway；
• 延迟/立刻调度Runnable、Callable；

在Flink中实现类为AkkaRpcService，是 Akka 的 ActorSystem 的封装，基本可以理解成 ActorSystem 的一个适配器。在ClusterEntrypoint（JobMaster）和TaskManagerRunner（TaskExecutor）启动的过程中初始化并启动。

AkkaRpcService中封装了ActorSystem，并保存了ActorRef到RpcEndpoint的映射关系。RpcService跟RpcGateway类似，也提供了获取地址和端口的方法。

在构造RpcEndpoint时会启动指定rpcEndpoint上的RpcServer，其会根据RpcEndpoint类型（FencedRpcEndpoint或其他）来创建不同的AkkaRpcActor（FencedAkkaRpcActor或AkkaRpcActor），并将RpcEndpoint和AkkaRpcActor对应的ActorRef保存起来，AkkaRpcActor是底层Akka调用的实际接收者，RPC的请求在客户端被封装成RpcInvocation对象，以Akka消息的形式发送。

最终使用动态代理将所有的消息转发到InvocationHandler，具体代码如下：

public <Cextends RpcEndpoint & RpcGateway> RpcServer startServer(C rpcEndpoint) {
... ...
// 生成RpcServer对象，而后对该server的调用都会进入Handler的invoke方法处理，handler实现了多个接口的方法
// 生成一个包含这些接口的代理，将调用转发到InvocationHandler
@SuppressWarnings("unchecked")
RpcServerserver = (RpcServer) Proxy.newProxyInstance(
classLoader,
im(newClass<?>[im()]),
akkaInvocationHandler);
return server;
}

2）RpcServer负责接收响应远端RPC消息请求。有两个实现：

• AkkaInvocationHandler
• FencedAkkaInvocationHandler

RpcServer的启动是通知底层的AkkaRpcActor切换为START状态，开始处理远程调用请求：

class AkkaInvocationHandler implements InvocationHandler, AkkaBasedEndpoint,RpcServer {
@Override
public void start() {
r());
}
}

4、AkkaRpcActor

AkkaRpcActor是Akka的具体实现，主要负责处理如下类型消息：

1）本地Rpc调用LocalRpcInvocation

会指派给RpcEndpoint进行处理，如果有响应结果，则将响应结果返还给Sender。

2）RunAsync & CallAsync

这类消息带有可执行的代码，直接在Actor的线程中执行。

3）控制消息ControlMessages

用来控制Actor行为，START启动，STOP停止，停止后收到的消息会丢弃掉。

5、RPC交互过程

RPC通信过程分为请求和响应。

1、 RPC请求发送

在RpcService中调用connect()方法与对端的RpcEndpoint建立连接，connect()方法根据给的地址返回InvocationHandler(AkkaInvocationHandler或FencedAkkaInvocationHandler，也就是RpcServer的实现)。

前面分析到客户端提供代理对象RpcServer，代理对象会调用AkkaInvocationHandler的invoke方法并传入RPC调用的方法和参数信息，代码如下：

AkkaInvoca

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
Class<?> declaringClass =me();
Object result;
// 判断方法所属的class
i) ||
declaringCla) ||
declaringCla) ||
declaringCla) ||
declaringCla)||
declaringCla)) {
result = me(this, args);
} else i)) {
throw new UnsupportedOperationException("AkkaInvocationHandler does not support thecall FencedRpcGateway#" +
me() + ". This indicates thatyou retrieved a FencedRpcGateway without specifying a " +
"fencingtoken. Please use RpcService#connect(RpcService, F, Time) with F being thefencing token to " +
"retrieve aproperly FencedRpcGateway.");
} else {
// rpc调用
result = invokeRpc(method, args);
}
return result;
}

代码中判断所属的类，如果是RPC方法，则调用invokeRpc方法。将方法调用封装为RPCInvocation消息。如果是本地则生成LocalRPCInvocation，本地消息不需要序列化，如果是远程调用则创建RemoteRPCInvocation。

判断远程方法调用是否需要等待结果，如果无需等待(void)，则使用向Actor发送tell类型的消息，如果需要返回结果，则向Acrot发送ask类型的消息，代码如下：

AkkaInvoca

private Object invokeRpc(Method method, Object[]args) throws Exception {
String methodName = me();
Class<?>[] parameterTypes =me();
Annotation[][] parameterAnnotations =me();
Time futureTimeout =extractRpcTimeout(parameterAnnotations, args, timeout);

final RpcInvocation rpcInvocation =createRpcInvocationMessage(methodName, parameterTypes, args);

Class<?> returnType =me();

final Object result;

if (returnType, Void.TYPE)) {
tell(rpcInvocation);

result = null;
} else {
// Capture the call stack. Itis significantly faster to do that via an exception than
// via T(),because exceptions lazily initialize the stack trace, initially only
// capture a lightweightnative pointer, and convert that into the stack trace lazily when needed.
final ThrowablecallStackCapture = captureAskCallStack ? new Throwable() : null;

// execute an asynchronouscall
// 异步调用等待返回
final CompletableFuture<?> resultFuture = ask(rpcInvocation, futureTimeout);

final CompletableFuture<Object> completableFuture = newCompletableFuture<>();
re((resultValue,failure) -> {
if (failure != null) {
com(resolveTimeoutException(failure,callStackCapture, method));
} else {
com(deserializeValueIfNeeded(resultValue,method));
}
});

if (returnType,Com)) {
// 如果返回值是CompletableFuture类型，不用阻塞等待返回，直接返回Future对象
result =completableFuture;
} else {
try {
// 如果返回值不是CompletableFuture类型，阻塞等待返回
result = com(),());
} catch(ExecutionException ee) {
throw new RpcException("Failure while obtaining synchronous RPC result.",Exce(ee));
}
}
}

return result;
}

2、 RPC请求响应

RPC消息通过RpcEndpoint所绑定的Actor的ActorRef发送的，AkkaRpcActor是消息接收的入口，AkkaRpcActor在RpcEndpoint中构造生成，负责将消息交给不同的方法进行处理。

AkkaR

public Receive createReceive() {
return ReceiveBuilder.create()
.match(RemoteHandshakeMessage.class,this::handleHandshakeMessage)
.match(ControlMessages.class,this::handleControlMessage)
.matchAny(this::handleMessage)
.build();
}

接收的消息有3种：

1）握手消息

在客户端构造时会通过ActorSelection发送过来。收到消息后检查接口、版本是否匹配。

AkkaR

private void handleHandshakeMessage(RemoteHandshakeMessagehandshakeMessage) {
if(!isCompatibleVersion())) {
// 版本不兼容异常处理
sendErrorIfSender(newAkkaHandshakeException(
S(
"Versionmismatch between source (%s) and target (%s) rpc component. Please verify thatall components have the same version.",
(),
getVersion())));
} else if(!isGatewaySupported())) {
// RpcGateway不匹配异常处理
sendErrorIfSender(newAkkaHandshakeException(
S(
"The rpcendpoint does not support the gateway %s.",
().getSimpleName())));
} else {
getSender().tell(new S),getSelf());
}
}

2）控制消息

在RpcEndpoint调用start方法后，会向自身发送一条Proce消息来转换当前Actor的状态为STARTED，STOP也类似，并且只有在Actor状态为STARTED时才会处理RPC请求。

AkkaR

private void handleControlMessage(ControlMessages controlMessage) {
try {
switch (controlMessage) {
case START:
state =(this);
break;
case STOP:
state =();
break;
case TERMINATE:
state =(this);
break;
default:
handleUnknownControlMessage(controlMessage);
}
} catch (Exception e) {
R(e);
throw e;
}
}

3）RPC消息

通过解析RpcInvocation获取方法名和参数类型，并从RpcEndpoint类中找到Method对象，通过反射调用该方法。如果有返回结果，会以Akka消息的形式发送回发送者。

AkkaR

private void handleMessage(final Object message) {
if ()) {
mainT();

try {
handleRpcMessage(message);
} finally {
mainT();
}
} else {
log.info("The rpcendpoint {} has not been started yet. Discarding message {} until processing isstarted.",
r().getName(),
me().getName());

sendErrorIfSender(newAkkaRpcException(
S("Discardmessage, because the rpc endpoint %s has not been started yet.",r())));
}
}

---

更多精彩内容：

Flink 流处理Api之Sink

一网打尽Flink高频面试题

大咖分享 | 通过制作一个迷你版Flink来学习Flink源码

用flink能替代spark的批处理功能吗

详解Flink通讯模型——Akka与Actor模型