akka-streams 有状态的子流

【问题标题】：akka-streams stateful substream flowakka-streams 有状态的子流
【发布时间】：2021-03-18 14:45:03
【问题描述】：

有一个有状态的流：

val stream = Flow[Event].statefulMapConcat {
  () =>

    val state = ...

    {
      element =>
        // change the state
        element :: Nil
    }
}

它是流程的一部分

Flow[Event]
  .groupBy(1000000, event => event.key2, allowClosedSubstreamRecreation = true)
  .via(stream)
  .mergeSubstreams

有没有办法让每个子流在stream 中有一个state（在这个例子中是groupBy 之后的每个键）？我认为它应该按子流实现，但不知道该怎么做。

【问题讨论】：

标签： scala akka akka-stream

【解决方案1】：

您确实会在该设置中获得每个子流的状态：

  val stream = Flow[Int].statefulMapConcat {
    () => {

      var state: List[Int] = Nil

      element => {
        state = element :: state
        List(state)
      }
    }
  }

  val groupByFlow =
  Flow[Int]
    .groupBy(1000000, identity, allowClosedSubstreamRecreation = true)
    .via(stream)
    .mergeSubstreams

  Source(List(1,1,2,3,3,3))
    .via(groupByFlow)
    .runForeach(i => println(i))

将打印

List(1)
List(1, 1)
List(3)
List(2)
List(3, 3)
List(3, 3, 3)

【讨论】：