当前位置: 首页 > news >正文

Flink源码之TaskManager启动流程

从启动命令flink-daemon.sh可以看出TaskManger入口类为org.apache.flink.runtime.taskexecutor.TaskManagerRunner

TaskManagerRunner::main
TaskManagerRunner::runTaskManagerProcessSecurely
TaskManagerRunner::runTaskManager //构造TaskManagerRunner并调用start()方法
TaskManagerRunner::new //核心

在TaskManagerRunner构造函数中,可以看出与JobManger类似,也是先构造出一些公共服务:

highAvailabilityServices//用于获取JobManger的地址
rpcService //将TaskExecutor包装为AkkaActor提供RPC服务
heartbeatServices //心跳服务,与JobManger通信
metricRegistry //metric服务,提供metric注册和查询
blobCacheService //缓存Blob

这些服务在构造TaskExecutor时作为构造函数参数传入

构造TaskExecutor前会先构造TaskManagerServices辅助TaskExecutor实现其核心功能

TaskManagerRunner::createTaskExecutorService
TaskManagerRunner::startTaskManager // 构造MetricGroup和相关服务
TaskManagerServices.fromConfiguration//读取TaskManger的配置信息启动TaskManager相关服务
TaskExecutor::new  //核心

启动TaskEexector后会与ResouceManager建立连接,将自身信息注册到RM后发送Slot报告给RM,具体调用链路如下:

TaskManagerRunner::start
TaskExecutorToServiceAdapter::start
TaskExecutor::start
TaskExecutor::onStart
TaskExecutor::startTaskExecutorServices //获取ResourceManager地址后与ResourceManager建立连接,发送Slot报告
ResourceManagerLeaderListener::notifyLeaderAddress
TaskExecutor::notifyOfNewResourceManagerLeader
TaskExecutor::reconnectToResourceManager
TaskExecutor::tryConnectToResourceManager
TaskExecutor::connectToResourceManager
TaskExecutorToResourceManagerConnection::start
RegisteredRpcConnection::start
RegisteredRpcConnection::createNewRegistration
TaskExecutorToResourceManagerConnection::generateRegistration
RetryingRegistration::startRegistration //与resourcemanager建立连接
RetryingRegistration::register
ResourceManagerRegistration::invokeRegistration //向ResourceManager注册TaskExecutorRegistration信息
ResourceManagerGateway.registerTaskExecutor
TaskExecutorToResourceManagerConnection::onRegistrationSuccess
ResourceManagerRegistrationListener::onRegistrationSuccess
TaskExecutor::establishResourceManagerConnection ResourceManagerGateway.sendSlotReport //发送自身slot信息给ResourceManagerHeartbeatManagerImpl::monitorTarget//与RM建立心跳连接,当接到来自RM的心跳请求时,就会将SlotReport发送给RM作为心跳回应

TaskExecutor提供了以下两个核心方法:

 //RM将Slot分配给JobMaster请求TM将具体Slot信息发送给JobMasterCompletableFuture<Acknowledge> requestSlot(SlotID slotId,JobID jobId,AllocationID allocationId,ResourceProfile resourceProfile,String targetAddress,ResourceManagerId resourceManagerId,@RpcTimeout Time timeout);//执行JobMaster提交的物理Task       
CompletableFuture<Acknowledge> submitTask(TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, @RpcTimeout Time timeout);    

TaskManager中管理Slot的实现类TaskSlotTableImpl,该实例记录了Slot的分配信息。

在这里插入图片描述

HeartBeat

在TaskExecutor构造函数中有两个HeartbeatManager,实现类都是HeartbeatManagerImpl,此类是接受心跳请求,发送心跳响应:

ResourceManagerHeartbeatManager //响应RM的心跳请求,心跳响应中带上SlotReport
JobManagerHeartbeatManager  //响应JobMaster的心跳请求, 心跳响应中带上AccumulatorReport

调用HeartbeatManagerImpl.monitorTarget(ResourceID resourceID, HeartbeatTarget heartbeatTarget) 与目标对象建立心跳连接。

