网络流控的概念与背景

分析一个简单的 Flink 流任务，下图是一个简单的 Flink 流任务执行图：任务首先从 Kafka 中读取数据、 map 算子对数据进行转换、keyBy 按照指定 key 对数据进行分区（相同 key 的数据经过 keyBy 后分到同一个 subtask 实例中），keyBy 后对数据接着进行 map 转换，然后使用 Sink 将数据输出到外部存储。

Flink任务简单示例图

众所周知，在大数据处理中，无论是批处理还是流处理，单点处理的性能总是有限的，我们的单个 Job 一般会运行在多个节点上，多个节点共同配合来提升整个系统的处理性能。图中，任务被切分成 4 个可独立执行的 subtask（ A0、A1、B0、B1），在数据处理过程中，就会存在 shuffle（数据传输）的过程。例如，subtask A0 处理完的数据经过 keyBy 后发送到 subtask B0、B1 所在节点去处理。

流控-存在的问题

• 如果 Receive Buffer 是有界的，这时候新到达的数据就只能被丢弃掉了。
• 如果 Receive Buffer 是无界的，Receive Buffer 会持续的扩张，最终会导致 Consumer 的内存耗尽，OOM，程序挂掉。

流控-静态限速

• 事先无法预估 Consumer 到底能承受多大的速率
• Consumer 的承受能力通常会动态地波动

流控-动态反馈

• 负反馈：接受速率小于发送速率时发生，告知 Producer 降低发送速率
• 正反馈：发送速率小于接收速率时发生，告知 Producer 可以把发送速率提上来

如下图所示，假如我们的 Job 分为 Task A、B、C，Task A 是 Source Task、Task B 处理数据、Task C 是 Sink Task。假如 Task C 由于各种原因吞吐量降低，会将负载信息反馈给 Task B，Task B 会降低向 Task C 发送数据的速率，此时如果 Task B 如果还是一直从 Task A 读取数据，那么按照同样的道理，数据会把 Task B 的 Send Buffer 和 Receive Buffer 撑爆，又会出现上面描述的问题。所以，当 Task B 的 Send Buffer 和 Receive Buffer 被用完后，Task B 会用同样的原理将负载信息反馈给 Task A，Task A 收到 Task B 的负载信息后，会降低给 Task B 发送数据的速率，以此类推。

如果下游 Task C 的负载能力提升后，会及时反馈给 Task B，于是 Task B 会提升往 Task C 发送数据的速率，Task B 又将负载提升的信息反馈给 Task A，Task A 就会提升从 Source 端读取数据的速率，从而提升整个系统的负载能力。

简单的 3 个 Task 图示

上面这个流程，就是 Flink 动态限流（反压机制）的简单描述。我们可以看到 Flink 的反压其实是从下游往上游传播的，一直往上传播到 Source Task 后，Source Task 最终会降低从 Source 端读取数据的速率。

Flink TCP-based 反压机制（ before V1.5 版本）

Flink V1.5 之前版本为什么没有 feedback 机制

Flink V1.5 以前版本之所以没有 feedback 机制，是因为 TCP 天然具备 feedback 流控机制，Flink 基于它来实现反压。

下图是 Flink 的 TaskManager 之间网络传输的数据流向：

Flink 网络传输的数据流向

图中，我们可以看到 TaskManager A 给 TaskManager B 发送数据，TaskManager A 做为 Producer，TaskManager B 做为 Consumer。Producer 端的 Operator 实例会产生数据，最后通过网络发送给 Consumer 端的 Operator 实例。

Producer 端 Operator 实例生产的数据首先缓存到 TaskManager 内部的 Network Buffer。Network 依赖 Netty 来做通信，Producer 端的 Netty 内部有 ChannelOutbound Buffer，Consumer 端的 Netty 内部有 ChannelInbound Buffer。Netty 最终还是要通过 Socket 发送网络请求，Socket 这一层也会有 Buffer，Producer 端有 Send Buffer，Consumer 端有 Receive Buffer。

总结一下，现在有两个 TaskManager A、B，TaskManager A 中 Producer Operator 处理完的数据由 TaskManager B 中 Consumer Operator 处理。那么 Producer Operator 处理完的数据是怎么到达 Consumer Operator 的？首先 Producer Operator 从自己的上游或者外部数据源读取到数据后，对一条条的数据进行处理，处理完的数据首先输出到 Producer Operator 对应的 Network Buffer 中。Buffer 写满或者超时后，就会触发将 Network Buffer 中的数据拷贝到 Producer 端 Netty 的 ChannelOutbound Buffer，之后又把数据拷贝到 Socket 的 Send Buffer 中，这里有一个从用户态拷贝到内核态的过程，最后通过 Socket 发送网络请求，把 Send Buffer 中的数据发送到 Consumer 端的 Receive Buffer。数据到达 Consumer 端后，再依次从 Socket 的 Receive Buffer 拷贝到 Netty 的 ChannelInbound Buffer，再拷贝到 Consumer Operator 的 Network Buffer，最后 Consumer Operator 就可以读到数据进行处理了。

