Flink StreamTask启动和执行源码分析

文章目录

前言

Flink的StreamTask的启动和执行是一个复杂的过程，涉及多个关键步骤。以下是StreamTask启动和执行的主要流程：

1. 初始化：StreamTask的初始化阶段涉及多个任务，包括Operator的配置、task特定的初始化以及初始化算子的State等。在这个阶段，Flink将业务处理函数抽象为operator，并通过operatorChain将业务代码串起来执行，以完成业务逻辑的处理。同时，还会调用具体task的init方法进行初始化。
2. 读取数据和事件：StreamTask通过mailboxProcessor读取数据和事件。
3. 运行业务逻辑：在StreamTask的beforeInvoke方法中，主要调用生成operatorChain并执行相关的业务逻辑。这些业务逻辑可能包括Source算子和map算子等，它们将被Chain在一起并在一个线程内同步执行。
4. 资源清理：在执行完业务逻辑后，StreamTask会进行关闭和资源清理的操作，这部分在afterInvoke阶段完成。

值得注意的是，从资源角度来看，每个TaskManager内部有多个slot，每个slot内部运行着一个subtask，即每个slot内部运行着一个StreamTask。这意味着StreamTask是由TaskManager（TM）部署并执行的本地处理单元。

总的来说，Flink的StreamTask启动和执行是一个由多个阶段和组件协同工作的过程，涉及数据的读取、业务逻辑的执行以及资源的清理等多个方面。这些步骤确保了StreamTask能够高效、准确地处理数据流，并满足实时计算和分析的需求。

StreamTask 部署启动

当 TaskExecutor 接收提交 Task 执行的请求，则调用：

TaskExecutor.submitTask(TaskDeploymentDescriptor tdd,JobMasterId jobMasterId,Time timeout){// 构造 Task 对象Task task =newTask(jobInformation, taskInformation,ExecutionAttemptId,AllocationId,SubtaskIndex,....);// 启动 Task 的执行
    task.startTaskThread();}

Task对象的构造方法

publicTask(.....){// 封装一个 Task信息对象 TaskInfo，（TaskInfo, JobInfo，JobMasterInfo）this.taskInfo =newTaskInfo(....);// 各种成员变量赋值......// 一个Task的执行有输入也有输出： 关于输出的抽象： ResultPartition 和ResultSubPartition（PipelinedSubpartition）
// 初始化 ResultPartition 和 ResultSubPartitionfinalResultPartitionWriter[] resultPartitionWriters =
shuffleEnvironment.createResultPartitionWriters(....);this.consumableNotifyingPartitionWriters =ConsumableNotifyingResultPartitionWriterDecorator.decorate(....);// 一个Task的执行有输入也有输出： 关于输入的抽象： InputGate 和 InputChannel（从上有
一个Task节点拉取数据）
// InputChannel 可能有两种实现： Local Remote// 初始化 InputGate 和 InputChannelfinalIndexedInputGate[] gates = shuffleEnvironment.createInputGates(.....);// 初始化一个用来执行 Task 的线程，目标对象，就是 Task 自己
executingThread =newThread(TASK_THREADS_GROUP,this, taskNameWithSubtask);}

Task 线程启动

Task 的启动，是通过启动 Task 对象的内部 executingThread 来执行 Task 的，具体逻辑在 run 方法中：

privatevoiddoRun(){// 1、先更改 Task 的状态： CREATED ==> DEPLOYINGtransitionState(ExecutionState.CREATED,ExecutionState.DEPLOYING);// 2、准备 ExecutionConfigfinalExecutionConfig executionConfig =
serializedExecutionConfig.deserializeValue(userCodeClassLoader);// 3、初始化输入和输出组件, 拉起 ResultPartition 和 InputGatesetupPartitionsAndGates(consumableNotifyingPartitionWriters,
inputGates);// 4、注册 输出for(ResultPartitionWriter partitionWriter :
consumableNotifyingPartitionWriters){
taskEventDispatcher.registerPartition(partitionWriter.getPartitionId());}//5、初始 环境对象 RuntimeEnvironment, 包装在 Task 执行过程中需要的各种组件
Environment env =newRuntimeEnvironment(jobId, vertexId, executionId,....);// 6、初始化 调用对象// 两种最常见的类型： SourceStreamTask、OneInputStreamTask、TwoInputStreamTask// 父类： StreamTask// 通过反射实例化 StreamTask 实例(可能的两种情况： SourceStreamTask，OneInputStreamTask)AbstractInvokable invokable =loadAndInstantiateInvokable(userCodeClassLoader, nameOfInvokableClass, env);// 7、先更改 Task 的状态： DEPLOYING ==> RUNNINGtransitionState(ExecutionState.DEPLOYING,ExecutionState.RUNNING);// 8、真正把 Task 启动起来了
invokable.invoke();// 9、StreamTask 需要正常结束，处理 buffer 中的数据for(ResultPartitionWriter partitionWriter :
consumableNotifyingPartitionWriters){if(partitionWriter !=null){
partitionWriter.finish();}}//10、先更改 Task 的状态： RUNNING==>FINISHEDtransitionState(ExecutionState.RUNNING,ExecutionState.FINISHED);

