Wait

在 Wait Processor 中处理多个 signal id 的最佳实践

当 Wait Processor 需要处理多个 signal id（通过在 'Release Signal Identifier' 中配置 FlowFile Attribute 的表达式语言实现）时，为了获得预期结果，有几件事需要考虑。Processor 的 配置会因需求而异。同时，你还需要对 Wait Processor 的工作原理有总体了解：

• Wait Processor 一次只处理一个 signal id
• Wait Processor 的运行频率由 'Run Schedule' 定义
• 处理哪个 FlowFile 由 Prioritizer 决定
• 不仅限于 Wait Processor，对所有 Processor 而言，如果没有设置 Prioritizer， Connection 中排队 FlowFile 的顺序是不确定的

常见模式参见以下小节：

• 信号一经通知即释放对应 FlowFile
• 在每个 signal id 内优先释放高优先级 FlowFile

信号一经通知即释放对应 FlowFile

这是最常见的用例。FlowFile 之间相互独立，可以按任意顺序释放。

重要配置：

• 在 'wait' Relationship 上使用 FirstInFirstOutPrioritizer（FIFO） （如果 'Wait Mode' 为 'Keep in the upstream connection'，则在上游 Connection 上使用）

下表演示了每个 Wait 运行周期中：已通知的 signal id、排队中的 FlowFile 以及会发生什么。

在每个 signal id 内优先释放高优先级 FlowFile

多个 FlowFile 共享同一个 signal id，且释放顺序很重要。

重要配置：

• 在 'wait' Relationship 上使用 FIFO 之外的、符合你需求的一个（或一组）Prioritizer （如果 'Wait Mode' 为 'Keep in the upstream connection'，则在上游 Connection 上使用）， 例如 PriorityPrioritizer
• 指定合适的 'Wait Penalty Duration'，例如 "3 sec"
• 'Wait Penalty Duration' 应大于 'Run Schedule'，例如 "3 sec" > "1 sec"
• 增大 'Run Duration'，以避免 signal id 数量的限制（参见下文说明）

下表演示了每个 Wait 运行周期中：已通知的 signal id、排队中的 FlowFile 以及会发生什么。 示例使用 PriorityPrioritizer 来控制同一 signal id 内 FlowFile 的处理顺序。如果配置了 'Wait Penalty Duration'，Wait Processor 会跟踪未释放的 signal id 及其惩罚（表示何时再次检查）。

使用 'Wait Penalty Duration' 时 'Run Duration' 的重要性

基于 'Run Schedule' 与 'Wait Penalize Duration' 的组合，可检查的信号数量存在上限。一旦触及该 限制，即使某些 FlowFile 的 signal id 已被通知，它们也可能不被处理而一直留在 'wait' Relationship 中。假设 Wait 的配置为：

• Run Schedule = 1 sec
• Wait Penalize Duration = 3 sec
• Release Signal Identifier = ${uuid}

并且 'wait' Relationship 中有 5 个 FlowFile：F1、F2 ... F5。此时 F5 的信号被通知。Wait 的 工作方式如下：

• 00:00 时，Wait 检查 F1 的信号，未找到，对 F1 施加惩罚（至 00:03）
• 00:01 时，Wait 检查 F2 的信号，未找到，对 F2 施加惩罚（至 00:04）
• 00:02 时，Wait 检查 F3 的信号，未找到，对 F3 施加惩罚（至 00:05）
• 00:03 时，Wait 检查 F4 的信号，未找到，对 F4 施加惩罚（至 00:06）
• 00:04 时，Wait 再次检查 F1 的信号，因为它已不再处于惩罚期

上述循环反复进行，因此在 F1 ... F4 中某个被释放之前，F5 不会被释放。

为缓解这一限制，建议增大 'Run Duration'。增大 'Run Duration' 后，Wait Processor 可以在该时长内 持续被调度。例如，'Run Duration' 设为 500 ms 时，Wait 应当能在单次运行中遍历全部 5 个排队的 FlowFile。

使用计数器（counter）

计数器本质上是一个标签，用于区分缓存内的不同信号（这里的"缓存"是指包含具有相同 signal identifier 的信号的一个"容器"）。

假设缓存中有如下信号（注意，这些不是 Wait Processor 传入（或 wait）Connection 上的 FlowFile，而是存储在缓存中的释放信号）：

在此状态下，以下 FlowFile 被 Wait Processor 处理（该 FlowFile 带有 signal_counter_name Attribute，且 Wait Processor 被配置为使用该 Attribute 的值作为 Signal Counter Name 属性的值）：

尽管 Signal ID "A" 对应的缓存中有信号，上面的 FlowFile 仍会被发送到 'wait' Relationship， 因为缓存中没有属于名为 "counter_3" 的计数器的信号。

假设缓存状态与上面相同，以下 FlowFile 被 Wait Processor 处理：

该 FlowFile 会被传输到 'success' Relationship，因为缓存 "A" 中有信号，且存在属于 "counter_1" 的信号。输出的 FlowFile 会被追加以下 Attribute 及其值：

• wait.counter.counter_1 : 2
• wait.counter.counter_2 : 1
• wait.counter.total : 3

这里的关键点是：计数器可用于区分缓存内的不同信号。如果使用了计数器，经过 Wait Processor 的 每个计数器都会为 FlowFile 追加一个新 Attribute。如果一个缓存中使用了大量计数器，经过 Wait Processor 的 FlowFile 会被追加大量 Attribute。为避免这种情况，建议使用多个缓存、每个 缓存只含少量计数器，而不是一个缓存含很多计数器。

例如：

• Release Signal ID "A" 对应的缓存包含计数器："counter_1" 和 "counter_2"
• Release Signal ID "B" 对应的缓存包含计数器："counter_3" 和 "counter_4"
• Release Signal ID "C" 对应的缓存包含计数器："counter_5" 和 "counter_6"

（计数器名不需要在缓存之间唯一，缓存 "A" 使用的计数器名也可以在缓存 "B" 和 "C" 中复用。）