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

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

多模态相关论文笔记

多模态相关论文笔记

(cilp) Learning Transferable Visual Models From Natural Language Supervision 从自然语言监督中学习可迁移的视觉模型 openAI 2021年2月 48页 PDF CODE CLIP(Contrastive Language-Image Pre-Training)对比语言图像预训练模型 引言 它比ImageNet模型效果更好&#xff0c…
阅读更多...
32单片机基础：OLED调试工具的使用

32单片机基础：OLED调试工具的使用

下面会介绍OLED显示屏的驱动函数模块&#xff0c;先学会如何使用&#xff0c;至于OLED屏幕的原理和代码编写&#xff0c; 我们之后会再写一篇。 现在我们就是用OLED当一个调试的显示屏&#xff0c;方便我们调试程序。 为什么要调试呢&#xff0c;是为了方便我们看现象&#…
阅读更多...
vue中组件和插件有什么区别？

vue中组件和插件有什么区别？

vue中组件和插件有什么区别&#xff1f; 回答思路&#xff1a;组件是什么&#xff1f;-->插件是什么&#xff1f;-->组件和插件的区别&#xff1f;-->他们的使用场景&#xff1f;组件是什么&#xff1f;组件的优势&#xff1a; 插件是什么&#xff1f;组件和插件的区别…
阅读更多...
unity——shader基础知识点 学习笔记【个人复习向/侵删/有不足之处欢迎斧正】

unity——shader基础知识点 学习笔记【个人复习向/侵删/有不足之处欢迎斧正】

一、shaderLab 1.pass渲染通道 具体实现着色器代码的地方&#xff0c;每个subshader至少有一个pass。 可以利用UsePass命令在其他Shader当中复用该Pass的代码&#xff0c;只需要在其他Shader当中使用 UsePass ”Shader路径/Pass名”&#xff0c;Unity内部会把Pass名称转换为大…
阅读更多...
嵌入式学习之Linux入门篇——使用VMware创建Unbuntu虚拟机

嵌入式学习之Linux入门篇——使用VMware创建Unbuntu虚拟机

目录 主机硬件要求 VMware 安装 安装Unbuntu 18.04.6 LTS 新建虚拟机 进入Unbuntu安装环节 主机硬件要求 内存最少16G 硬盘最好分出一个单独的盘&#xff0c;而且最少预留200G&#xff0c;可以使用移动固态操作系统win7/10/11 VMware 安装 版本&#xff1a;VMware Works…
阅读更多...
Java Clip 播放MP3音频

Java Clip 播放MP3音频

Java Clip 播放MP3音频 1 依赖2 接口3 实现4 测试 项目Value音频格式 添加依赖*.wav(JDK 原生支持)*.pcm(JDK 原生支持)*.au(JDK 原生支持)*.aiff(JDK 原生支持)*.mp3mp3spi.jar*.flacjflac-codec.jar 1 依赖 <dependency><groupId>com.googlecode.soundlibs<…
阅读更多...
Python | OS模块操作

Python | OS模块操作

一、介绍 Python的os模块提供了许多与操作系统交互的函数&#xff0c;可以用于文件和目录的操作、进程管理、环境变量的访问等。以下是os模块的一些常用功能&#xff1a; 文件和目录操作&#xff1a;os模块提供了许多函数来进行文件和目录的操作&#xff0c;如创建文件夹(os.mk…
阅读更多...
小红书如何写标题吸引人？

小红书如何写标题吸引人？

好奇类&#xff1a;利用观众得不到&#xff0c;没体验过的事物&#xff0c;激发好奇心 万能公式:&#xff08; )是一种什么体验&#xff1f; 例如:日入过千是一种什么体验? 极限类&#xff1a;将抽象的内容具象化&#xff0c;吸引力强&#xff0c;让观众引起重视 万能公式…
阅读更多...
CQT新里程碑：SOC 2 数据安全认证通过，加强其人工智能支持

CQT新里程碑：SOC 2 数据安全认证通过，加强其人工智能支持

Covalent Network&#xff08;CQT&#xff09;发展新里程碑&#xff1a;SOC 2 数据安全认证通过&#xff0c;进一步加强了其人工智能支持 Covalent Network&#xff08;CQT&#xff09;现已完成并通过了严格的 Service Organization Control&#xff08;SOC) 2 Type II 的合规性…
阅读更多...
物联网和人工智能的融合

物联网和人工智能的融合

物联网和人工智能的融合 1. 物联网和人工智能的融合2. 芯片技术的进步3. 安全和隐私保护挑战4. 软件开发和调试技术的创新5. 自动化和智能化趋势 1. 物联网和人工智能的融合 随着物联网和人工智能技术的快速发展&#xff0c;嵌入式系统将更多地与物联网设备和人工智能算法相结…
阅读更多...
vivo 基于 StarRocks 构建实时大数据分析平台，为业务搭建数据桥梁

vivo 基于 StarRocks 构建实时大数据分析平台，为业务搭建数据桥梁

