详解Flink组件通信——RPC协议

Posted 大数据那些事

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了详解Flink组件通信——RPC协议相关的知识,希望对你有一定的参考价值。

点击上方蓝字
关注我吧

Flink组件通讯过程

详解Flink组件通信——RPC协议

RPC(本地/远程)调用,底层是通过Akka提供的tell/ask方法进行通信。

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

详解Flink组件通信——RPC协议


1

RpcGateway


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

详解Flink组件通信——RPC协议

从类继承图可以看到基本上所有组件都实现了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和endpointIdthis.rpcService =checkNotNull(rpcService, "rpcService");this.endpointId =checkNotNull(endpointId, "endpointId");// 通过RpcService启动RpcServerthis.rpcServer =rpcService.startServer(this);// 主线程执行器,所有调用在主线程中串行执行this.mainThreadExecutor= new MainThreadExecutor(rpcServer, this::validateRunsInMainThread);}


构造的时候调用rpcService.startServer()启动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);

  • 延迟/立刻调度Runnable、Callable;

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

在构造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, implementedRpcGateways.toArray(newClass<?>[implementedRpcGateways.size()]), akkaInvocationHandler);return server;}


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

  • AkkaInvocationHandler

  • FencedAkkaInvocationHandler

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

class AkkaInvocationHandler implements InvocationHandler, AkkaBasedEndpoint,RpcServer { @Override public void start() { rpcEndpoint.tell(ControlMessages.START,ActorRef.noSender()); }}


4

AkkaRpcActor


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

1)本地Rpc调用LocalRpcInvocation

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

2)RunAsync & CallAsync

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

3)控制消息ControlMessages

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


5

RPC交互过程


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

1、 RPC请求发送

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

AkkaInvocationHandler.java

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { Class<?> declaringClass =method.getDeclaringClass(); Object result;// 判断方法所属的classif(declaringClass.equals(AkkaBasedEndpoint.class) || declaringClass.equals(Object.class) || declaringClass.equals(RpcGateway.class) || declaringClass.equals(StartStoppable.class) || declaringClass.equals(MainThreadExecutable.class)|| declaringClass.equals(RpcServer.class)) { result = method.invoke(this, args); } else if(declaringClass.equals(FencedRpcGateway.class)) {throw new UnsupportedOperationException("AkkaInvocationHandler does not support thecall FencedRpcGateway#" + method.getName() + ". 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类型的消息,代码如下:

AkkaInvocationHandler.java

private Object invokeRpc(Method method, Object[]args) throws Exception { String methodName = method.getName(); Class<?>[] parameterTypes =method.getParameterTypes(); Annotation[][] parameterAnnotations =method.getParameterAnnotations(); Time futureTimeout =extractRpcTimeout(parameterAnnotations, args, timeout);  final RpcInvocation rpcInvocation =createRpcInvocationMessage(methodName, parameterTypes, args);  Class<?> returnType =method.getReturnType();  final Object result;  if (Objects.equals(returnType, Void.TYPE)) { tell(rpcInvocation);  result = null; } else { // Capture the call stack. Itis significantly faster to do that via an exception than // via Thread.getStackTrace(),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<>(); resultFuture.whenComplete((resultValue,failure) -> { if (failure != null) { completableFuture.completeExceptionally(resolveTimeoutException(failure,callStackCapture, method)); } else { completableFuture.complete(deserializeValueIfNeeded(resultValue,method)); } });  if (Objects.equals(returnType,CompletableFuture.class)) { // 如果返回值是CompletableFuture类型,不用阻塞等待返回,直接返回Future对象 result =completableFuture; } else { try { // 如果返回值不是CompletableFuture类型,阻塞等待返回 result = completableFuture.get(futureTimeout.getSize(),futureTimeout.getUnit()); } catch(ExecutionException ee) { throw new RpcException("Failure while obtaining synchronous RPC result.",ExceptionUtils.stripExecutionException(ee)); } } }  return result;}

2、 RPC请求响应

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

AkkaRpcActor.java

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


接收的消息有3种:

1)握手消息

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

AkkaRpcActor.java

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


2)控制消息

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

AkkaRpcActor.java

private void handleControlMessage(ControlMessages controlMessage) { try { switch (controlMessage) { case START: state =state.start(this); break; case STOP: state =state.stop(); break; case TERMINATE: state =state.terminate(this); break; default: handleUnknownControlMessage(controlMessage); } } catch (Exception e) { this.rpcEndpointTerminationResult= RpcEndpointTerminationResult.failure(e); throw e; }}


3)RPC消息

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

AkkaRpcActor.java

private void handleMessage(final Object message) { if (state.isRunning()) { mainThreadValidator.enterMainThread();  try { handleRpcMessage(message); } finally { mainThreadValidator.exitMainThread(); } } else { log.info("The rpcendpoint {} has not been started yet. Discarding message {} until processing isstarted.", rpcEndpoint.getClass().getName(), message.getClass().getName());  sendErrorIfSender(newAkkaRpcException( String.format("Discardmessage, because the rpc endpoint %s has not been started yet.",rpcEndpoint.getAddress()))); }}




详解Flink组件通信——RPC协议

详解Flink组件通信——RPC协议

B站|大数据那些事


想获取更多更全资料

扫码加好友入群

欢迎各位大佬加入开源共享

共同面对大数据领域疑难问题

来稿请投邮箱:miaochuanhai@126.com





以上是关于详解Flink组件通信——RPC协议的主要内容,如果未能解决你的问题,请参考以下文章

Flink内核原理学习组件通信RPC

Flink之间的组件通信

FlinkFLink 通讯组件 RPC

Hadoop - YARN 通信协议

RPC机制之AMQP协议

阅读源码|Spark 与 Flink 的 RPC 实现