Flink源码解析(三)——从RM与TM的心跳交互分析Flink心跳机制

database

0. 说明

基于Flink 1.12

1. 背景知识

1.1 Actor模型

Flink底层RPC是通过AKKA实现的,AKKA是基于Actor模型实现的框架。下面,将大致介绍一下actor模型。

在Actor模型中,一切事物都是actor,一个actor是一个基本的计算单元,每个actor是完全隔离的,不会共享内存,也就不会有共享数据带来的并发问题;它们是自己维护自身的状态,该状态不会被其他actor直接修改。

整体模型大致是:多个actor同时运行,每个actor接收消息,并根据消息做出相应的反应。消息本身是通过异步的形式发送给actor的,消息会被存储在一个叫做“邮箱(mailbox)”的地方,actor会顺序的处理收到的信息,避免锁的使用。从描述可以了解到actor模型中,消息的发送者和已发送消息解耦,是以并发的形式处理数据的。

1.2 RPC

RPC作用是让远程调用像本地调用,封装调用的细节。

Flink定义了各个组件的Gateway,通过回调的方式隐藏实现细节,将业务本身和通信解绑了,方便RPC调用。目前,Flink的RPC请求的底层通信是通过AKKA的实现的。

1.2.1 RPC相关的接口

  • RPCGateway

    所有Rpc组件的网关,定义了各组件的Rpc接口,提供了获取地址和主机名的功能;

  • RPCEndpoint

    RPCEndpoint是Flink RPC调用的基类,所有具有分布式调用能力的组件都需要继承该接口。

  • RpcService

    RPC服务提供者,提供开始、停止服务等功能,以及提供远程功能;

    三者的关系如下:

    public abstract class RpcEndpoint implements RpcGateway, AutoCloseableAsyn{

    //启动sever和获取RPC Gateway

    /** RPC service to be used to start the RPC server and to obtain rpc gateways. */

    private final RpcService rpcService;

    //RpcServer用于启动和连接到RpcEndpoint, 连接到rpc服务器将返回一个RpcGateway,为RpcService提供RPC服务/连接远程Server

    /** Interface to access the underlying rpc server. */

    protected final RpcServer rpcServer;

    }

2. Flink心跳机制

2.1 核心接口

2.1.1 HeartbeatTarget

是可以发送心跳和请求心跳相应组件接口,是对具备心跳能力对象的一种抽象。

HeartbeatTarget的函数具备以下两种动作:

  • receiveHeartbeat 向某个节点发送心跳响应,其参数heartbeatOrigin就是该节点;
  • requestHeartbeat

    要求某个节点发送心跳信息,其参数requestOrigin就是心跳信息上报的节点。

2.1.2 HeartbeatMonitor

HeartbeatMonitor管理HeartbeatTarget的心跳状态。当在指定时间内未收到心跳信息时,monitor将会通知对应的HeartbeatListener,收到心跳信息后会重置其定时器。其工厂接口如下:

     HeartbeatMonitor<O> createHeartbeatMonitor(

ResourceID resourceID,

HeartbeatTarget<O> heartbeatTarget,

ScheduledExecutor mainThreadExecutor,

HeartbeatListener<?, O> heartbeatListener, //用于处理心跳信息

long heartbeatTimeoutIntervalMs);

2.1.3 HeartbeatListener

HeartbeatListener是和HeartbeatManager交互的接口,Flink的业务的处理逻辑需要继承该接口以处理心跳结果,其三个回调函数如下:

  • notifyHeartbeatTimeout :通知心跳超时;
  • reportPayload:处理节点发来的Payload载荷;
  • retrievePayLoad:获取对某节点发下一次心跳请求的Payload载荷

2.1.4 HeartbeatManager

心跳的管理者,用于开始/停止对HeartbeatTarget的心跳监控,以及会处理某个节点的心跳超时。

