Flink源码之StreamTask启动流程
每个ExecutionVertex分配Slot后，JobMaster就会向Slot所在的TaskExecutor提交RPC请求执行Task，接口为TaskExecutorGateway::submitTask
CompletableFuture<Acknowledge> submitTask(TaskDeploymentDescriptor tdd, JobMasterId jobMasterId, @RpcTimeout Time timeout);
TaskDeploymentDescriptor 中包含当前Task的执行逻辑、Job信息、输入输出信息
submitTask 方法核心就是构造org.apache.flink.runtime.taskmanager.Task实例，该实例继承自Runnable接口，有个Thread成员变量，构造完成后就启动线程执行Task逻辑。
TaskExecutor::submitTask
Task.startTaskThread
Task.run
Task.doRun
Task::setupPartitionsAndGates //初始化Task的输入输出
RuntimeEnvironment::new //封装task执行上下文信息
Task::loadAndInstantiateInvokable //TaskInvokables实例化
StreamTask::newStreamTask::createRecordWriterDelegate //创建Writer,为每个StreamEdge创建一个WriterStreamTask::createStateBackend //创建StateBackend,一个task一个StateBackend实例StreamTask::createCheckpointStorageSubtaskCheckpointCoordinatorImpl::new
Task::restoreAndInvoke
TaskInvokable::restore
TaskInvokable::invoke //处理输入元素
TaskInvokable::cleanUp
Task的Invokable Class是在StreamGraph中添加Operator形成StreamNode时确定的，对不同的算子有不同的InvokableClass:
- SourceStreamTask.class //LegacySource算子
- SourceOperatorStreamTask //Source算子
- OneInputStreamTask.class //输入是一个算子
- TwoInputStreamTask:class //输入是两个算子
- MultipleInputStreamTask.class //输入有多个算子
以上这些类都继承自org.apache.flink.streaming.runtime.tasks.StreamTask
在调用TaskInvokable::restore时会执行：
StreamTask::restore
StreamTask::restoreInternal //创建OperatorChain
RegularOperatorChain::new
OperatorChain::new
OperatorChain::createOutputCollector
OperatorChain::createOperatorChain
OperatorChain::createOperator
StreamOperatorFactoryUtil.createOperator //创建Operator，在每个算子的StreamConfig中定义了每个Operator具体类型，比如StreamMap， StreamFlatMap
SimpleOperatorFactory::createStreamOperator //创建StreamOperator包装了用户函数，, StreamOperator包装了代码中用户函数，会调用用户函数中的open/close等生命周期函数AbstractUdfStreamOperator::setupAbstractStreamOperator::setup //设置用用自定义函数中的RuntimeContext成员变量StreamingRuntimeContext::new //StreamTask::init //子类做初始化，创建InputGate、StreamTaskInput、DataOutput及InputProcessor
StreamTask::restoreGatesStreamTask::createStreamTaskStateInitializerStreamTaskStateInitializerImpl::new //OperatorChain::initializeStateAndOpenOperators //调用每个Operator的initializeState和Open方法AbstractStreamOperator::initializeState StreamTaskStateInitializerImpl::streamOperatorStateContext //此时会创建keyedStatedBackend和operatorStateBackendStreamOperatorStateHandler::new //初始化StreamOperator的stateHandler成员变量，用于状态管理StreamOperatorStateHandler::initializeOperatorStateStateInitializationContextImpl::newAbstractUdfStreamOperator::initializeState//调用用户定义函数中的initializeState方法，可获取Operator StateStreamingFunctionUtils::restoreFunctionStateStreamingRuntimeContext::setKeyedStateStoreStreamOperator::open //调用getRuntimeContext().getState可获取keySate
StreamTask::invoke
StreamTask::runMailboxLoop
MailboxProcessor::runMailboxLoop
StreamTask::processInput
整个过程在StreamTask.java的注释中有说明:
* -- invoke()* |* +----> Create basic utils (config, etc) and load the chain of operators* +----> operators.setup()* +----> task specific init()* +----> initialize-operator-states()* +----> open-operators()* +----> run()* +----> finish-operators()* +----> close-operators()* +----> common cleanup* +----> task specific cleanup()
- 首先创建OperatorChain，依次创建出每个StreamOperator
- 调用Operator的setup方法，初始化StreamingRuntimeContext
- 调用子类init方法初始化
- 调用initializeState初始化每个算子的状态，此时会为每个StreamOperator创建keyedStatedBackend和operatorStateBackend，然后会调用用户定义函数中的initializeState方法，用于创建Operator State
- 调用算子的open方法，便于用户在自定义函数open中进行初始化，比如初始化keyState
- 调用processInput处理流中数据
SourceStreamTask重载了StreamTask::processInput，该方法中直接起一个线程调用SourceFunction::run方法。
OneInputStreamTask则不同，它重载了StreamTask的init方法，在init方法中创建了StreamOneInputProcessor
OneInputStreamTask::init
OneInputStreamTask::createCheckpointedInputGate
OneInputStreamTask::createDataOutput //创建StreamTaskNetworkOutput
OneInputStreamTask::createTaskInput //创建StreamTaskNetworkInput
StreamOneInputProcessor::new
在StreamTask::processInput则是调用InputProcessor::processInput不断读取数据进行处理
StreamOneInputProcessor::processInput
StreamTaskNetworkInput::emitNext(StreamTaskNetworkOutput)
AbstractStreamTaskNetworkInput::emitNext //循环不断从buffer中读取StreamElement
处理
AbstractStreamTaskNetworkInput::processElementStreamTaskNetworkOutput::emitRecord //调用operator的setKeyContextElement和processElementOneInputStreamOperator::setKeyContextElementAbstractStreamOperator::setKeyContextElement1AbstractStreamOperator::setCurrentKey //StreamOperatorStateHandler::setCurrentKey //设置状态当前keyInput::processElement //调用StreamOperator的processElement方法
以上Task从提交到起线程执行起来的整个过程，在初始化过程中为每个StreamOperator进行状态后端的初始化相当重要，后续处理流的过程中会使用这些状态后端存储管理状态。