NetworkStack提供了高效的网络I/O和反压控制

除了各个组件之间进行RPC通信之外，在Flink集群中TaskManager和TaskManager节点之间也会发生数据交换，尤其当用户提交的作业涉及Task实例运行在不同的TaskManager上时。Task实例之间的数据交换主要借助Flink中的NetworkStack实现。NetworkStack不仅提供了非常高效的网络I/O，也提供了非常灵活的反压控制。

 

1. NetworkStack整体架构

通过Netty协议实现的NetworkStack

Flink NetworkStack整体架构在不同的TaskManager之间建立TCP连接，而TCP连接则主要依赖Netty通信框架实现。Netty是一个NIO网络编程框架，可以快速开发高性能、高可靠性的网络服务器/客户端程序，能够极大简化TCP和UDP等网络编程。

流程举例：

TaskManager中会运行多个Task实例，例如在TaskManager 1中运行了Task A-1和Task A-2，在TaskManager 2中运行了Task B-1和Task B-2，Task A中从外部接入数据并处理后，会通过基于Netty构建的TCP连接发送到Task B中继续进行处理。整个数据传输过程主要基于Flink的NetworkStack框架进行。

 

上游数据流转逻辑：二进制buffer->ResultSubPartition队列->InputChannel

对于上游的Task A实例来讲，经过Operator处理后的数据，最终会通过RecordWriter组件写入网络栈，即算子输出的数据并不是直接写入网络，而是先将数据元素转换为二级制Buffer数据，并将Buffer缓存在ResultSubPartition队列中，接着写入下游Task对应的InputChannel。在上游的Task中会创建LocalBufferPool为数据元素申请对应Buffer的存储空间，且上游的Task会创建NettyServer作为网络连接服务端，并与下游Task内部的NettyClient之间建立网络连接。

 

 

下游Task数据接收逻辑：InputGate的InputChannel接收->StreamTaskInput取数据并处理（反序列化）->OperatorChain

• 对下游的Task实例来讲，会通过InputGate组件接收上游Task发送的数据，在InputGate中包含了多个InputChannel。InputChannel实际上是将Netty中Channel进行封装，数量取决于Task的并行度。
• 上游Task的ResultPartition会根据ChannelSelector选择需要将数据下发到哪一个InputChannel中，其实现类似Shuffe的数据洗牌操作。
• 在下游的Task实例中可以看出，InputGate中接收到的二进制数据，会转换为Buffer数据结构并存储到本地的Buffer队列中，最后被StreamTaskInput不断地从队列中拉取出来并处理。StreamTaskInput会将Buffer数据进行反序列化操作，将Buffer数据转换为StreamRecord并发送到OperatorChain中继续处理。

 

2. StreamTask内数据流转过程

流式作业中OperatorChain转为StreamTask

在ExecutionGraph调度和执行ExecutionVertex节点的过程中，会将OperatorChain提交到同一个Task实例中运行。如果被调度的作业为流式类型，则AbstractInvokable的实现类就为StreamTask。最终StreamTask会被TaskManager中的Task线程触发执行。

根据数据源不同，StreamTask分为两种类型：

1. 直接从外部源数据读取数据的SourceStreamTask和SourceReaderStreamTask；
2. 支持从网络或本地获取数据的OneInputStreamTask和TwoInputStreamTask；

 

以OneInputStreamTask为例，分析从Task层面介绍数据从网络接入并发送到OperatorChain中进行处理，接着通过Output组建输出到下游网络中的过程。

 

OneInputStreamTask包含一个StreamInputProcessor，用于对输入数据进行处理和输出。在StreamInputProcessor组件中包含StreamTaskInput、OperatorChain以及DataOutput三个组成部分。

 
task内部数据流转：StreamTaskNetworkIutput -> StreamTaskNetworkOutput -> OperatorChain中的HeaderOperator -> task实例算子->Output->下游算子...->RecordWriter->网络。详细过程如下：

1. StreamTaskInput从Task外部获取数据。

根据不同的数据来源，StreamTaskInput的实现主要分为从网络获取数据的StreamTaskNetworkInput和从外部系统获取数据的StreamTaskSourceInput。

2. DataOutput负责将StreamTaskInput接收的数据发送到当前Task实例的OperatorChain的HeadOperator中进行处理。

DataOutput主要有StreamTaskNetworkOutput（用于处理StreamTaskNetworkInput接收的数据）和StreamTaskSourceOutput（用于处理StreamTaskSourceInput接收的数据）两种实现。

3. HeaderOperator接收数据，算子开始接收数据并进行处理

OperatorChain负责将能够运行在同一个Task实例中的Operator连接起来，然后形成算子链，且算子链中HeaderOperator会暴露给StreamTask。当StreamTaskNetworkIutput接收到网络数据后，就会通过StreamTaskNetworkOutput组件将数据元素发送给OperatorChain中的HeaderOperator进行处理，此时Task实例中的算子就能够接收数据并进行处理了。

4. 上一个算子处理的数据会通过Output组件发送到下一个算子中继续处理

1. 在OperatorChain中，除了具有HeaderOperator之外，还包含了其他算子，这些算子会按照拓扑关系连接到HeaderOperator之后，每个算子之间的数据传输通过Output组件相连，即在OperatorChain中，上一个算子处理的数据会通过Output组件发送到下一个算子中继续处理。注意：DataOutput强调的是从外部接入数据到Task实例后再转发到HeaderOperator中，Output则更加强调算子链内部的数据传递。
2. Output组件的实现主要有ChainingOutput、BroadcastingOutputCollector、DirectedOutput和RecordWriterOutput等类型，它们最大的区别在于数据下发的方式不同，例如ChainingOutput代表直接向下游算子推送数据。

5. RecordWriterOutput中RecordWriter组件将数据发送到网络

经过算子链处理后的数据，需要发送到网络中供下游的Task实例继续处理，此时需要通过RecordWriterOutput完成数据的网络输出。RecordWriterOutput中包含了RecordWriter组件，用于将数据输出到网络中，下游Task实例就能通过StreamTaskInput组件从网络中获取数据，并继续传递到Task内部的算子链进行处理。

小结：
在StreamTask中接入数据，然后通过OperatorChain进行处理，再通过RecordWriterOutput发送到网络中，下游Task节点则继续从网络中获取数据并继续处理，最后组合这些Task节点就形成了整个Flink作业的计算拓扑。

注意：Task节点的数据输入也可以是本地类型，这种情况主要出现在Task实例被执行在同一台TaskManager时，数据不需要经过网络传输。