Воспроизвести асинхронный вывод akka-stream

Я новичок в akka-stream, поэтому хочу спросить, как воспроизвести поведение, представленное в этой статье http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.2/scala/stream-rate.html < / а>

Для данного кода

Source(1 to 3)
  .map { i => println(s"A: $i"); i }
  .map { i => println(s"B: $i"); i }
  .map { i => println(s"C: $i"); i }
  .runWith(Sink.ignore)

Получите такие похожие

A: 1
A: 2
B: 1
A: 3
B: 2
C: 1
B: 3
C: 2
C: 3

Я пробовал добавить случайные Thread.sleep, создать поток из бесконечного итератора. Но Akka, согласно выводам отладки, всегда использует один и тот же поток для обработки.

Итак, вопрос: как воспроизвести асинхронное поведение (каждый этап должен выполняться асинхронно) с использованием akka-stream?


multithreading scala akka-stream
person alatom    schedule 07.02.2016    source источник
comment
Что вы имеете в виду под async? Какое поведение ожидается от вас?   -  person 1esha    schedule 08.02.2016
comment
что-то, кроме этого A: 1 B: 1 C: 1 A: 2 B: 2 C: 2 A: 3 B: 3 C: 3   -  person alatom    schedule 08.02.2016
comment
Тогда сделать все за один шаг карты?   -  person 1esha    schedule 08.02.2016

Ответы (1)


arrow_upward
2
arrow_downward

Причина, по которой вы видите последовательные операции, заключается в том, что все ваши операции выполняются из одного источника и, следовательно, находятся в пределах одной асинхронной границы. Чтобы получить "асинхронное поведение", которое вы ищете, вам нужно добавить Потоки:

implicit val actorSystem = ActorSystem()
implicit val actorMaterializer = ActorMaterializer()

Source(1 to 3).via(Flow[Int].map{i => println(s"A: $i"); i })
              .via(Flow[Int].map{i => println(s"B: $i"); i })
              .via(Flow[Int].map{i => println(s"C: $i"); i })
              .runWith(Sink.ignore)

Каждый Поток материализуется в отдельного Актера. Примечание: чтобы получить истинный параллелизм, пул потоков, в котором работает ActorSystem, должен иметь более 1 потока.

Следует помнить одну вещь: преимущество ActorSystem заключается в том, что она берет на себя ответственность за низкоуровневый контроль операций, так что разработчик может сосредоточиться на «бизнес-логике». Это тоже может быть недостатком. В зависимости от конфигурации ActorSystem, конфигурации JVM и конфигурации оборудования порядок операций может быть синхронным.

person Ramón J Romero y Vigil    schedule 08.02.2016
comment
Нет никаких гарантий, как это будет выполнено. Каждый может выполняться в отдельном потоке. Также следует помнить о буферизации - person 1esha; 08.02.2016
comment
@ 1esha, полностью согласен. Моя рекомендация просто ослабляет одно из ограничений, которое может привести к последовательной обработке. Я буду обновлять соответственно на основе вашего комментария. Спасибо. - person Ramón J Romero y Vigil; 08.02.2016