ACAL Lynxus 深入解析

系统正在积极处理的 FlowFile 保存在 JVM 内存中的哈希映射中（更多内容见 深入视图：内存和磁盘中的 FlowFile）。这使得处理它们非常高效，但需要一种辅助机制来在进程重启时提供数据的持久性，原因包括断电、内核恐慌、系统升级和维护周期等。FlowFile 存储库是当前存在于系统中的每个 FlowFile 元数据的"预写日志"（或数据记录）。此 FlowFile 元数据包括与 FlowFile 关联的所有属性、指向 FlowFile 实际内容的指针（存在于内容存储库中）以及 FlowFile 的状态，例如 FlowFile 所属的连接/队列。此预写日志为 Lynxus 提供了处理重启和意外系统故障所需的弹性。

FlowFile 存储库充当 Lynxus 的预写日志，因此当 FlowFile 流经系统时，每个更改都在 FlowFile 存储库中作为事务性工作单元记录在发生之前。这允许系统确切地知道节点在处理数据时处于哪个步骤。如果节点在处理数据时宕机，它可以在重启时轻松从中断处恢复（更多内容见 系统故障对事务的影响）。日志中 FlowFile 的格式是一系列沿途发生的增量（或更改）。Lynxus 通过恢复 FlowFile 的"快照"（在存储库检查点时创建）然后重播每个增量来恢复 FlowFile。

系统定期自动拍摄快照，为每个 FlowFile 创建新快照。系统通过序列化哈希映射中的每个 FlowFile 并将其写入磁盘（文件名为".partial"）来计算新的基础检查点。随着检查点的进行，新的 FlowFile 基线被写入".partial"文件。一旦检查点完成，旧的"snapshot"文件被删除，".partial"文件被重命名为"snapshot"。

系统检查点之间的周期可以在 'lynxus.properties' 文件中配置（在 Lynxus 系统管理员指南 中有记录）。默认是两分钟间隔。

内容存储库只是本地存储中所有 FlowFile 内容存在的地方，通常是三个存储库中最大的。正如介绍部分所述，此存储库利用不可变性和写时复制范式来最大化速度和线程安全。影响内容存储库的核心设计决策是将 FlowFile 的内容保存在磁盘上，仅在需要时才将其读入 JVM 内存。这允许 Lynxus 处理微小和巨大的对象，而无需生产者和消费者处理器在内存中保存完整对象。因此，拆分、聚合和转换非常大的对象等操作很容易做到而不会损害内存。

与 JVM 堆具有垃圾回收过程以在需要空间时回收不可达对象的方式相同，Lynxus 中存在一个专用线程来分析内容存储库中未使用的内容（更多信息在"深入视图：检查点后删除"部分）。在 FlowFile 的内容被识别为不再使用后，它将被删除或归档。如果在 'lynxus.properties' 中启用了归档，则 FlowFile 的内容将存在于内容存储库中，直到它被淘汰（在一定时间后删除）或因内容存储库占用太多空间而删除。归档和/或删除的条件在 'lynxus.properties' 文件中配置（"lynxus.content.repository.archive.max.retention.period"、"lynxus.content.repository.archive.max.usage.percentage"），并在 Lynxus 系统管理员指南 中概述。有关内容删除的更多信息，请参阅"数据退出"部分。

溯源存储库是存储每个 FlowFile 历史的地方。此历史用于提供每个数据的数据血统（也称为监管链）。每次 FlowFile 发生事件（FlowFile 被创建、分叉、克隆、修改等）时，都会创建一个新的溯源事件。此溯源事件是 FlowFile 在时间点看起来和适应流的快照。当创建溯源事件时，它会复制 FlowFile 的所有属性和指向 FlowFile 内容的指针，并将其与 FlowFile 的状态（例如它与其他溯源事件的关系）聚合到溯源存储库中的一个位置。此快照不会更改，除非数据过期。溯源存储库在完成后的一段时间内保存所有这些溯源事件，如 'lynxus.properties' 文件中指定的。

由于 FlowFile 的所有属性和指向内容的指针都保存在溯源存储库中，数据流管理器不仅能够查看该数据块的血统或处理历史，还能够稍后查看数据本身，甚至从流中的任何点重播数据。一个常见的用例是当下游系统声称未收到数据时。数据血统可以准确显示数据何时交付给下游系统、数据的样子、文件名以及数据发送到的 URL——或者可以确认数据确实从未发送。在任何一种情况下，都可以通过单击按钮（或通过访问适当的 HTTP API 端点）重播发送事件，以便仅将数据重新发送到该特定下游系统。或者，如果数据处理不当（也许应该先进行某些数据操作），可以修复流，然后将数据重播到新流中，以便正确处理数据。