HeartbeatManager继承了HeartbeatTarget,其具有了HeartbeatTarget的函数功能以外,该接口还有以下四种函数:

  • monitorTarget

    开始监控HeartbeatTarget,HeartbeatTarget的心跳超时后,将会通知HeartbeatListener;

  • unmonitorTarget 停止监控某节点;
  • stop 停止HeartbeatManager;
  • getLastHeartbeatFrom 返回特定节点最近一次心跳信息;

核心接口交互的大致过程:HeartbeatManager将HeartbeatTarget放入到监控列表中,当心跳超时时,HeartbeatMonitor回通知HeartbeatListener处理,通过对HeartbeatListener的实现,完成相关处理心跳超时的逻辑。

2.2. 核心接口的实现

下面通过分析1.3.1中核心接口的实现类,来具体分析心跳处理的过程。

2.2.1 HearbeatManagerImpl

该manager维护了一个heartbeat 的监控对象(HeartbeatMonitor)和资源ID信息,当收到新的心跳信息是,monitor对象将会被更新;心跳超时时,将会通知HeartbeatListenter对象。

public class HeartbeatManagerImpl<I, O> implements HeartbeatManager<I, O> {

//心跳间隔

/** Heartbeat timeout interval in milli seconds. */

private final long heartbeatTimeoutIntervalMs;

//心跳

/** Heartbeat listener with which the heartbeat manager has been associated. */

private final HeartbeatListener<I, O> heartbeatListener;

//使用一个map存放资源-心跳的monitor信息,其monitorTarget方法就是将对应信息放入该map中

/** Map containing the heartbeat monitors associated with the respective resource ID. */

private final ConcurrentHashMap<ResourceID, HeartbeatMonitor<O>> heartbeatTargets;

/** Running state of the heartbeat manager. */

protected volatile boolean stopped;

HearbeatManagerImpl实现的主要函数有:

  • monitorTarget将一个节点加入监控列表中

    该方法会根据ResourceID和HeartbeatTarget生成一个HeartbeatMonitor对象,然后将resourceID和该对象组成KV的形式放入heartbeatTargets中。

  • requestHeartbeat

    心跳请求方调用requestHeartbeat要求上报一个心跳信息,该请求会通过RPC异步调用到心跳的上报方(HearbeatManagerImpl的创建者)的requestHeartbeat,以要求上报方向requestOrigin节点发起一个心跳响应。具体过程如下:

    • requestHeartbeat会记录下这个请求时间点,然后取消超时,重新创建一个ScheduleFuture去判断requestOrigin的心跳是否超时。后续若是超时了,则将heartbeatMonitor的state置为timeout状态,若是特定时间内requestOrigin响应了,则ScheduleFuture取消,monitor的状态依旧为RUNNING。
    • 调用heartbeatListener#reportPayload处理心跳信息,其具体过程依据具体的实现。
    • 最后调用receiveHearbeat函数,响应一个心跳给请求方。

2.2.2 HeartbeatManagerSenderImpl

继承于HearbeatManagerImpl,HeartbeatManagerSenderImpl向其监控的heartbeatTarget对象请求心跳的响应,属于主动触发心跳请求。实现了Runnable接口,在其run方法中,会遍历heartbeatMonitor,通过requestHeartbeat()方法向节点获取心跳信息。

  public class HeartbeatManagerSenderImpl<I, O> extends HeartbeatManagerImpl<I, O> implements Runnable {

@Override

public void run() {

if (!stopped) {

log.debug("Trigger heartbeat request.");

for (HeartbeatMonitor<O> heartbeatMonitor : getHeartbeatTargets().values()) {

requestHeartbeat(heartbeatMonitor);

}

// 周期性调度,事件周期可配

getMainThreadExecutor().schedule(this, heartbeatPeriod, TimeUnit.MILLISECONDS);

}

}

// 主动发起心跳请求

private void requestHeartbeat(HeartbeatMonitor<O> heartbeatMonitor) {

O payload = getHeartbeatListener().retrievePayload(heartbeatMonitor.getHeartbeatTargetId());

final HeartbeatTarget<O> heartbeatTarget = heartbeatMonitor.getHeartbeatTarget();

// 调用Target的 requestHeartbeat函数

heartbeatTarget.requestHeartbeat(getOwnResourceID(), payload);

}

}

2.2.3 HeartbeatMonitorImpl

HeartbeatMonitor管理心跳目标,它在初始化会启动一个ScheduledExecutor。

