разница между весами и Broadcast веером в Akka Streams

Question

У меня есть немного путаницы с веером стратегий в Akka streams, я прочитал, что Broadcast - (1 вход, N выходов), данные входной элемент испускает к каждому выходу, а Balance - (1 вход, N выходов), если входной элемент испускает один из его выходных портов.

Можете ли вы объяснить мне:

1. Как баланс работает с несколькими потребителями?
2. Значение фразы "излучает на один из ее выходных портов"
3. Порт аналогичен нисходящему потоку?
4. Подходит ли «Баланс» для репликации входного потока в несколько разделов вывода.
5. Что означает «баланс позволяет графам разделить друг от друга, а несколько экземпляров последующих абонентов, реплицированных для обработки тома» означает?

Answer 1

Из документации ... радиовещание излучает (передает) элемент каждому потребителю. баланс исходит только от первого доступного потребителя.

broadcast

Emit каждый входящий элемент каждого из п выходов.

balance

Фан-поток из нескольких потоков. Каждый элемент выше по потоку - , испускаемый первому доступному нижерасположенному потребителю.

EDIT от комментариев:

С вашей сущности, вы должны сделать две averageCarrierDelay функции, по одному для каждого Z и F. Затем вы можете увидеть все элементы, отправленные каждому.

val averageCarrierDelayZ = 
    Flow[FlightDelayRecord] 
     .groupBy(30, _.uniqueCarrier) 
     .fold(("", 0, 0)){ 
      (x: (String, Int, Int), y:FlightDelayRecord) => { 
      println(s"Z Received Element: ${y}") 
      val count = x._2 + 1 
      val totalMins = x._3 + Try(y.arrDelayMins.toInt).getOrElse(0) 
      (y.uniqueCarrier, count, totalMins) 
      } 
     }.mergeSubstreams 


val averageCarrierDelayF = 
    Flow[FlightDelayRecord] 
     .groupBy(30, _.uniqueCarrier) 
     .fold(("", 0, 0)){ 
      (x: (String, Int, Int), y:FlightDelayRecord) => { 
      println(s"F Received Element: ${y}") 
      val count = x._2 + 1 
      val totalMins = x._3 + Try(y.arrDelayMins.toInt).getOrElse(0) 
      (y.uniqueCarrier, count, totalMins) 
      } 
     }.mergeSubstreams 

Edit 2: Для того, чтобы проверить вещи в будущем я бы рекомендовал универсальный регистратор для этапов потока, так что вы можете увидеть, что происходит.

def logElement[A](msg: String) = Flow[A].map { a => println(s"${msg} ${a}"); a } 

Это позволит вам сделать что-то вроде:

D ~> logElement[FlightDelayRecord]("F received: ") ~> F 
D ~> logElement[FlightDelayRecord]("Z received: ") ~> Z 

Таким образом, вы можете проверить области вашего графика для странного поведения, которые могут или не могут быть ожидающей.

Answer 2

Как и другие уже говорил, широковещательный испускает свой вклад в всех выходных портов, в то время как баланс испускает свой входной сигнал на один выходного порта на основе противодавления.

Когда вы используете GraphStage, вам нужно выбрать, какой порт вывода вы хотите использовать.Рассмотрим этот пример:

val q1 = Source.queue[Int](10, OverflowStrategy.fail) 
val q2 = Source.queue[Int](10, OverflowStrategy.fail) 
GraphDSL.create(q1, q2)(Keep.both) { implicit b => (input1, input2) => 
    import GraphDSL.Implicits._ 

    val broadcast = b.add(Broadcast[Int](2)) 
    val balance = b.add(Balance[Int](2)) 

    val consumer1, consumer2, consumer3, consumer4 = b.add(Sink.foreach[Int](println)) 

    input1 ~> broadcast.in 
    input2 ~> balance.in 

    broadcast.out(0) ~> consumer1 
    broadcast.out(1) ~> consumer2 

    balance.out(0) ~> consumer3 
    balance.out(1) ~> consumer4 

    ClosedShape 
} 

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

В этом конкретном случае, когда вы запускаете поток, элементы, проходящие через первый вход, будут переданы как consumer1, так и этап consumer2, поскольку этап трансляции копирует свой вход на все его выходы (и здесь два выхода), и элементы, проходящие через второй вход, будут равномерно распределены по consumer3 и consumer4 на основе скорости вашего терминала (т.е. скорость println), так как Sink.foreach противодавления, когда его функция выполняется в течение длительного времени.

Обратите внимание на то, что мы указали, что этапы вещания и баланса имеют по 2 порта каждый (при вызове их заводских методов), и что мы указали, какой выходной порт мы подключаем к какому потребителю.