Объединение двух потоков в потоке Акка

Question 1

Я пытаюсь объединить два потока и не могу объяснить результат моей реализации.

val source = Source(1 to 10)
val sink = Sink.foreach(println)

val flow1 = Flow[Int].map(s => s + 1)
val flow2 = Flow[Int].map(s => s * 10)

val flowGraph = Flow.fromGraph(
    GraphDSL.create() { implicit builder =>
      import GraphDSL.Implicits._

      val concat = builder.add(Concat[Int](2))
      val broadcast = builder.add(Broadcast[Int](2))

      broadcast ~> flow1 ~> concat.in(0)
      broadcast ~> flow2 ~> concat.in(1)

      FlowShape(broadcast.in, concat.out)
    }
  )

source.via(flowGraph).runWith(sink)

Я ожидаю следующего вывода из этого кода.

Вместо этого я вижу, что печатается только "2". Не могли бы вы, пожалуйста, объяснить, что не так в моей импликации и как я должен изменить программу, чтобы получить желаемый результат.

Question 2

Из документов API Akka Stream:

Concat:

Выдает, когда в текущем потоке доступен элемент; если текущий ввод завершен, он пытается выполнить следующий

Broadcast:

Излучается, когда все выходы прекращают обратное давление и доступен входной элемент

Два оператора не будут работать совместно, так как существует конфликт в том, как они работают -- Concat пытается вытащить все элементы из одного из Broadcastвыходы перед переключением на другой, в то время как Broadcast не будет излучать, если не будет спроса на ВСЕ его продукты.

Для того, что вам нужно, вы можете объединить, используяconcat как предположили комментаторы:

source.via(flow1).concat(source.via(flow2)).runWith(sink)

или, что эквивалентно, используйтеSource.combine как показано ниже:

Source.combine(source.via(flow1), source.via(flow2))(Concat[Int](_)).runWith(sink)

Question 3

С помощью GraphDSL, который является упрощенной версией реализации Source.combine:

val sg = Source.fromGraph(
  GraphDSL.create(){ implicit builder =>
    import GraphDSL.Implicits._

    val concat = builder.add(Concat[Int](2))

    source ~> flow1 ~> concat
    source ~> flow2 ~> concat

    SourceShape(concat.out)
  }
)

sg.runWith(sink)

Leo C · Answer 1 · 2021-10-21T22:34:04

Из документов API Akka Stream:

Concat:

Выдает, когда в текущем потоке доступен элемент; если текущий ввод завершен, он пытается выполнить следующий

Broadcast:

Излучается, когда все выходы прекращают обратное давление и доступен входной элемент

Два оператора не будут работать совместно, так как существует конфликт в том, как они работают -- Concat пытается вытащить все элементы из одного из Broadcastвыходы перед переключением на другой, в то время как Broadcast не будет излучать, если не будет спроса на ВСЕ его продукты.

Для того, что вам нужно, вы можете объединить, используяconcat как предположили комментаторы:

source.via(flow1).concat(source.via(flow2)).runWith(sink)

или, что эквивалентно, используйтеSource.combine как показано ниже:

Source.combine(source.via(flow1), source.via(flow2))(Concat[Int](_)).runWith(sink)

Johny T Koshy · Answer 2 · 2021-10-26T19:23:56

С помощью GraphDSL, который является упрощенной версией реализации Source.combine:

val sg = Source.fromGraph(
  GraphDSL.create(){ implicit builder =>
    import GraphDSL.Implicits._

    val concat = builder.add(Concat[Int](2))

    source ~> flow1 ~> concat
    source ~> flow2 ~> concat

    SourceShape(concat.out)
  }
)

sg.runWith(sink)

Объединение двух потоков в потоке Акка

Вопрос

Лучший ответ

На других языках

Эта страница на других языках

Популярное в этой категории

Популярные вопросы в этой категории