在大数据时代&#xff0c;数据分析和处理能力对于企业的决策和发展至关重要。 vivo 作为一家全球移动互联网智能终端公司&#xff0c;需要基于移动终端的制造、物流、销售等各个方面的数据进行分析以满足业务决策。 而随着公司数字化服务的演进&#xff0c;业务诉求和技术架构有…
阅读更多...
ELK Stack 日志平台搭建

ELK Stack 日志平台搭建

前言 最近在折腾 ELK 日志平台&#xff0c;它是 Elastic 公司推出的一整套日志收集、分析和展示的解决方案。 专门实操了一波&#xff0c;这玩意看起来简单&#xff0c;但是里面的流程步骤还是很多的&#xff0c;而且遇到了很多坑。在此记录和总结下。 本文亮点&#xff1a;…
阅读更多...
windows系统中jenkins构建报错提示“拒绝访问”

windows系统中jenkins构建报错提示“拒绝访问”

一.背景 之前徒弟在windows中安装的jenkins,运行的时候用的是java -jar jenkins.war来运行的。服务器只有1个盘符C盘。今天说构建错误了&#xff0c;问我修改了啥&#xff0c;我年前是修改过构建思路的。 二.问题分析 先看jenkins构建任务的日志&#xff0c;大概是xcopy命令执…
阅读更多...
【linux】使用g++调试内存泄露：AddressSanitizer

【linux】使用g++调试内存泄露：AddressSanitizer

1、简介 AddressSanitizer（又名 ASan）是 C/C++ 的内存错误检测器。它可以用来检测： 释放后使用（悬空指针） 堆缓冲区溢出 堆栈缓冲区溢出 全局缓冲区溢出 在作用域之后使用 初始化顺序错误 内存泄漏这个工具非常快，只将被检测的程序速度减慢约2倍，而Valgrind将会是程序…
阅读更多...
JavaScript：问号?的多种用法

JavaScript：问号?的多种用法

文章目录 条件运算符 (三元运算符)可选链操作符 (?.)空值合并操作符 (??)逻辑赋值运算符&#xff08;?? &#xff09;补充&#xff1a;&#xff08;&&、||&#xff09; 正则表达式中 条件运算符 (三元运算符) 早在ES1&#xff08;ECMAScript 1st Edition&#xf…
阅读更多...
word软件中硬件图像加速有什么用处？禁用硬件图形加速(G)会影响word文档中插入图片的分辨率吗？

word软件中硬件图像加速有什么用处？禁用硬件图形加速(G)会影响word文档中插入图片的分辨率吗？

问题描述&#xff1a;word软件中硬件图像加速有什么用处&#xff1f;禁用硬件图形加速(G)会影响word文档中插入图片的分辨率吗&#xff1f; 问题解答&#xff1a; 在 Microsoft Word 中&#xff0c;硬件图形加速主要用于提高图形元素的渲染速度和性能&#xff0c;特别是处理大…
阅读更多...
FPS游戏之账号登录流程

FPS游戏之账号登录流程

在游戏开发中&#xff0c;账号登录的接入流程主要包含以下几个步骤&#xff1a; 创建账号系统&#xff1a;首先需要创建一个账号系统&#xff0c;能够让用户注册账号、登录账号以及找回和修改密码等。这个账号系统可以是自己开发的&#xff0c;也可以是接入的第三方服务如Goog…
阅读更多...
如何添加或编辑自定义WordPress侧边栏

如何添加或编辑自定义WordPress侧边栏

WordPress侧边栏是许多WordPress网站上的固定装置。它为您的内容提供了一个垂直空间&#xff0c;您可以在其中帮助读者导航、增加电子邮件列表或社交关注、展示广告等。 因为它是许多WordPress网站不可或缺的一部分&#xff0c;所以我们认为侧边栏值得拥有自己的大型指南。在这…
阅读更多...
【leetcode题解C++】763.划分字母区间 and 56.合并区间 and 738.单调递增的数字

【leetcode题解C++】763.划分字母区间 and 56.合并区间 and 738.单调递增的数字

763. 划分字母区间 给你一个字符串 s 。我们要把这个字符串划分为尽可能多的片段&#xff0c;同一字母最多出现在一个片段中。注意&#xff0c;划分结果需要满足&#xff1a;将所有划分结果按顺序连接&#xff0c;得到的字符串仍然是 s 。 返回一个表示每个字符串片段的长度的…
阅读更多...
【AIGC】开源声音克隆GPT-SoVITS

【AIGC】开源声音克隆GPT-SoVITS

GPT-SoVITS 是由 RVC 创始人 RVC-Boss 与 AI 声音转换技术专家 Rcell 共同开发的一款跨语言 TTS 克隆项目&#xff0c;被誉为“最强大中文声音克隆项目” 相比以往的声音克隆项目&#xff0c;GPT-SoVITS 对硬件配置的要求相对较低&#xff0c;一般只需 6GB 显存以上的 GPU 即可…
阅读更多...
最新文章

Copyright @ 2022~2023