StreamTask 初始化

StreamTask 初始化指的就是 SourceStreamTask 和 OneInputStreamTask 的实例对象的构建！Task 这个类，只是一个笼统意义上的 Task，就是一个通用 Task 的抽象，不管是批处理的，还是流式处理的，不管是 源Task， 还是逻辑处理 Task， 都被抽象成 Task 来进行调度执行！

privateSourceStreamTask(Environment env,Object lock)throwsException{super(
                env,null,FatalExitExceptionHandler.INSTANCE,StreamTaskActionExecutor.synchronizedExecutor(lock));this.lock =Preconditions.checkNotNull(lock);this.sourceThread =newLegacySourceFunctionThread();getEnvironment().getMetricGroup().getIOMetricGroup().setEnableBusyTime(false);}@Overrideprotectedvoidinit(){// we check if the source is actually inducing the checkpoints, rather// than the triggerSourceFunction<?> source = mainOperator.getUserFunction();if(source instanceofExternallyInducedSource){
            externallyInducedCheckpoints =true;ExternallyInducedSource.CheckpointTrigger triggerHook =newExternallyInducedSource.CheckpointTrigger(){@OverridepublicvoidtriggerCheckpoint(long checkpointId)throwsFlinkException{// TODO - we need to see how to derive those. We should probably not// encode this in the// TODO -   source's trigger message, but do a handshake in this task// between the trigger// TODO -   message from the master, and the source's trigger// notificationfinalCheckpointOptions checkpointOptions =CheckpointOptions.forConfig(CheckpointType.CHECKPOINT,CheckpointStorageLocationReference.getDefault(),
                                            configuration.isExactlyOnceCheckpointMode(),
                                            configuration.isUnalignedCheckpointsEnabled(),
                                            configuration.getAlignedCheckpointTimeout().toMillis());finallong timestamp =System.currentTimeMillis();finalCheckpointMetaData checkpointMetaData =newCheckpointMetaData(checkpointId, timestamp, timestamp);try{SourceStreamTask.super.triggerCheckpointAsync(
                                                checkpointMetaData, checkpointOptions).get();}catch(RuntimeException e){throw e;}catch(Exception e){thrownewFlinkException(e.getMessage(), e);}}};((ExternallyInducedSource<?,?>) source).setCheckpointTrigger(triggerHook);}getEnvironment().getMetricGroup().getIOMetricGroup().gauge(MetricNames.CHECKPOINT_START_DELAY_TIME,this::getAsyncCheckpointStartDelayNanos);
        recordWriter.setMaxOverdraftBuffersPerGate(0);}

StreamTask 执行

核心步骤如下：

publicfinalvoidinvoke()throwsException{// Task 正式工作之前beforeInvoke();// Task 开始工作: 针对数据执行正儿八经的逻辑处理runMailboxLoop();// Task 要结束afterInvoke();// Task 最后执行清理cleanUpInvoke();}

总结一下要点：

• 在 beforeInvoke() 中，主要是初始化 OperatorChain，然后调用 init() 执行初始化，然后恢复状态，更改 Task 自己的状态为 isRunning = true
• 在 runMailboxLoop() 中，主要是不停的处理 mail，这里是 FLink-1.10 的一项改进，使用了mailbox 模型来处理任务
• 在 afterInvoke() 中，主要是完成 Task 要结束之前需要完成的一些细节，比如，把 Buffer 中还没flush 的数据 flush 出来
• 在 cleanUpInvoke() 中，主要做一些资源的释放，执行各种关闭动作：set false，interrupt， shutdown，close，cleanup，dispose 等