基于akka流的条件的替代流程

Question

有一个自定义流的流,在某个阶段我想拆分流,并有两个替代数据处理,稍后将再次合并.

例如

                  -> F3 -> F6 
Src -> F1 -> F2                > Merge -> Sink 
                  -> F4 -> F5

F2应该有条件说数据是否包含格式A然后它应该流动F3,否则转到F4.

据我所知,每个流程在每个方向上只能有一个端口(如果是双向的话,则为两个端口) - 那么我如何支持这样的流程呢？

Answer 1

您可以使用Broadcast拆分流,然后您可以使用filter或collect在每个流上过滤所需的数据.

val split = builder.add(Broadcast[Int](2))

Src -> F1 -> split -> filterCondA -> F3 -> F6 -> Merge -> Sink
                   -> filterCondB -> F4 -> F5 -> Merge

此外,还有一个Partition阶段,它处理输出端口的数量和从值到端口号的映射功能f: T => Int.

val portMapper(value: T): Int = value match {
  case CondA => 0
  case CondB => 1
}

val split = builder.add(Partition[T](2, portMapper))

Src -> F1 -> split -> F3 -> F6 -> Merge -> Sink
             split -> F4 -> F5 -> Merge

也许有一些更简单的方法.