HeartbeatManager还有个实现类是HeartbeatManagerSenderImpl,用于主动向监控目标发送心跳请求,比如在ResourceManager中创建的就是HeartbeatManagerSenderImpl,TaskManager启动时向ResourceManager注册后,RM就会调用HeartbeatManagerSenderImpl.monitor监控TM, 并定时向TM的HeartbeatManagerImpl发送心跳请求。同样,在JobMaster中创建的也是HeartbeatManagerSenderImpl,JobMaster定时向执行当前Job的TM发送心跳请求,TM响应与该Job相关信息。

综上,TM启动后向RM注册,与TM通过心跳信息同步Slot分配状况,接受RM的Slot分配请求向JobMaster提供Slot后,就可以接受JobMaster 执行具体的物理Task了。

相关文章:

Flink源码之TaskManager启动流程

从启动命令flink-daemon.sh可以看出TaskManger入口类为org.apache.flink.runtime.taskexecutor.TaskManagerRunner TaskManagerRunner::main TaskManagerRunner::runTaskManagerProcessSecurely TaskManagerRunner::runTaskManager //构造TaskManagerRunner并调用start()方法 …...

加入微软MCPP有什么优势?

目录 专业认可 技术支持 销售和市场推广支持 培训和认证 业务机会和合作伙伴网络...

leetcode做题笔记78子集

给你一个整数数组 nums &#xff0c;数组中的元素 互不相同 。返回该数组所有可能的子集&#xff08;幂集&#xff09;。 解集 不能 包含重复的子集。你可以按 任意顺序 返回解集。 思路一&#xff1a;回溯 void backtracking(int* nums, int numsSize, int** res, int* ret…...

Skywalking-9.6.0系列之本地源码编译并启动

Skywalking相信有很多人使用过&#xff0c;通过容器或者下载安装包进行安装的&#xff0c;今天从源代码角度&#xff0c;拉取、构建、启动。 官方文档步骤简洁明了&#xff0c;我这边会结合自己遇到的一些问题做出总结。 当前构建资源版本&#xff1a; MAC 10.15.7IDEA 2021.…...

proteus结合keil-arm编译器构建STM32单片机项目进行仿真

proteus是可以直接创建设计图和源码的&#xff0c;但是源码编译它需要借助keil-arm编译器&#xff0c;也就是我们安装keil-mdk之后自带的编译器。 下面给出一个完整的示例&#xff0c;主要是做一个LED灯闪烁的效果。 新建工程指定路径&#xff0c;Schematic,PCB layout都选择默…...

第五十三天

●剪辑——Pr 剪辑(Film editing)&#xff0c;即将影片制作中所拍摄的大量素材&#xff0c;经过选择、取舍、分解与组接&#xff0c;最终完成一个连贯流畅、含义明确、主题鲜明并有艺术感染力的作品。 •线性编辑 将素材按时间顺序连接成新的连续画面的技术 •非线性编辑 …...

gorm基本操作

一、gorm安装 1.下载gorm go get -u gorm.io/gorm //gorm框架 go get -u gorm.io/driver/mysql //驱动2.mysql准备工作 mysql> create database godb; mysql> grant all on *.* to admin% identified by golang123!; mysql> flush privileges;3.导入gorm框架 impo…...

华为OD机试 - 排队游戏(Java JS Python)

题目描述 新来的老师给班里的同学排一个队。 每个学生有一个影力值。 一些学生是刺头,不会听老师的话,自己选位置,非刺头同学在剩下的位置按照能力值从小到大排。 对于非刺头同学,如果发现他前面有能力值比自己高的同学,他不满程度就增加,增加的数量等于前面能力值比…...

滚动条样式更改

::-webkit-scrollbar 滚动条整体部分&#xff0c;可以设置宽度啥的 ::-webkit-scrollbar-button 滚动条两端的按钮 ::-webkit-scrollbar-track 外层轨道 ::-webkit-scrollbar-track-piece 内层滚动槽 ::-webkit-scrollbar-thumb 滚动的滑块 ::-webkit-scrollbar…...

掌握Python的X篇_33_MATLAB的替代组合NumPy+SciPy+Matplotlib

numPy 通常与 SciPy( Scientific Python )和 Matplotlib (绘图库)一起使用&#xff0c;这种组合广泛用于替代 MatLab&#xff0c;是一个强大的科学计算环境&#xff0c;有助于我们通过 Python 学习数据科学或者机器学习。 文章目录 1. numpy1.1 numpy简介1.2 矩阵类型的nparra…...