  • timeout超时会通知heartbeatListener执行响应的超时逻辑;
  • 在规定时间内收到心跳信息,会重置ScheduledExecutor,重新开始;

public class HeartbeatMonitorImpl<O> implements HeartbeatMonitor<O>, Runnable {

/** Resource ID of the monitored heartbeat target. */

private final ResourceID resourceID; //监控的资源ID

/** Associated heartbeat target. */

private final HeartbeatTarget<O> heartbeatTarget; //心跳目标

private final ScheduledExecutor scheduledExecutor;

/** Listener which is notified about heartbeat timeouts. */

private final HeartbeatListener<?, ?> heartbeatListener;

HeartbeatMonitorImpl(

ResourceID resourceID,

HeartbeatTarget<O> heartbeatTarget,

ScheduledExecutor scheduledExecutor,

HeartbeatListener<?, O> heartbeatListener,

long heartbeatTimeoutIntervalMs) {

this.resourceID = Preconditions.checkNotNull(resourceID);

this.heartbeatTarget = Preconditions.checkNotNull(heartbeatTarget);

this.scheduledExecutor = Preconditions.checkNotNull(scheduledExecutor);

this.heartbeatListener = Preconditions.checkNotNull(heartbeatListener);

Preconditions.checkArgument(

heartbeatTimeoutIntervalMs > 0L,

"The heartbeat timeout interval has to be larger than 0.");

this.heartbeatTimeoutIntervalMs = heartbeatTimeoutIntervalMs;

lastHeartbeat = 0L;

//初始化的时候,就启动一个定时任务

resetHeartbeatTimeout(heartbeatTimeoutIntervalMs);

}

@Override

public void run() {

// The heartbeat has timed out if we"re in state running

if (state.compareAndSet(State.RUNNING, State.TIMEOUT)) {

//通知heartbeatListener处理

heartbeatListener.notifyHeartbeatTimeout(resourceID);

}

}

void resetHeartbeatTimeout(long heartbeatTimeout) {

if (state.get() == State.RUNNING) {

cancelTimeout();

//重新开启新的定时任务

futureTimeout =

scheduledExecutor.schedule(this, heartbeatTimeout, TimeUnit.MILLISECONDS);

// Double check for concurrent accesses (e.g. a firing of the scheduled future)

if (state.get() != State.RUNNING) {

cancelTimeout();

}

}

}

}

2.2.4 HeartbeatServices

HeartbeatServices为所有需要心跳服务的创建heartbeat receivers and heartbeat senders。

  public class HeartbeatServices {

/**

* 创建 heartbeat receivers

* Creates a heartbeat manager which does not actively send heartbeats.

*/

public <I, O> HeartbeatManager<I, O> createHeartbeatManager(

ResourceID resourceId,

HeartbeatListener<I, O> heartbeatListener,

ScheduledExecutor mainThreadExecutor,

Logger log) {

return new HeartbeatManagerImpl<>(

heartbeatTimeout, resourceId, heartbeatListener, mainThreadExecutor, log);

}

/**

* 创建 heartbeat sender

* Creates a heartbeat manager which actively sends heartbeats to monitoring targets.

*/

public <I, O> HeartbeatManager<I, O> createHeartbeatManagerSender(

ResourceID resourceId,

HeartbeatListener<I, O> heartbeatListener,

ScheduledExecutor mainThreadExecutor,

Logger log) {

return new HeartbeatManagerSenderImpl<>(

heartbeatInterval,

heartbeatTimeout,

resourceId,

heartbeatListener,

mainThreadExecutor,

log);

}

// 从配置文件配置心跳间隔时间和心跳超时时间

//两者的关系 0 < 心跳间隔时间 < 心跳超时时间

public static HeartbeatServices fromConfiguration(Configuration configuration) {

long heartbeatInterval = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_INTERVAL);

long heartbeatTimeout = configuration.getLong(HeartbeatManagerOptions.HEARTBEAT_TIMEOUT);

return new HeartbeatServices(heartbeatInterval, heartbeatTimeout);

}

}

