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

文章目录

    • 零. RpcService服务概述
    • 1. AkkaRpcService的创建和初始化
    • 2.通过AkkaRpcService初始化RpcServer
    • 3. ResourceManager中RPC服务的启动
    • 4. 实现相互通讯能力

零. 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);
}

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

 

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设计与实现：核心原理与源码解析》–张利兵

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/400669.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

CSS基础属性

CSS基础属性

【三】基础属性 【1】高度和宽度 &#xff08;1&#xff09;参数 width&#xff08;宽度&#xff09;&#xff1a;用于设置元素的宽度。可以使用具体的数值&#xff08;如像素值&#xff09;或百分比来指定宽度。 height&#xff08;高度&#xff09;&#xff1a;用于设置元…
阅读更多...
小程序端学习

小程序端学习

P2 创建Uni-app 分离窗口 一样的Ctrl S P3 细节知识点 创建新的小程序页面
阅读更多...
设计模式浅析(六) ·命令模式

设计模式浅析(六) ·命令模式

设计模式浅析(六) 命令模式 日常叨逼叨 java设计模式浅析&#xff0c;如果觉得对你有帮助&#xff0c;记得一键三连&#xff0c;谢谢各位观众老爷&#x1f601;&#x1f601; 命令模式 概念 命令模式&#xff08;Command Pattern&#xff09;是一种行为设计模式&#xff0c…
阅读更多...
C#知识点-13（进程、多线程、使用Socket实现服务器与客户端通信）

C#知识点-13（进程、多线程、使用Socket实现服务器与客户端通信）

进程 定义&#xff1a;每一个正在运行的应用程序&#xff0c;都是一个进程 进程不等于正在运行的应用程序。而是为应用程序的运行构建一个运行环境 Process[] pros Process.GetProcesses();//获取电脑中所有正在运行的进程//通过进程&#xff0c;直接打开文件//告诉进程&…
阅读更多...
解决IDEA搜不到插件

解决IDEA搜不到插件

File -> Settings -> Plugins https://plugins.jetbrains.com/ 完成以上操作即可搜到插件
阅读更多...
小程序--组件通信

小程序--组件通信

一、父传子 与vue利用props类似&#xff0c;小程序是利用自定义属性&#xff1a;properties // components/my-nav/my-nav.js Component({// 小程序组件默认样式是隔离&#xff0c;addGlobalClass设置为true可允许外部修改样式options: {addGlobalClass: true,// 只要使用到具…
阅读更多...
面试经典150题——生命游戏

面试经典150题——生命游戏

​"Push yourself, because no one else is going to do it for you." - Unknown 1. 题目描述 2. 题目分析与解析 2.1 思路一——暴力求解 之所以先暴力求解&#xff0c;是因为我开始也没什么更好的思路&#xff0c;所以就先写一种解决方案&#xff0c;没准写着写…
阅读更多...
OJ链接——打印从1到最大的n位数

OJ链接——打印从1到最大的n位数

目录 1. 题目描述2. 示例3. 分析思路4. 完整代码 1. 题目描述 输入数字 n&#xff0c;按顺序打印出从 1 到最大的 n 位十进制数。比如输入 3&#xff0c;则打印出 1、2、3 一直到最大的 3 位数 999。 用返回一个整数列表来代替打印n 为正整数&#xff0c;0 < n < 5 链接在…
阅读更多...
mac m1调试aarch64 android kernel最终方案

mac m1调试aarch64 android kernel最终方案

问题 这是之前的&#xff0c;调试android kernel的方案还是太笨重了 完美调试android-goldfish(linux kernel) aarch64的方法 然后&#xff0c;看GeekCon AVSS 2023 Qualifier&#xff0c;通过 sdk-repo-linux_aarch64-emulator-8632828.zip 进行启动 完整编译的aosp kernnl…
阅读更多...
Code-Audit(代码审计)习题记录4-5

Code-Audit(代码审计)习题记录4-5

4、习题4 题目内容如下&#xff1a; <?php error_reporting(0); show_source(__FILE__); $a $_REQUEST[hello]; eval("var_dump($a);"); 函数解释 $REQUEST — HTTP Request 变量&#xff0c;默认情况下包含了 [$GET]&#xff0c;[$POST] 和 [$COOKIE]的数…
阅读更多...
Git合并固定分支的某一部分至当前分支

Git合并固定分支的某一部分至当前分支