这就是两个 TaskManager 之间的数据传输过程，我们可以看到发送方和接收方各有三层的 Buffer。

TCP 流控机制

TCP Packet

TCP流控: 滑动窗口

Job 提交执行流程

示例: WindowWordCount

编译阶段: 生成 JobGraph

运行阶段: 调度 ExecutionGraph

问题拆解: 反压传播两个阶段

1. 跨 TaskManager，反压如何从 InputChannel 传播到 ResultSub
2. TaskManager 内，反压如何从 ResultSub 传播到 InputChannel

下面将以简单的 3 个 Task 的反压进行说明

• 跨 TaskManager
  • 当下游 Task C 的 Receive Buffer 满了，如何告诉上游 Task B 应该降低数据发送速率
  • 当下游 Task C 的 Receive Buffer 空了，如何告诉上游 Task B 应该提升数据发送速率
• TaskManager 内
  • 当 Task B 的 Send Buffer 满了，如何告诉 Task B 内部的 Receive Buffer 下游 Send Buffer 满了、下游处理性能不行了？因为要让 Task B 的 Receive Buffer 感受到压力，才能把下游的压力传递到 Task A
  • 当 Task B 的 Send Buffer 空了，如何告诉 Task B 内部的 Receive Buffer 下游 Send Buffer 空了，下游处理性能很强，上游加快处理数据吧

跨 TaskManager 数据传输

如下图所示，我们先介绍跨 TaskManager 网络传输使用的 buffer。

跨 TaskManager 数据传输

跨 TaskManager 反压过程

跨 TaskManager 数据反压

跨 TaskManager 数据反压

跨 TaskManager 数据反压

跨 TaskManager 数据反压

跨 TaskManager 数据反压

跨 TaskManager 数据反压

跨 TaskManager 数据反压

跨 TaskManager 数据反压

<font style="color:rgb(48, 48, 48);">Local BufferPool</font> 和 <font style="color:rgb(48, 48, 48);">Network BufferPool</font> 都用尽后整个 Operator 就会停止写数据，达到跨 TaskManager 的反压。

TaskManager 内反压过程

Task 内数据反压

整个 Flink 的反压是从下游往上游传播的，一直传播到 Source Task，Source Task 有压力后，会降低从外部组件中读取数据的速率，例如：Source Task 会降低从 Kafka 中读取数据的速率，来降低整个 Flink Job 中缓存的数据，从而降低负载。

Flink TCP-based 反压机制（ since V1.5 版本）

Flink V1.5 之前版本的反压策略存在的问题

• 在一个 TaskManager 中可能要执行多个 Task，如果多个 Task 的数据最终都要传输到下游的同一个 TaskManager 就会复用同一个 Socket 进行传输，这个时候如果单个 Task 产生反压，就会导致复用的 Socket 阻塞，其余的 Task 也无法使用传输，checkpoint barrier 也无法发出导致下游执行 checkpoint 的延迟增大。
• 依赖最底层的 TCP 去做流控，会导致反压传播路径太长，导致生效的延迟比较大。

Credit 的反压策略实现原理

如下图所示，反压机制作用于 Flink 的应用层，即在 ResultSubPartition 和 InputChannel 这一层引入了反压机制。每次上游 SubTask A.2 给下游 SubTask B.4 发送数据时，会把 Buffer 中的数据和上游 ResultSubPartition 堆积的数据量 Backlog size 发给下游，下游会接收上游发来的数据，并向上游反馈目前下游现在的 Credit 值，Credit 值表示目前下游可以接收上游的 Buffer 量，1 个Buffer 等价于 1 个 Credit 。

Credit 数据传输

例如，上游 SubTask A.2 发送完数据后，还有 5 个 Buffer 被积压，那么会把发送数据和 Backlog size = 5 一块发送给下游 SubTask B.4，下游接受到数据后，知道上游积压了 5 个Buffer，于是向 Buffer Pool 申请 Buffer，由于容量有限，下游 InputChannel 目前仅有 2 个 Buffer 空间，所以，SubTask B.4 会向上游 SubTask A.2 反馈 Channel Credit = 2。然后上游下一次最多只给下游发送 2 个 Buffer 的数据，这样每次上游发送的数据都是下游 InputChannel 的 Buffer 可以承受的数据量，所以通过这种反馈策略，保证了不会在公用的 Netty 和 TCP 这一层数据堆积而影响其他 SubTask 通信。ResultSubPartition 会把 buffer 和backlog size 同时发送给下游，下游向上游反馈 credit。

Credit-based 反压过程

总结与思考

总结

• 网络流控是为了在上下游速度不匹配的情况下，防止下游出现过载
• 网络流控有静态限速和动态反压两种手段
• Flink 1.5 之前是基于 TCP 流控 + bounded buffer 实现反压
• Flink 1.5 之后实现了自己托管的 credit - based 流控机制，在应用层模拟 TCP 的流控机制

思考

参考文档

一文搞懂 Flink 网络流控与反压机制

Apache Flink 进阶（七）：网络流控及反压剖析