3. RM和TM的交互

3.1. 总述

在一个Flink集群中只有一个ResourceManager(RM),和一个或多个TaskManager(TM)。两者的交互过程为:TM启动时会向RM注册,注册成功之后,RM会主动要求TM上报心跳信息。通过RM和TM的心跳信息,双方知道对方是否存活。

在2.2.4小节总,我们知道HeartbeatManagerSenderImpl属于Sender,HeartbeatManagerImpl属于Receiver。sender要对心跳目标上报心跳信息,receiver收到信息请求后返回一个response。

3.2. 初始化过程

3.2.1 ResourceManager

  • RM启动

  public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>

extends FencedRpcEndpoint<ResourceManagerId>

implements ResourceManagerGateway, LeaderContender {

// RM启动时运行的方法

@Override

public final void onStart() throws Exception {

try {

// 启动RMServices

startResourceManagerServices();

} catch (Throwable t) {

final ResourceManagerException exception =

new ResourceManagerException(

String.format("Could not start the ResourceManager %s", getAddress()),

t);

onFatalError(exception);

throw exception;

}

}

}

  • leaderElectionService#start方法

    leaderElectionService#start方法有多个实现,其中,主要是DefaultLeaderElectionService和StandaloneLeaderElectionService,前者是依赖外部组价的,这里我们以standalone模式分析。

    在standalone模式下,Flink集群中的leader是通过配置文件配置,所以在调用启动leader选举方法时,会直接将leadership赋给指定的节点,在赋予leadership角色过程会初始化心跳服务,大致的流程如下:

    StandaloneLeaderElectionService#start

    |

    ResourceManager#grantLeadership

    |

    ResourceManager#tryAcceptLeadership

    |

    ResourceManager#startServicesOnLeadership //其具体实现如下

    private void startServicesOnLeadership() {

    //启动心跳服务

    startHeartbeatServices();

    //slotManager是RM中管理slot的组件,其具体过程后续博客分析

    slotManager.start(getFencingToken(), getMainThreadExecutor(), new ResourceActionsImpl());

    //周期性判断是否存在未满足的slot请求

    onLeadership();

    }

    启动心跳服务,就是创建分别创建了taskManagerHeartbeatManager和jobManagerHeartbeatManager用于RM和TM、RM和JM的心跳服务

      private void startHeartbeatServices() {

taskManagerHeartbeatManager =

heartbeatServices.createHeartbeatManagerSender(

resourceId,

new TaskManagerHeartbeatListener(),

getMainThreadExecutor(),

log);

jobManagerHeartbeatManager =

heartbeatServices.createHeartbeatManagerSender(

resourceId,

new JobManagerHeartbeatListener(),

getMainThreadExecutor(),

log);

}

结合2.2.2小节,RM在心跳服务在和TM与JM的心跳过程中,充当的是请求心跳请求的发起方,即RM是主动去拉取心跳信息的。

3.2.2 TaskExecutor

TaskExecutor在创建时,就初始化了心跳组件。

  public TaskExecutor(

RpcService rpcService,

TaskManagerConfiguration taskManagerConfiguration,

HighAvailabilityServices haServices,

TaskManagerServices taskExecutorServices,

ExternalResourceInfoProvider externalResourceInfoProvider,

HeartbeatServices heartbeatServices,

TaskManagerMetricGroup taskManagerMetricGroup,

@Nullable String metricQueryServiceAddress,

BlobCacheService blobCacheService,

FatalErrorHandler fatalErrorHandler,

TaskExecutorPartitionTracker partitionTracker,

BackPressureSampleService backPressureSampleService) {

//创建HeartbeatManagerImpl,对JM的心跳进行相应

this.jobManagerHeartbeatManager =

createJobManagerHeartbeatManager(heartbeatServices, resourceId);

// 创建HeartbeatManagerImpl,对RM的心跳进行相应

this.resourceManagerHeartbeatManager =

createResourceManagerHeartbeatManager(heartbeatServices, resourceId);

}

