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

发布时间：2020-12-16 19:09:41 所属栏目：安全来源：网络整理

导读：有一个自定义流的流,在某个阶段我想拆分流,并有两个替代数据处理,稍后将再次合并. 例如. - F3 - F6 Src - F1 - F2 Merge - Sink - F4 - F5 F2应该有一个条件,说如果数据包含格式A,那么它应该转到流F3,否则转到F4. 据我所知,每个流程在每个方向上只能有一个端

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

例如.

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

F2应该有一个条件,说如果数据包含格式A,那么它应该转到流F3,否则转到F4.

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

解决方法

您可以使用Broadcast来拆分流,然后您就可以在每个流上使用过滤器或收集来过滤所需的数据.

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

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

此外,还有分区阶段,它处理输出端口的数量和映射函数,从值到端口号f：T =>诠释.

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

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

（编辑：李大同）

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