Streams transformieren

Akka Streams bietet eine umfangreiche Sammlung von Operatoren für Source und Flow, die der API von Scala-Sammlungen ähnelt, aber asynchron ausgeführt wird und Backpressure berücksichtigt.

Jeder Operator gibt einen neuen Bauplan zurück. Dadurch werden Transformationen deklarativ zusammengestellt, bevor der Stream überhaupt ausgeführt wird.

map wendet eine synchrone Funktion auf jedes Element an; filter verwirft Elemente, für die ein Prädikat nicht erfüllt ist. Diese beiden Operatoren bilden das grundlegende Werkzeug für elementweise Transformationen.

Beide behalten die Reihenfolge bei und leiten Abschluss und Fehler an nachgelagerte Stufen weiter.

Wenn aus einer Eingabe mehrere Ausgaben entstehen sollen, verwenden Sie mapConcat. Der Operator erwartet eine Funktion, die ein iterierbares Ergebnis zurückgibt, und flacht die Ergebnisse zu einem Stream ab.

Das Zurückgeben einer leeren Sammlung verwirft das Element effektiv.

grouped(n) bündelt aufeinanderfolgende Elemente in einer Seq mit bis zu n Elementen, was beispielsweise für umfangreiche Datenbankschreibvorgänge nützlich ist. sliding(n) gibt sich überlappende Fenster aus.

Das Bündeln verringert den Aufwand pro Element in I/O-intensiven Pipelines.

scan gibt nach jedem Element den aktuellen Akkumulator aus und erzeugt so einen sich verändernden Zustandsstream. fold gibt erst dann einmalig den endgültig akkumulierten Wert aus, wenn der Upstream abgeschlossen ist.

Verwenden Sie scan für Live-Zähler und fold für abschließende Aggregate.

mapAsync(parallelism) ruft eine Funktion auf, die einen Future zurückgibt, und gibt die Ergebnisse in der richtigen Reihenfolge aus. Dabei werden bis zu parallelism Futures gleichzeitig ausgeführt.

Verwenden Sie den Operator für asynchrone Aufrufe wie Datenbankabfragen oder HTTP-Anfragen, wenn die Reihenfolge wichtig ist.

mapAsyncUnordered verhält sich wie mapAsync, gibt jedoch jedes Ergebnis sofort nach dessen Fertigstellung aus und ignoriert dabei die Reihenfolge der Eingaben.

Das kann den Durchsatz erhöhen, wenn die nachgelagerte Verarbeitung keine bestimmte Reihenfolge benötigt, da ein langsamer Future schnellere Ergebnisse nicht mehr blockiert.

Für elementweise Transformationen, die einen veränderlichen lokalen Zustand benötigen, erstellt statefulMapConcat pro Materialisierung einen neuen Zustand und gibt ein iterierbares Ergebnis von Ausgaben zurück.

Damit können Sie Zähler oder Puffer sicher verwalten, ohne den Zustand zwischen verschiedenen Stream-Ausführungen zu teilen.

Streams können sowohl anhand der Zeit als auch anhand ihres Inhalts transformiert werden. throttle begrenzt die Ausgaberate, groupedWithin bündelt nach Größe oder verstrichener Zeit, und takeWithin begrenzt die Dauer.

Diese Operatoren sind für die Begrenzung der Rate externer APIs unverzichtbar.

Eine Ausnahme innerhalb eines Operators schlägt standardmäßig im gesamten Stream fehl. Eine Supervisionsstrategie kann stattdessen resume verwenden, um das fehlerhafte Element zu verwerfen, oder die Stufe mit restart neu starten.

Hängen Sie die Strategie mit withAttributes an den Flow an.

Kleine Flows lassen sich mit via zu größeren Flows zusammensetzen, wodurch ein einzelner wiederverwendbarer Flow entsteht. So bleibt jede Transformation überschaubar und kann unabhängig getestet werden.

Der zusammengesetzte Flow hat den Eingabetyp des ersten und den Ausgabetyp des letzten Flows.

Sie haben Transformationsoperatoren kennengelernt: elementweise map/filter-Transformationen, Eins-zu-viele-Transformationen mit mapConcat, Bündelung mit grouped, Akkumulation mit scan und fold sowie asynchrone Verarbeitung mit mapAsync.

Außerdem haben Sie zustandsbehaftete Transformationen, zeitbasierte Operatoren wie throttle, Supervisionsstrategien für Fehler und das Zusammensetzen von Flows mit via kennengelernt. Als Nächstes geht es darum, wie Backpressure diese Stufen sicher hält.