Trasformare gli stream

Akka Streams fornisce un ricco insieme di operatori per Source e Flow, simili alle API delle collezioni di Scala, ma eseguiti in modo asincrono e nel rispetto della backpressure.

Ogni operatore restituisce un nuovo blueprint, quindi le trasformazioni vengono composte in modo dichiarativo prima ancora che lo stream venga eseguito.

map applica una funzione sincrona a ogni elemento; filter elimina gli elementi che non soddisfano un predicato. Sono gli strumenti fondamentali per le trasformazioni elemento per elemento.

Entrambi mantengono l'ordine e propagano il completamento e gli errori a valle.

Quando un input deve produrre diversi output, usi mapConcat. Riceve una funzione che restituisce un iterable e appiattisce i risultati nello stream.

Restituire una collezione vuota equivale di fatto a eliminare l'elemento.

grouped(n) raggruppa gli elementi consecutivi in una Seq contenente fino a n elementi, utile per le scritture batch nei database. sliding(n) emette finestre sovrapposte.

Il raggruppamento riduce il sovraccarico per elemento nelle pipeline con molte operazioni di I/O.

scan emette l'accumulatore aggiornato dopo ogni elemento, creando uno stream dello stato in evoluzione. fold emette soltanto il valore accumulato finale, una volta completato lo stream a monte.

Usi scan per i contatori in tempo reale e fold per gli aggregati finali.

mapAsync(parallelism) chiama una funzione che restituisce un Future ed emette i risultati in ordine, eseguendo contemporaneamente fino a parallelism Future.

Lo usi per chiamate asincrone come le ricerche nei database o le richieste HTTP quando l'ordine è importante.

mapAsyncUnordered si comporta come mapAsync, ma emette ogni risultato appena viene completato, ignorando l'ordine degli input.

Può migliorare il throughput quando l'ordine non è importante per lo stage a valle, perché un Future lento non blocca più quelli più veloci.

Per le trasformazioni elemento per elemento che richiedono uno stato locale mutabile, statefulMapConcat crea un nuovo stato a ogni materializzazione e restituisce un iterable di output.

È il modo sicuro per mantenere contatori o buffer senza condividere lo stato tra diverse esecuzioni dello stream.

Gli stream possono trasformare i dati in base al tempo oltre che al contenuto. throttle limita la frequenza di emissione, groupedWithin raggruppa in base alla dimensione o al tempo trascorso e takeWithin limita la durata.

Questi operatori sono essenziali per limitare la frequenza delle API esterne.

Per impostazione predefinita, un'eccezione generata all'interno di un operatore causa il fallimento dell'intero stream. Una supervision strategy può invece resume (eliminando l'elemento non valido) o restart lo stage.

Applichi la strategia al Flow con withAttributes.

È possibile comporre piccoli Flow in Flow più grandi con via, ottenendo un unico Flow riutilizzabile. In questo modo ogni trasformazione rimane focalizzata e può essere testata autonomamente.

Il Flow composto ha il tipo di input del primo Flow e il tipo di output dell'ultimo.

Consideri le trasformazioni asincrone e le relative garanzie sull'ordine.

Ha esplorato gli operatori di trasformazione: map/filter elemento per elemento, mapConcat uno-a-molti, grouped per il raggruppamento, scan e fold per l'accumulazione e le operazioni asincrone con mapAsync.

Ha inoltre visto le trasformazioni con stato, gli operatori basati sul tempo come throttle, le supervision strategy per gli errori e la composizione dei Flow con via. Ora vedrà come la backpressure mantiene sicuri questi stage.