3.3 TM向RM注册过程

3.3.1 流程图

  • TaskExecutor的启动过程如下:

  TaskExecutor#onStart

|

TaskExecutor#startTaskExecutorServices

|

StandaloneLeaderRetrievalService#start //以standalone模式分析

|

|//在standalone模式下,已知晓JobManager的地址,会直接去链接RM

TaskExecutor.ResourceManagerLeaderListener#notifyLeaderAddress

|

TaskExecutor#notifyOfNewResourceManagerLeader

|

TaskExecutor#reconnectToResourceManager

|

|//在该方法中会主动调用TaskExecutorToResourceManagerConnection类的start方法去链接RM

TaskExecutor#connectToResourceManager

|

| //在该函数的createNewRegistration方法中的回调函数,处理注册成功后的逻辑

RegisteredRpcConnection#start

|

|//z在该方法中会先链接RM,然后连接成功后发起注册请求

RetryingRegistration#startRegistration

|

RetryingRegistration#register

|

TaskExecutorToResourceManagerConnection#invokeRegistration

到此,TM向RM发起了注册,通过AKKA RPC,请求来到了RM中。

  • ResourceManager处理逻辑

  ResourceManager#registerTaskExecutor

|

|// 该方法的返回值是RegistrationResponse,在该方法中会将调用taskManagerHeartbeatManager.monitorTarget,监控节点的心跳信息

ResourceManager#registerTaskExecutorInternal

|

return new TaskExecutorRegistrationSuccess(

registration.getInstanceID(), resourceId, clusterInformation)

  • 注册成功后TaskExecutor的逻辑

  //注册成功后将会走start方法中createNewRegistration创建registration时的回调函数

RegisteredRpcConnection#start

|

TaskExecutorToResourceManagerConnection#onRegistrationSuccess

|

TaskExecutor#onRegistrationSuccess

|

| //和RM建立联系,开始监控RM

TaskExecutor#establishResourceManagerConnection

|

resourceManagerHeartbeatManager#monitorTarget

3.3.2 具体分析

下面主要分析心跳交互过程。

  • TaskExecutor的启动过程

    • 启动TaskExecutor服务第一步会与RM取得联系,并在此注册一个ResourceManagerLeaderListener用来监控RM leader的变化;
    • 获取RM leader的地址信息后,会调用listener的notifyLeaderAddress方法,该方法会异步的调用notifyOfNewResourceManagerLeader方法;
    • 在notifyOfNewResourceManagerLeader方法中,会根据了leader的地址调用reconnectToResourceManager方法去链接RM;
    • 通过执行resourceManagerConnection#start()去和RM建立链接,该方法是调用其父类RegisteredRpcConnection的start()方法,在该方法中,会定义链接成功的处理逻辑,具体代码如下:

    public void start() {

checkState(!closed, "The RPC connection is already closed");

checkState(

!isConnected() && pendingRegistration == null,

"The RPC connection is already started");

//会在创建newRegistration时,定义链接成功后处理逻辑

final RetryingRegistration<F, G, S, R> newRegistration = createNewRegistration();

if (REGISTRATION_UPDATER.compareAndSet(this, null, newRegistration)) {

// 启动注册

newRegistration.startRegistration();

} else {

// concurrent start operation

newRegistration.cancel();

}

}

  • RetryingRegistration#startRegistration方法中,我们可以看到在TM向RM注册之前,会尝试链接RM,当链接成功之后才会TM才会向RM注册;

