Pool di lavoratori con Akka Streams

Question

Come descritto in akka streams documentation ho cercato di creare una riserva di lavoratori (flussi):

def balancer[In, Out](worker: Flow[In, Out, NotUsed], workerCount: Int): Flow[In, Out, NotUsed] = { 
    import GraphDSL.Implicits._ 

    Flow.fromGraph(GraphDSL.create() { implicit b => 
     val balancer = b.add(Balance[In](workerCount)) 
     val merge = b.add(Merge[Out](workerCount)) 

     for (_ <- 1 to workerCount) { 
     balancer ~> worker ~> merge 
     } 
     FlowShape(balancer.in, merge.out) 
    }) 
    } 

poi usato questa funzione per eseguire un flusso in parallelo:

def main(args: Array[String]) { 
    val system = ActorSystem() 
    implicit val mat = ActorMaterializer.create(system) 

    val flow = Flow[Int].map(e => { 
     println(e) 
     Thread.sleep(1000) // 1 second 
     e 
    }) 

    Source(Range.apply(1, 10).toList) 
     .via(balancer(flow, 3)) 
     .runForeach(e => {}) 
    } 

Ottengo l'output previsto 1, 2, 3, 4, 5, 6, 7, 8, 9 ma i numeri appaiono ad una velocità 1 al secondo (nessun parallelismo). Cosa sto facendo di sbagliato?

Answer 1

Come sottolineato da Endre Varga, il flusso stesso deve essere contrassegnato con .async.

Ma anche in questo caso, il comportamento non è deterministico perché gli stadi asincroni hanno una dimensione predefinita del buffer di 16 e il bilanciatore può inviare tutti i messaggi allo stesso operatore.

Di conseguenza, balancer ~> worker.async.addAttributes(Attributes.inputBuffer(1, 1)) ~> merge comporterebbe il comportamento desiderato.

Per una risposta data da un progetto di vedere: https://github.com/akka/akka/issues/20146#issuecomment-201381356

Answer 2

I documenti in quella sezione non sono aggiornati e verranno risolti nella prossima versione. Fondamentalmente tutto ciò che serve è chiamare .async sul flusso stesso. In questo modo, si disegna una "scatola" attorno al flusso (che si può immaginare come una scatola con una porta di input e output) che impedirà la fusione attraverso quella casella. In questo modo, essenzialmente, tutti i lavoratori saranno su attori dedicati. Il resto del grafico (gli stadi di trasmissione e fusione) condividerà un altro attore (non verranno eseguiti su attori separati, la casella asincrona protegge solo il flusso, le cose all'esterno saranno comunque fuse).