请记住，由于溯源不复制内容存储库中的内容，而只是复制 FlowFile 指向内容的指针，内容可能在引用它的溯源事件被删除之前被删除。这意味着用户将无法再查看内容或稍后重播 FlowFile。但是，用户仍然能够查看 FlowFile 的血统并了解数据发生了什么。例如，即使数据本身不可访问，用户仍然能够查看数据的唯一标识符、其文件名（如果适用）、何时收到、从哪里收到、如何操作、发送到哪里等等。此外，由于 FlowFile 的属性可用，数据流管理器能够理解数据为何以这种方式处理，为理解和调试数据流提供了关键工具。

有关溯源存储库背后的设计决策，请查看 持久溯源存储库设计。

当生产者处理器调用"ProcessSession.create()"然后适当调用 ProvenanceReporter 时，FlowFile 在系统中创建。"ProcessSession.create()"调用创建一个空的 FlowFile，具有几个核心属性（标准进程会话的文件名、路径和 uuid），但没有任何内容或父级的血统（create 方法被重载以允许父 FlowFile 的参数）。然后生产者处理器将内容和属性添加到 FlowFile。

ProvenanceReporter 用于发出 FlowFile 的溯源事件。如果文件由 Lynxus 从外部实体未接收的数据创建，则应发出"CREATE"事件。如果数据是从外部源接收的数据创建的，则应发出"RECEIVE"事件。溯源事件分别使用"ProvenanceReporter.create()"和"ProvenanceReporter.receive()"制作。

在我们的 WebCrawler 流中，InvokeHTTP 处理器（称为从种子 URL 获取）使用"ProcessSession.create()"创建初始 FlowFile，并使用"ProvenanceReporter.receive()"记录数据的接收。此方法调用还提供接收数据的 URL、传输数据所需的时间以及添加到 FlowFile 的任何 FlowFile 属性。例如，HTTP 标头可以作为 FlowFile 属性添加。

基于流编程的一个重要方面是黑盒之间资源受限关系的概念。在 Lynxus 中，这些分别是队列和处理器。FlowFile 通过简单地传递 FlowFile 的引用（类似于 EIP 中的"声明检查"模式）从一个处理器路由到另一个处理器通过队列。

在 WebCrawler 流中，InvokeHTTP 处理器使用 HTTP GET 请求到达 URL，并根据 HTTP 服务器的响应向 FlowFile 添加状态码属性。在更新 FlowFile 的文件名后（在 InvokeHttp 之后的 UpdateAttribute 处理器中），有一个 RouteOnAttribute 处理器将具有成功状态码属性的 FlowFile 路由到两个不同的处理器。那些不匹配的被 RouteOnAttribute 处理器"丢弃"（请参阅数据退出部分），因为它被配置为自动终止任何不匹配任何路由规则的数据。进入 RouteOnAttribute 处理器有一个 FlowFile (F1)，包含状态码属性并指向内容 (C1)。有一个溯源事件指向 C1 并包含 F1 的快照，但为了更好地关注路由而省略。此信息分别位于 FlowFile、内容和溯源存储库中。

在 RouteOnAttribute 处理器检查 FlowFile 的状态码属性后，它确定应该将其路由到两个不同的位置。首先发生的是处理器克隆 FlowFile 以创建 F2。这复制了所有属性和指向内容的指针。由于它只是路由和分析属性，内容不会更改。然后将 FlowFile 添加到各自的连接队列，等待下一个处理器检索它们进行处理。

ProvenanceReporter 记录发生的更改，其中包括 CLONE 和两个 ROUTE 事件。每个事件都有指向相关内容的指针，并包含快照形式的各自 FlowFile 的副本。

除了基于属性路由 FlowFile 外，一些处理器还基于内容路由。虽然效率不高，但有时这是必要的，因为您想将 FlowFile 的内容拆分为多个 FlowFile。

一个例子是 SplitText 处理器。此处理器分析内容查找行尾字符，并创建包含可配置行数的新 FlowFile。Web 爬虫流使用它将潜在 URL 拆分为单行以进行 URL 提取并作为 InvokeHttp 的请求。SplitText 处理器的一个好处是，由于处理器正在拆分连续块（没有 FlowFile 内容是不相交或重叠的），处理器可以在不复制任何内容的情况下进行此路由。它所做的只是创建新的 FlowFile，每个都有指向原始 FlowFile 内容部分的指针。这是通过 Lynxus API 中内置的内容划分和拆分功能实现的。虽然并非总是可以以这种方式拆分，但当可行时，性能优势是可观的。