Python解决-力扣002-两数相加

两数相加&#xff1a;链表表示的逆序整数求和 在这篇技术博客中&#xff0c;我们将讨论一个力扣&#xff08;LeetCode&#xff09;上的编程题目&#xff1a;两数相加。这个问题要求我们处理两个非空链表&#xff0c;它们表示两个非负整数。每个链表中的数字都是逆序存储的&…...

nginx基于源码安装的方式对静态页面、虚拟主机(IP、端口、域名)和日志文件进行配置

一.静态页面 1.更改页面内容 2.更改配置文件 3.测试 二.虚拟主机配置 1.基于IP &#xff08;1&#xff09;在html目录下新建目录存放测试文件 &#xff08;2&#xff09;修改nginx.conf文件&#xff0c;在htttp模块中配置两个server模块分别对应两个IP &#xff08;3&am…...

[FPAG开发]使用Vivado创建第一个程序

1 打开Vivado软件&#xff0c;新建项目 选择一个纯英文路径 选择合适的型号 产品型号ZYNQ-7010xc7z010clg400-1ZYNQ-7020xc7z010clg400-2 如果型号选错&#xff0c;可以单击这里重新选择 2 创建工程源文件 可以看到文件创建成功 双击文件打开&#xff0c;插入代码 modul…...

使用 Python 在 NLP 中进行文本预处理

一、说明 自然语言处理 &#xff08;NLP&#xff09; 是人工智能 &#xff08;AI&#xff09; 和计算语言学的一个子领域&#xff0c;专注于使计算机能够理解、解释和生成人类语言。它涉及计算机和自然语言之间的交互&#xff0c;允许机器以对人类有意义和有用的方式处理、分析…...

[足式机器人]Part3机构运动微分几何学分析与综合Ch03-1 空间约束曲线与约束曲面微分几何学——【读书笔记】

本文仅供学习使用 本文参考&#xff1a; 《机构运动微分几何学分析与综合》-王德伦、汪伟 《微分几何》吴大任 Ch01-4 平面运动微分几何学 3.1 空间曲线微分几何学概述3.1.1 矢量表示3.1.2 Frenet标架 连杆机构中的连杆与连架杆构成运动副&#xff0c;该运动副元素的特征点或特…...

pytest框架快速进阶篇-pytest前置和pytest后置,skipif跳过用例

一、Pytest的前置和后置方法 1.Pytest可以集成unittest实现前置和后置 importunittestimportpytestclassTestCase(unittest.TestCase):defsetUp(self)->None:print(unittest每个用例前置)deftearDown(self)->None:print(unittest每个用例后置)classmethoddefsetUpClass…...

Python 基础语法 | 常量表达式,变量,注释,输入输出

常量和表达式 我们可以把 Python 当成一个计算器&#xff0c;来进行一些算术运算 print(1 2 - 3) # 0 print(1 2 * 3) # 7 print(1 2 / 3) # 1.6666666666666665注意&#xff1a; print 是一个 Python 内置的 函数可以使用 - * / () 等运算符进行算术运算&#xff0c;先…...

SQL | 分组数据

10-分组数据 两个新的select子句&#xff1a;group by子句和having子句。 10.1-数据分组 上面我们学到了&#xff0c;使用SQL中的聚集函数可以汇总数据&#xff0c;这样&#xff0c;我们就能够对行进行计数&#xff0c;计算和&#xff0c;计算平均数。 目前为止&#xff0c…...

软件测试技术之如何编写测试用例(6)

四、客户端兼容性测试 1、平台测试 市场上有很多不同的操作系统类型&#xff0c;最常见的有Windows、Unix、Macintosh、Linux等。Web应用系统的最终用户究竟使用哪一种操作系统&#xff0c;取决于用户系统的配置。这样&#xff0c;就可能会发生兼容性问题&#xff0c;同一个应…...

论文阅读——Adversarial Eigen Attack on Black-Box Models

