Flink反压机制剖析

网络流控的概念与背景

1.为什么需要网络流控

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

2.网络流控的实现：静态限速

我们在Producer端实现一个静态限流，Producer经过限流器流量降低到和Consumer端相同，这样的话 Producer 端的发送速率跟 Consumer 端的处理速率就可以匹配起来了，就不会导致上述问题。但是这个解决方案有两点限制：
1.事先无法预估 Consumer 到底能承受多大的速率；
2.Consumer 的承受能力通常会动态地波动。

3.网络流控的实现：动态反馈/自动反压

我们需要Consumer及时的给Producer做一个feedback，告知Producer能够承受的速率是多少(flink的反压就采用动态反馈)。
动态反馈分为两种：
1.负反馈：接受速率小于发送速率时发生，告知 Producer 降低发送速率；
2正反馈：发送速率小于接收速率时发生，告知 Producer 可以把发送速率提上来。

Flink反压机制

TCP流控机制

TCP 包的格式结构，有 Sequence number 这样一个机制给每个数据包做一个编号，还有 ACK number 这样一个机制来确保 TCP 的数据传输是可靠的，除此之外还有一个很重要的部分就是 Window Size，接收端在回复消息的时候会通过 Window Size 告诉发送端还可以发送多少数据。

TCP流控：滑动窗口

接收端根据消费速率确定一个窗口大小，每次通信时会返回发送端一个可用的窗口大小，发送端根据窗口大小发送数据。

当消费端消费出现问题，window=0的时候，发送端会定期的发送 1 个字节的探测消息，这时候接收端就会把 window 的大小进行反馈。当接收端的消费恢复了之后，接收到探测消息就可以将 window 反馈给发送端端了从而恢复整个流程。TCP 就是通过这样一个滑动窗口的机制实现 feedback。

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

当Producer速率大于Consumer速率的时候，一段时间后 InputChannel 的 Buffer 被用尽，于是他会向 Local BufferPool 申请新的 Buffer ,Local BufferPool 用尽向 Network BufferPool申请，当Network BufferPool也用尽的时候，这时 Netty AutoRead 就会被禁掉，Netty 就不会从 Socket 的 Buffer 中读取数据了。过不多久Socket的buffer也会被用尽，这是window=0发送给发送端。这时候socket停止发送。

很快发送端的 Socket 的 Buffer 也被用尽，Netty 检测到 Socket 无法写了之后就会停止向 Socket 写数据。所有的数据就会阻塞在 Netty 的 Buffer 当中，很快Netty的buffer也不能在写数据了，数据就会积压到ResultSubPartition中。和接收端一样ResultSubPartition会不断的向 Local BufferPool 和 Network BufferPool 申请内存。
Local BufferPool 和 Network BufferPool 都用尽后整个 Operator 就会停止写数据，达到跨 TaskManager 的反压。

Flink Credit-based 反压机制（since V1.5）

这个机制简单的理解起来就是在 Flink 层面实现类似 TCP 流控的反压机制来解决上述的弊端，Credit 可以类比为 TCP 的 Window 机制。

每一次 ResultSubPartition 向 InputChannel 发送消息的时候都会发送一个 backlog size 告诉下游准备发送多少消息，下游就会去计算有多少的 Buffer 去接收消息，算完之后如果有充足的 Buffer 就会返还给上游一个 Credit 告知他可以发送消息。

当上流速率大于下游速率的时候，下游的 TaskManager 的 Buffer 已经到达了申请上限，这时候下游就会向上游返回 Credit = 0，ResultSubPartition 接收到之后就不会向 Netty 去传输数据，上游 TaskManager 的 Buffer 也很快耗尽，达到反压的效果，这样在 ResultSubPartition 层就能感知到反压，不用通过 Socket 和 Netty 一层层地向上反馈，降低了反压生效的延迟。同时也不会将 Socket 去阻塞，解决了由于一个 Task 反压导致 TaskManager 和 TaskManager 之间的 Socket 阻塞的问题。

Flink反压机制演变

flink1.5之前使用的是TCP的反压机制，但是反压实现依赖于底层TCP的反压，流程较长，延迟比较高。且TaskManager中一个任务线程数据积压就会造成TCP缓冲区数据积压，阻塞Socket，使其余的Task也都无法传输数据，同时barrier也无法传输到下流，造成下流做checkpoint的延时增大。
因此，1.5之后flink的反压改为了信任度机制，在应用层通过信任度模拟TCP的流量控制，实现反压。