当前位置:   article > 正文

【Flink集群RPC通讯机制(二)】创建AkkaRpcService、启动RPC服务、实现相互通信

【Flink集群RPC通讯机制(二)】创建AkkaRpcService、启动RPC服务、实现相互通信

零. 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中的RpcServer
   this.rpcServer = rpcService.startServer(this);
   this.mainThreadExecutor = new MainThreadExecutor(rpcServer, 
   this::validateRunsInMainThread);
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8

具体看下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等接口。

  1. 创建好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;  
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52
  • 53
  • 54
  • 55
  • 56
  • 57
  • 58
  • 59
  • 60
  • 61
  • 62
  • 63
  • 64
  • 65
  • 66
  • 67
  • 68
  • 69
  • 70
  • 71
  • 72

 

3. ResourceManager中RPC服务的启动

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

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

ClusterEntrypoint.startCluster->runCluster
->dispatcherResourceManagerComponentFactory.create
->resourceManager.start();
=>
public final void start() {  
    rpcServer.start();  
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7

继续探索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);  
            });  
}
  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20

具体看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());  
}

  • 1
  • 2
  • 3
  • 4
  • 5
  • 6
  • 7
  • 8
  • 9
  • 10
  • 11
  • 12
  • 13
  • 14
  • 15
  • 16
  • 17
  • 18
  • 19
  • 20
  • 21
  • 22
  • 23
  • 24
  • 25
  • 26
  • 27
  • 28
  • 29
  • 30
  • 31
  • 32
  • 33
  • 34
  • 35
  • 36
  • 37
  • 38
  • 39
  • 40
  • 41
  • 42
  • 43
  • 44
  • 45
  • 46
  • 47
  • 48
  • 49
  • 50
  • 51
  • 52

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

 

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

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/羊村懒王/article/detail/174895
推荐阅读
相关标签
  

闽ICP备14008679号