RouteText 是一个处理器，展示了为什么某些样式的路由可能需要复制内容。此处理器分析每一行并将其路由到一个或多个基于可配置属性的关系。当多于一行被路由到同一关系时（对于同一输入 FlowFile），这些行被组合成一个 FlowFile。由于行可能是不相交的（第 1 行和第 100 行路由到同一关系），一个指针无法准确描述 FlowFile 的内容，处理器必须将内容复制到新位置。例如，在 Web 爬虫流中，RouteText 处理器将所有包含"lynxus"的行路由到"Lynxus"关系。因此，当有一个输入 FlowFile 在网页上多次包含"lynxus"时，只会发送一封电子邮件（通过后续的 PutEmail 处理器）。

漏斗是一个组件，它从一个或多个连接接收输入并将它们路由到一个或多个目的地。典型用例在用户指南中描述。无论用例如何，如果漏斗下游只有一个处理器，则漏斗不会发出溯源事件，它在溯源图中看起来是不可见的。如果有多个下游处理器，如 WebCrawler 中的那个，则会发生克隆事件。参考下面的图形，您可以看到新的 FlowFile (F2) 从原始 FlowFile (F1) 克隆，就像上面的路由一样，新的 FlowFile 只有指向相同内容的指针（内容不被复制）。

从开发者的角度来看，您可以将漏斗视为一个非常简单的处理器。当它被安排运行时，它只是执行"ProcessSession.get()"然后"ProcessSession.transfer()"到输出连接。如果有多个输出连接（如下面的示例），则运行"ProcessSession.clone()"。最后调用"ProcessSession.commit()"，完成事务。

在前面的示例中，只有路由但没有更改 FlowFile 的内容。下一个示例关注 CompressContent 处理器，它压缩包含要分析的网页的合并 FlowFile 的捆绑包。

在此示例中，FlowFile F1 的内容 C1 正在 CompressContent 处理器中压缩。由于 C1 是不可变的，我们想要完整的可重播溯源历史，我们不能只是覆盖 C1。为了"修改"C1，我们执行"写时复制"，这是通过在内容复制到内容存储库中的新位置时修改内容来实现的。这样做时，FlowFile 引用 F1 更新为指向新的压缩内容 C2，并创建新的溯源事件 P2，引用新的 FlowFile F1.1。由于 FlowFile 存储库是不可变的，不是修改旧的 F1，而是创建新的增量 (F1.1)。以前的溯源事件仍然有指向内容 C1 的指针并包含旧属性，但它们不是 FlowFile 的最新版本。

处理 FlowFile 的属性是 Lynxus 的核心方面。假设属性足够小，可以在处理器每次执行时完全读入本地内存。因此它们易于处理非常重要。由于属性是路由和处理 FlowFile 的核心方式，拥有只更改 FlowFile 属性的处理器是非常常见的。一个这样的例子是 UpdateAttribute 处理器。UpdateAttribute 处理器所做的只是根据处理器的属性更改传入 FlowFile 的属性。

查看图表，在处理器之前有 FlowFile (F1)，它具有属性和指向内容 (C1) 的指针。处理器通过创建仍然具有指向内容 (C1) 的指针的新增量 (F1.1) 来更新 FlowFile 的属性。发生这种情况时会发出"ATTRIBUTES_MODIFIED"溯源事件。

在此示例中，以前的处理器 (InvokeHTTP) 从 URL 获取信息并创建了具有与请求 FlowFile 相同的文件名属性的新响应 FlowFile。这无助于描述响应 FlowFile，因此 UpdateAttribute 处理器将文件名属性修改为更相关的内容（URL 和事务 ID）。

最终，Lynxus 中的数据将达到一个点，要么它已被加载到另一个系统中，我们可以停止处理它，要么我们过滤掉了 FlowFile 并确定我们不再关心它。无论哪种方式，FlowFile 最终将被"丢弃"。"DROP"是一个溯源事件，意味着我们不再在流中处理 FlowFile，它可以被删除。它保留在 FlowFile 存储库中，直到下一个存储库检查点。溯源存储库将溯源事件保留 'lynxus.properties' 中声明的时间量（默认是 24 小时）。内容存储库中的内容在 FlowFile 离开 Lynxus 时被标记为删除，预写日志的后台检查点处理以压缩/删除发生。除非另一个 FlowFile 引用相同的内容或 'lynxus.properties' 中启用了归档。如果启用了归档，内容存在直到达到磁盘的最大百分比或达到最大保留期限（也在 'lynxus.properties' 中设置）。