Akka Streams扇出后过滤无输出问题咨询

问题核心 #

你遇到的情况是：Akka Streams中对扇出的两个分支分别设置不同过滤器时无输出，但两个分支用相同过滤器时有正常输出。核心原因大概率是流的完成机制导致空分支阻塞了整个流的终止。

常见原因分析 #

1. 分支合并逻辑依赖双分支完成
   如果你用了zip这类需要两个分支都产生元素才能输出的操作，当其中一个分支没有元素通过过滤器时（比如filter1筛"2"但输入中符合条件的元素不足或没有），zip会一直等待该分支的元素，导致整个流无法完成，自然看不到任何输出。而当两个过滤器都筛"1"时，两个分支都有元素，zip能正常配对输出。

2. 过滤器条件误写
   虽然概率低，但可以再检查下过滤器的判断逻辑，比如是否把test == "2"写成了test != "2"，导致没有元素通过。

3. 扇出组件配置错误
   比如使用Broadcast时，输出端口数量和实际连接的分支数不匹配，导致部分分支没有正确接收元素。

解决方案 #

1. 调整合并逻辑，避免空分支阻塞 #

如果不需要两个分支的元素配对输出，不要用zip，而是用Merge合并两个分支，或者分别处理每个分支的输出：

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Broadcast, Flow, Sink, Source}

case class Test(test: String, tester: Double)

object StreamTest extends App {
  implicit val system: ActorSystem = ActorSystem("StreamTest")
  
  val source = Source(List(
    Test("1", 23.0),
    Test("1", 23.0),
    Test("2", 45.0)
  ))

  val broadcast = Broadcast[Test](2)
  
  val filter1 = Flow[Test].filter(_.test == "2")
  val filter2 = Flow[Test].filter(_.test == "1")

  // 方案1：分别处理两个分支的输出
  source.via(broadcast)
    .alsoTo(filter1.to(Sink.foreach(println)))
    .via(filter2)
    .runWith(Sink.foreach(println))

  // 方案2：用Merge合并后输出
  /*
  val merge = Merge[Test](2)
  source.via(broadcast)
    .via(filter1)
    .merge(source.via(broadcast).via(filter2))
    .runWith(Sink.foreach(println))
  */
}

2. 给空分支添加终止逻辑 #

如果必须保留zip这类配对逻辑，可以给可能为空的分支添加concat(Source.empty)，确保该分支能正常完成：

import akka.stream.scaladsl.Zip

// 给分支添加空元素流确保正常完成
val completedFilter1 = filter1.concat(Source.empty[Test])
val completedFilter2 = filter2.concat(Source.empty[Test])

source.via(broadcast)
  .via(completedFilter2)
  .zip(source.via(broadcast).via(completedFilter1))
  .runWith(Sink.foreach(println))

3. 检查扇出与过滤器配置 #

确认Broadcast的输出端口数和连接的分支数一致，过滤器的条件完全符合预期（比如字符串匹配是否区分大小写，这里注意是字符串"1"/"2"而非数字）。

内容的提问来源于stack exchange，提问作者blue-sky