原理解析 | 深入了解Apache Flink的网络协议栈

下图从上面的扩展了更高级别的视图，其中包含网络协议栈及其周围组件的更多详细信息，从发送算子发送记录(Record)到接收算子获取它：

在生成 Record 并将其传递出去之后，例如通过 Collector#collect()，它被传递给 RecordWriter，RecordWriter 会将 Java 对象序列化为字节序列，最终存储在 Buffer 中按照上面所描述的在网络协议栈中进行处理。RecordWriter 首先使用 SpanningRecordSerializer 将 Record 序列化为灵活的堆上字节数组。然后，它尝试将这些字节写入目标网络 Channel 的 Buffer 中。我们将在下面的章节回到这一部分。

在接收方，底层网络协议栈(Netty)将接收到的 Buffer 写入相应的输入通道(Channel)。流任务的线程最终从这些队列中读取并尝试在 RecordReader 的帮助下通过 SpillingAdaptiveSpanningRecordDeserializer 将累积的字节反序列化为 Java 对象。与序列化器类似，这个反序列化器还必须处理特殊情况，例如跨越多个网络 Buffer 的 Record，或者因为记录本身比网络缓冲区大(默认情况下为32KB，通过 taskmanager.memory.segment-size 设置)或者因为序列化 Record 时，目标 Buffer 中已经没有足够的剩余空间保存序列化后的字节数据，在这种情况下，Flink 将使用这些字节空间并继续将其余字节写入新的网络 Buffer 中。

4.1 将网络 Buffer 写入 Netty

在上图中，Credit-based 流控制机制实际上位于“Netty Server”(和“Netty Client”)组件内部，RecordWriter 写入的 Buffer 始终以空状态(无数据)添加到 Subpartition 中，然后逐渐向其中填写序列化后的记录。但是 Netty 在什么时候真正的获取并发送这些 Buffer 呢?显然，不能是 Buffer 中只要有数据就发送，因为跨线程(写线程与发送线程)的数据交换与同步会造成大量的额外开销，并且会造成缓存本身失去意义(如果是这样的话，不如直接将将序列化后的字节发到网络上而不必引入中间的 Buffer)。

在 Flink 中，有三种情况可以使 Netty 服务端使用(发送)网络 Buffer：

• 写入 Record 时 Buffer 变满，或者
• Buffer 超时未被发送，或
• 发送特殊消息，例如 Checkpoint barrier。

▼ 在 Buffer 满后发送

RecordWriter 将 Record 序列化到本地的序列化缓冲区中，并将这些序列化后的字节逐渐写入位于相应 Result subpartition 队列中的一个或多个网络 Buffer中。虽然单个 RecordWriter 可以处理多个 Subpartition，但每个 Subpartition 只会有一个 RecordWriter 向其写入数据。另一方面，Netty 服务端线程会从多个 Result subpartition 中读取并像上面所说的那样将数据写入适当的多路复用信道。这是一个典型的生产者 - 消费者模式，网络缓冲区位于生产者与消费者之间，如下图所示。在(1)序列化和(2)将数据写入 Buffer 之后，RecordWriter 会相应地更新缓冲区的写入索引。一旦 Buffer 完全填满，RecordWriter 会(3)为当前 Record 剩余的字节或者下一个 Record 从其本地缓冲池中获取新的 Buffer，并将新的 Buffer 添加到相应 Subpartition 的队列中。这将(4)通知 Netty服务端线程有新的数据可发送(如果 Netty 还不知道有可用的数据的话4)。每当 Netty 有能力处理这些通知时，它将(5)从队列中获取可用 Buffer 并通过适当的 TCP 通道发送它。

注释4：如果队列中有更多已完成的 Buffer，我们可以假设 Netty 已经收到通知。

▼ 在 Buffer 超时后发送

为了支持低延迟应用，我们不能只等到 Buffer 满了才向下游发送数据。因为可能存在这种情况，某种通信信道没有太多数据，等到 Buffer 满了在发送会不必要地增加这些少量 Record 的处理延迟。因此，Flink 提供了一个定期 Flush 线程(the output flusher)每隔一段时间会将任何缓存的数据全部写出。可以通过 StreamExecutionEnvironment#setBufferTimeout 配置 Flush 的间隔，并作为延迟5的上限(对于低吞吐量通道)。下图显示了它与其他组件的交互方式：RecordWriter 如前所述序列化数据并写入网络 Buffer，但同时，如果 Netty 还不知道有数据可以发送，Output flusher 会(3,4)通知 Netty 服务端线程数据可读(类似与上面的“buffer已满”的场景)。当 Netty 处理此通知(5)时，它将消费(获取并发送)Buffer 中的可用数据并更新 Buffer 的读取索引。Buffer 会保留在队列中——从 Netty 服务端对此 Buffer 的任何进一步操作将在下次从读取索引继续读取。

注释5：严格来说，Output flusher 不提供任何保证——它只向 Netty 发送通知，而 Netty 线程会按照能力与意愿进行处理。这也意味着如果存在反压，则 Output flusher 是无效的。

▼ 特殊消息后发送

一些特殊的消息如果通过 RecordWriter 发送，也会触发立即 Flush 缓存的数据。其中最重要的消息包括 Checkpoint barrier 以及 end-of-partition 事件，这些事件应该尽快被发送，而不应该等待 Buffer 被填满或者 Output flusher 的下一次 Flush。

▼ 进一步的讨论

（编辑：核心网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!