在此过程中有好多回调,需要慢慢的体会。

  • ResourceManager处理逻辑

    通过RPC调用,请求来到了RM中,RM的处理具体如下:

    • 调用ResourceManager#registerTaskExecutor方法,在方法RM会首先去链接TM,若是成功了,会注册一个新的TM;
    • 调用来到ResourceManager#registerTaskExecutorInternal中,在该方法中,若是该TM的ID已经存在则删除,重新以KV的形式将注册信息放入到RM的TM注册列表中;
    • 通过taskManagerHeartbeatManager#monitorTarget监控TM,并返回一个注册成功信息;

  • TM的处理过程

    注册成功之后,调用回到TaskExecutor。

    • 回调TaskExecutorToResourceManagerConnection#onRegistrationSuccess方法;
    • 通过异步的形式,和RM建立链接并监控RM;
    • 通过resourceManagerHeartbeatManager.monitorTarget 把RM注册到TM中;

3.4 TM和RM的心跳过程

3.4.1 RM请求心跳的过程

由RM的初始化的分析,我们了解到,RM会主动要求TM上报心跳,其过程如下:

  // 在该该方法中会创建一个HeartbeatManagerSenderImpl

ResourceManager#startHeartbeatServices

|

| //这里会一步步调用构造函数中,在该构造函数中,会将心跳检查加入周期性任务列表中

HeartbeatManagerImpl

|

| //在任务启动时,会调用HeartbeatManagerSenderImpl的run方法,在该方法中会循环遍历HeartbeatMonitor,通过requestHeartbeat要求target上报心跳信息

HeartbeatManagerSenderImpl#run

|

| //该调用会跑到ResourceManager#TaskManagerHeartbeatListener中,这里返回为null是因为RM不是任何组件的receiver,即不会有组件向RM请求心跳信息,并要求其返回心跳。

getHeartbeatListener().retrievePayload

|

| //这里会调用TM向RM注册时指定的requestHeartbeat

heartbeatTarget.requestHeartbeat

|

| //ResouceManager#registerTaskExecutorInternal

taskExecutorGateway.heartbeatFromResourceManager

3.4.2 TM处理心跳请求

通过RPC调用,请求来到了TM中,其过程如下:

  TaskExecutor#heartbeatFromResourceManager

|

HeartbeatManagerImpl#requestHeartbeat

|

HeartbeatMonitorImpl#reportHeartbeat

|

| //在该方法中,判断若是running则会取消之前的Timeout定时任务ScheduledFuture,重新开始检查是否timeout超时的定时任务。

HeartbeatMonitorImpl#resetHeartbeatTimeout

|

| //因为从RM发来的请求中heartbeatPayload为null,则TM直接走返回心跳反应的流程

HeartbeatMonitorImpl#reportHeartbea->heartbeatTarget.receiveHeartbeat

|

| //这里生成TM的心跳信息,包括slot信息

TaskExecutor.ResourceManagerHeartbeatListener#retrievePayload

|

| //通过在TM向RM注册过程中定义的receiveHeartbeat方法来实现调用RM中方法

TaskExecutor#establishResourceManagerConnection-> resourceManagerGateway.heartbeatFromTaskManager

3.4.3 RM处理周期性心跳信息过程

RM收到TM的心跳信息,主要做了两件事:重置RM的Monitor线程;解析TM上报信息

  ResourceManager#heartbeatFromTaskManager

|

HeartbeatManagerImpl#reportHeartbeat

|

| //和TM一样,重置了monitor线程

HeartbeatManagerImpl#reportHeartbeat->reportHeartbeat

|

| //在该方法中处理上报的slot信息,

ResourceManager.TaskManagerHeartbeatListener#reportPayload

周期性心跳的具体分析过程见上述流程中的注释。

4. 参考文章

  • 从TimeoutException看Flink的心跳机制
  • 弄清Flink1.8的远程过程调用(RPC)
  • AKKA ~ 概念篇
  • Actor并发模型

以上是 Flink源码解析(三)——从RM与TM的心跳交互分析Flink心跳机制 的全部内容, 来源链接: utcz.com/z/535772.html

回到顶部