深入了解 Flink 的网络协议栈( 二 ) _网络协议栈

文章插图

每个子任务的输出结果称为ResultPartition，每个ResultPartition被分成多个单独的ResultSubpartition– 每个逻辑通道一个。Flink的网络协议栈在这一点的处理上，不再处理单个记录，而是将一组序列化的记录填充到网络缓冲区中进行处理。每个子任务本地缓冲区中最多可用buffer数目为（每个发送方和接收方各一个）:
#channels * buffers-per-channel + floating-buffers-per-gate
单个TaskManager上的网络层buffer总数通常不需要配置。有关如何在需要时进行配置的详细信息，请参阅配置网络缓冲区的文档。
▼ 造成反压（1）
每当子任务的数据发送缓冲区耗尽时——数据驻留在 Subpartition 的缓冲区队列中或位于更底层的基于 Netty 的网络堆栈内，生产者就会被阻塞，无法继续发送数据，而受到反压。接收端以类似的方式工作：Netty 收到任何数据都需要通过网络 Buffer 传递给 Flink 。如果相应子任务的网络缓冲区中没有足够可用的网络 Buffer，Flink 将停止从该通道读取，直到 Buffer 可用。这将反压该多路复用上的所有发送子任务，因此也限制了其他接收子任务。下图说明了过载的子任务 B.4，它会导致多路复用的反压，也会导致子任务 B.3 无法接受和处理数据，即使是 B.3 还有足够的处理能力。

文章插图

为了防止这种情况发生，Flink 1.5引入了自己的流量控制机制。
3.Credit-based流量控制Credit-based 流量控制可确保发送端已经发送的任何数据，接收端都具有足够的能力(Buffer)来接收。新的流量控制机制基于网络缓冲区的可用性，作为 Flink 之前机制的自然延伸。每个远程输入通道(RemoteInputChannel)现在都有自己的一组独占缓冲区(Exclusive buffer)，而不是只有一个共享的本地缓冲池(LocalBufferPool) 。与之前不同，本地缓冲池中的缓冲区称为流动缓冲区(Floating buffer)，因为它们会在输出通道间流动并且可用于每个输入通道。
数据接收方会将自身的可用 Buffer 作为 Credit 告知数据发送方（1 buffer = 1 credit）。每个 Subpartition 会跟踪下游接收端的 Credit（也就是可用于接收数据的 Buffer 数目）。只有在相应的通道(Channel)有 Credit 的时候 Flink 才会向更底层的网络协议栈发送数据(以 Buffer 为粒度)，并且每发送一个 Buffer 的数据，相应的通道上的 Credit 会减 1 。除了发送数据本身外，数据发送端还会发送相应 Subpartition 中有多少正在排队发送的 Buffer 数（称之为 Backlog）给下游。数据接收端会利用这一信息(Backlog)去申请合适数量的 Floating buffer 用于接收发送端的数据，这可以加快发送端堆积数据的处理。接收端会首先申请和 Backlog 数量相等的 Buffer，但可能无法申请到全部，甚至一个都申请不到，这时接收端会利用已经申请到的 Buffer 进行数据接收，并监听是否有新的 Buffer 可用。

文章插图

Credit-based 的流控使用 buffers-per-channel 来指定每个channel有多少独占的buffer，使用 floating-buffers-per-gate 来指定共享的本地缓冲池(local buffer pool)大小（可选3），通过共享本地缓冲池，credit-based流控可以使用的buffer数目可以达到与原来非credit-based流控同样的大小。这两个参数的默认值是被精心选取的，以保证新的 credit-based 流控在网络健康延迟正常的情况下至少可以达到与原策略相同的吞吐。您可以根据您实际的网络 RRT(round-trip-time)和带宽对这两个参数进行调整。
注释3：如果没有足够的 Buffer 可用，则每个缓冲池将获得全局可用 Buffer 的相同份额（±1）。
▼ 造成反压（2）
与没有流量控制的接收端反压机制不同，Credit 提供了更直接的控制：如果接收端的处理速度跟不上，最终它的 Credit 会减少成 0，此时发送端就不会在向网络中发送数据（数据会被序列化到 Buffer 中并缓存在发送端）。由于反压只发生在逻辑链路上，因此没必要阻断从多路复用的 TCP 连接中读取数据，也就不会影响其他的接收者接收和处理数据。
▼ Credit-based 的优势与问题
由于通过 credit-based 流控机制，多路复用中的一个信道不会由于反压阻塞其他逻辑信道，因此整体资源利用率会增加。此外，通过完全控制正在发送的数据量，我们还能够加快 checkpoint alignment：如果没有流量控制，通道需要一段时间才能填满网络协议栈的内部缓冲区并表明接收端不再读取数据了。在这段时间里，大量的 buffer 不会被处理。任何 checkpoint barrier（触发 checkpoint 的消息）都必须在这些数据 buffer 区后排队，因此必须等到所有这些数据都被处理后才能够触发 checkpoint（“barrier 不会再数据之前被处理！”）。