Akka溪流减少为较小的溪流

一种选择是使用 statefulMapConcat 组合器:

Source(List("A", "A", "B", "B", "B", "C", "C", ""))
      .statefulMapConcat({ () =>
        var lastChar = ""
        var count = 0

        char => if(lastChar == char) {
            count += 1
            List.empty
          } else {
            val charCount = (lastChar, count)
            lastChar = char
            count = 1
            List(charCount)
          }
      })
    .runForeach(println)

然而,这需要在输入流中附加一个元素来标记结束。

输出:

(,0)
(A,2)
(B,3)
(C,2)

感谢@chunjef在评论中的建议