Adversarial Eigen Attack on Black-Box Models 作者&#xff1a;Linjun Zhou&#xff0c; Linjun Zhou 攻击类别&#xff1a;黑盒&#xff08;基于梯度信息&#xff09;&#xff0c;白盒模型的预训练模型可获得&#xff0c;但训练数据和微调预训练模型的数据不可得&#xff…...

Linux链表操作全解析

Linux C语言链表深度解析与实战技巧 一、链表基础概念与内核链表优势1.1 为什么使用链表&#xff1f;1.2 Linux 内核链表与用户态链表的区别 二、内核链表结构与宏解析常用宏/函数 三、内核链表的优点四、用户态链表示例五、双向循环链表在内核中的实现优势5.1 插入效率5.2 安全…...

【2025年】解决Burpsuite抓不到https包的问题

环境&#xff1a;windows11 burpsuite:2025.5 在抓取https网站时&#xff0c;burpsuite抓取不到https数据包&#xff0c;只显示&#xff1a; 解决该问题只需如下三个步骤&#xff1a; 1、浏览器中访问 http://burp 2、下载 CA certificate 证书 3、在设置--隐私与安全--…...

k8s业务程序联调工具-KtConnect

概述 原理 工具作用是建立了一个从本地到集群的单向VPN&#xff0c;根据VPN原理&#xff0c;打通两个内网必然需要借助一个公共中继节点&#xff0c;ktconnect工具巧妙的利用k8s原生的portforward能力&#xff0c;简化了建立连接的过程&#xff0c;apiserver间接起到了中继节…...

用docker来安装部署freeswitch记录

今天刚才测试一个callcenter的项目&#xff0c;所以尝试安装freeswitch 1、使用轩辕镜像 - 中国开发者首选的专业 Docker 镜像加速服务平台 编辑下面/etc/docker/daemon.json文件为 {"registry-mirrors": ["https://docker.xuanyuan.me"] }同时可以进入轩…...

根据万维钢·精英日课6的内容,使用AI(2025)可以参考以下方法:

根据万维钢精英日课6的内容&#xff0c;使用AI&#xff08;2025&#xff09;可以参考以下方法&#xff1a; 四个洞见 模型已经比人聪明&#xff1a;以ChatGPT o3为代表的AI非常强大&#xff0c;能运用高级理论解释道理、引用最新学术论文&#xff0c;生成对顶尖科学家都有用的…...

Device Mapper 机制

Device Mapper 机制详解 Device Mapper&#xff08;简称 DM&#xff09;是 Linux 内核中的一套通用块设备映射框架&#xff0c;为 LVM、加密磁盘、RAID 等提供底层支持。本文将详细介绍 Device Mapper 的原理、实现、内核配置、常用工具、操作测试流程&#xff0c;并配以详细的…...

初学 pytest 记录

安装 pip install pytest用例可以是函数也可以是类中的方法 def test_func():print()class TestAdd: # def __init__(self): 在 pytest 中不可以使用__init__方法 # self.cc 12345 pytest.mark.api def test_str(self):res add(1, 2)assert res 12def test_int(self):r…...

AI病理诊断七剑下天山,医疗未来触手可及

一、病理诊断困局&#xff1a;刀尖上的医学艺术 1.1 金标准背后的隐痛 病理诊断被誉为"诊断的诊断"&#xff0c;医生需通过显微镜观察组织切片&#xff0c;在细胞迷宫中捕捉癌变信号。某省病理质控报告显示&#xff0c;基层医院误诊率达12%-15%&#xff0c;专家会诊…...

关于uniapp展示PDF的解决方案

在 UniApp 的 H5 环境中使用 pdf-vue3 组件可以实现完整的 PDF 预览功能。以下是详细实现步骤和注意事项&#xff1a; 一、安装依赖 安装 pdf-vue3 和 PDF.js 核心库&#xff1a; npm install pdf-vue3 pdfjs-dist二、基本使用示例 <template><view class"con…...

k8s从入门到放弃之HPA控制器

k8s从入门到放弃之HPA控制器 Kubernetes中的Horizontal Pod Autoscaler (HPA)控制器是一种用于自动扩展部署、副本集或复制控制器中Pod数量的机制。它可以根据观察到的CPU利用率&#xff08;或其他自定义指标&#xff09;来调整这些对象的规模&#xff0c;从而帮助应用程序在负…...