在 Git 中&#xff0c;通常使用 git merge 命令来将一个分支的更改合并到另一个分支。如果你只想合并某个分支的一部分代码&#xff0c;可以使用以下两种方法&#xff1a; 1.批量文件合并 1.1.创建并切换到一个新的临时分支 首先&#xff0c;从要合并的源分支&#xff08;即要…
阅读更多...
S281 LoRa网关助力智慧城市建设的智能交通管理

S281 LoRa网关助力智慧城市建设的智能交通管理

S281 LoRa网关作为智慧城市建设中的重要组成部分&#xff0c;发挥着关键的作用&#xff0c;特别是在智能交通管理方面。通过连接各类传感器设备和物联网终端&#xff0c;S281 LoRa网关实现了对城市交通系统的远程监控、智能调度和信息化管理&#xff0c;为城市交通管理部门提供…
阅读更多...
学习鸿蒙基础（5）

学习鸿蒙基础（5）

一、honmonyos的page路由界面的路径 新建了一个page,然后删除了。运行模拟器的时候报错了。提示找不到这个界面。原来是在路由界面没有删除这个page。新手刚接触找了半天才找到这个路由。在resources/base/profile/main_pages.json 这个和微信小程序好类似呀。 吐槽&#xf…
阅读更多...
MKS T3BI集成蝶阀说明T3B-T3PRS-232Supplement

MKS T3BI集成蝶阀说明T3B-T3PRS-232Supplement

MKS T3BI集成蝶阀说明T3B-T3PRS-232Supplement
阅读更多...
常见的排序算法整理

常见的排序算法整理

1.冒泡排序 1.1 冒泡排序普通版 每次冒泡过程都是从数列的第一个元素开始&#xff0c;然后依次和剩余的元素进行比较&#xff0c;若小于相邻元素&#xff0c;则交换两者位置&#xff0c;同时将较大元素作为下一个比较的基准元素&#xff0c;继续将该元素与其相邻的元素进行比…
阅读更多...
人工智能_CPU安装运行ChatGLM大模型_安装清华开源人工智能AI大模型ChatGlm-6B_004---人工智能工作笔记0099

人工智能_CPU安装运行ChatGLM大模型_安装清华开源人工智能AI大模型ChatGlm-6B_004---人工智能工作笔记0099

上一节003节我们安装到最后,本来大模型都可以回答问题了,结果, 5分钟后给出提示,需要GPU,我去..继续看官网,如何配置CPU运行 没办法继续看: https://github.com/THUDM/ChatGLM-6B 这里是官网可以看到 需要gcc的版本是11.3.0,这里我们先没有去安装,直接试试再说 yum instal…
阅读更多...
在UE5中使用OverlayMaterial制作多材质效果

在UE5中使用OverlayMaterial制作多材质效果

UE5.1中新增了OverlayMaterial&#xff0c;可以让物体套用2个材质球效果&#xff0c;如A材质球为正常材质内容&#xff0c;B材质球为菲涅尔&#xff0c;或是B材质球是法线外拓描边等&#xff0c;该功能类似Unity的多pass效果&#xff0c;方便了日常使用。 下面就讲将怎么用Ove…
阅读更多...
手把手教你如何搭建性能测试环境

手把手教你如何搭建性能测试环境

前言 在进行性能则试前&#xff0c;需要完成性能测试的搭建工作&#xff0c;一般包括硬件环境、软件环境及网络环境&#xff0c;可以要求配置和开发工程师协助完成&#xff0c;但是作为一个优秀性能测试工程师&#xff0c;这也是你的必备技能之一。 性能测试环境与功能测试环…
阅读更多...
4款文案神器，为你写作高质量文案

4款文案神器，为你写作高质量文案

4款文案神器&#xff0c;为你写作高质量文案!在当今信息爆炸的时代&#xff0c;写作成为了一项重要的技能&#xff0c;无论是在工作中、学习中还是生活中&#xff0c;我们都需要用文字来传达信息、表达想法。然而&#xff0c;要写出高质量的文案并非易事&#xff0c;需要不断地…
阅读更多...
开源软件的影响力及未来发展趋势

开源软件的影响力及未来发展趋势

开源软件的影响力 在当今数字化时代&#xff0c;开源软件已经成为技术创新、商业模式和安全风险等方面不可或缺的一部分。本文将从开源软件如何推动技术创新、开源软件的商业模式、开源软件的安全风险、开源软件的未来发展趋势以及开源软件在各行业的应用案例几个方面进行深入…
阅读更多...
最新文章

Copyright @ 2022~2023