Java Goes Parallel

Die Klasse Arrays bietet für einige Standardfälle eine Parallelverarbeitung an: Für das Befüllen, das Sortieren und die Berechnung eines Prefixes. Hierzu stehen verschiedene Überladungen der Methoden parallelSetAll, parallelSort und parallelPrefix zur Verfügung. Codebeispiel 1 zeigt die Verwendung. Zuerst wird hier ein int-Array erzeugt und parallel mit Zufallszahlen gefüllt. Danach wird es sortiert und die additive Prefixsumme berechnet. Da das Befüllen parallel ausgeführt wird, benötigen jetzt mehrere Threads gleichzeitig Zugriff auf einen  Zufallszahlengenerator. Das hat zur Konsequenz, dass man hier nicht einfach ein gewöhnliches Random-Objekt benutzen sollte. Der Zugriff auf ein Random-Objekt ist zwar "threadsicher", aber konkurrierende Zugriffe werden beim Zugriff auf das Objekt intern sequentialisiert und behindern somit den parallelen Ablauf. Java stellt für diesen Anwendungsfall deshalb die Klasse ThreadLocalRandom zur Verfügung, über die jeder Thread Zugriff auf einen eigenen Zufallsgenerator erhält. Man sieht bereits an diesem einfachen Beispiel, dass man die Parallelisierung nicht unbedarft einsetzen darf. Allgemein gilt, dass die Anweisungen für die parallel ablaufenden Threads, in der Regel Lambda-Ausdrücke, seiteneffektfrei sein müssen, d. h. sie dürfen sich nicht gegenseitig in die Quere kommen. Seiteneffekte werden in der Regel vermieden, indem die Lambda-Ausdrücke nur lokale Variable, zustandslose Methoden oder nichtveränderbare Objekte benutzen. Die Benutzung von synchronisierenden bzw. sequentialisierenden Methoden ist zwar auch unkritisch, wirkt sich aber negativ auf die Parallelisierung aus.

Codebeispiel 1: Parallele Array-Verarbeitung

final int size = 1_000_000;
int[] array = new int[size];
    
Arrays.parallelSetAll(array, i -> ThreadLocalRandom.current().nextInt(100));
Arrays.parallelSort(array);
Arrays.parallelPrefix(array, (i,j) -> i + j );

An diesem Beispiel sieht man auch die Eleganz der Formulierung. Der Entwickler muss sich überhaupt keine Gedanken über die Threaderzeugung oder  -verwaltung machen. Zur Umsetzung der Parallelverarbeitung wird intern das ForkJoin-Framework verwendet, das bereits von Java 7 eingeführt und in Java 8 nochmal verbessert  wurde. Das ForkJoin-Framework teilt hier das zugrundeliegende Array rekursiv, bis zu einer systemabhängigen Tiefe, in Teilbereiche auf (Fork-Phase), die dann parallel bearbeitet werden (Work-Phase). Anschließend werden die Teilergebnisse wieder zusammengefügt (Join-Phase). In Abb.1 ist der Prozess schematisch dargestellt.

Der Clou bei der Sache ist nun, dass die Verarbeitung einfach automatisch parallelisiert werden kann, in dem man statt eines gewöhnlichen Streams einen parallelen Stream benutzt.

Codebeispiel 2 zeigt, wie man aus einem int-Array einen parallelen Stream (vom Typ IntStream) erzeugt und die kleinste gerade Zahl bestimmt. Der Lambda-Ausdruck (vom Typ Predicate) für den Filter mustert die ungeraden Zahlen aus und min bestimmt die kleinste Zahl von den verbleibenden. Die Auswertung liefert ein OptionalInt, da ja die Möglichkeit besteht, dass gar kein Minimum gefunden wird, falls z. B. das Array nur ungerade Zahlen enthält.

Codebeispiel 2: Array-Verarbeitung mit Streams

int[] array = ...;
    
OptionalInt min = Arrays.stream(array)
                        .parallel()
                        .filter( i -> i%2 == 0 )
                        .min();
    
min.ifPresent( i-> System.out.println("Minimum " + i ) );

Auch hier muss man sich keine Gedanken über Threaderzeugung und -verwaltung machen. Bei der Realisierung der parallelen Streamverarbeitung wird wie im vorherigen Beispiel intern das ForkJoin-Framework benutzt. Die Datenquelle wird rekursiv in Teilbereiche aufgeteilt (man spricht hier von der Split-Phase), die Verarbeitung parallel auf den Teilbereichen durchgeführt und anschließend das Ergebnis aus den Teilergebnissen ermittelt (man spricht hier von der Reduce- bzw. Collect-Phase). Abb.3 illustriert schematisch den Prozess.

Ob sich nun eine parallele Streamverarbeitung lohnt, hängt von vielen Faktoren ab. Die Wichtigsten sind:

• Die  Größe, d. h. die Anzahl der Daten. Sind nur wenige Daten zu bearbeiten, lohnt sich der Aufwand des Aufteilens des Datenbereichs und des Zusammenfügens der Teilergebnisse nicht.
• Der Aufwand der einzelnen Verarbeitungsschritte. Besteht die Arbeit der Threads nur aus kurzen und schnell zu erledigenden Arbeitsschritten, so lohnt sich in vielen Fällen eine Parallelisierung auch nicht.

Als Faustregel kann hier die NQ-Formel angewendet werden. Wenn N*Q > 10.000 ist, dann kann eine Parallelverarbeitung den Ablauf beschleunigen. In der Formel entspricht N der Anzahl der Datenelemente und Q dem Aufwand der Verarbeitung für ein Datenelement. Im Zweifel helfen entsprechende Benchmarkmessungen, um die richtige Entscheidung zu treffen.

Die Anwendung der NQ-Formel alleine reicht aber noch nicht aus. Es gibt noch zwei weitere wichtige Punkte, die beachtet werden müssen. Nämlich, ob sich die Datenquelle effizient aufteilen lässt und ob sich die Teilergebnisse effizient zusammenführen lassen.

Schauen wir uns zuerst den ersten Punkt, das Aufteilen der Datenquelle, etwas näher an. Hierzu betrachten wir als Datenquellen diesmal Collections, nämlich die beiden List-Implementierungen ArrayList und LinkedList. Bei einer ArrayList kann sehr schnell auf eine beliebige Elementposition zugegriffen werden, indem einfach die Speicheradresse ausgerechnet wird. Greift man dagegen bei einer LinkedList auf eine bestimmte Position zu, muss immer traversiert werden, d. h. man muss von vorne beginnend die Elemente durchzählen. Eine ArrayList lässt sich somit sehr effizient rekursiv in Teilbereiche aufteilen, eine LinkedList dagegen nicht. So ist z. B. die in Codebeispiel 3 gezeigte Streamverarbeitung basierend auf der ArrayList für große Collections deutlich schneller als die Verarbeitung auf der LinkedList (vgl. Abb.4).

Codebeispiel 3: Listen als Streamquellen

List<String>  arrayList  = new ArrayList<>();
List<String>  linkedList = new LinkedList<>();

// Initialisiere Listen mit Zufallsstrings
    
long count1 = arrayList.parallelStream()
                       .filter( s -> s.contains("abcde"))
                       .count();
    
long count2 = linkedList.parallelStream()
                        .filter( s -> s.contains("abcde"))
                        .count();

Weitere Streams, die sich nicht sonderlich gut aufteilen lassen, sind z. B. solche, die aus einem Iterationsprozess erzeugt oder aus Dateien gewonnen werden. Sie alle haben keine wahlfreie Zugriffseigenschaft.

Beim Zusammensetzen der Teilergebnisse (terminale Operation bzw. Auswertung des Streams) müssen wir zwischen einer Reduktion und einer Sammlung unterscheiden. Bei einer Reduktion wird typischerweise ein Wert ermittelt, wie z. B. die Anzahl der verbleibenden Elemente im Codebeispiel 3. Solche Operationen sind in der Regel eher unkritisch. Wird eine neue Datensammlung erzeugt, müssen in der Collect-Phase Sammlungen zusammengefügt werden. Das Zusammensetzen von großen Listen ist hier eher unkompliziert, solange die Sortierung keine Rolle spielt. In dem Fall können ArrayLists einfach hintereinander kopiert bzw. LinkedLists einfach verkettet werden. Das Zusammensetzen von großen Sets oder Maps ist dagegen mit wesentlich mehr Aufwand verbunden, da hier die Elemente explizit in eine vorhandene Struktur eingefügt werden müssen.  Bei dem im Codebeispiel 4 gezeigten Programm ist für große Ergebnislisten das parallele Sammeln in eine Liste deutlich schneller, wie das parallele Sammeln in eine Set, wobei natürliche die Ergebnisse nicht äquivalent sind, da eine Set keine doppelten Einträge zulässt. In dem Beispiel ist es sogar so, dass die gewöhnliche, nicht parallele Streamverabreitung beim Sammeln in einem Set doppelt so schnell ist wie ihre parallele Variante (vgl. Abb.5).

Codebeispiel 4: Streamverarbeitung mit Collectoren

List<String>  arrayList = new ArrayList<>();

// Initialisiere Listen mit Zufallsstrings

List<String> list = arrayList.parallelStream()
                             .filter( s -> s.contains("a"))
                             .collect( Collectors.toList() );
    
Set<String> set = arrayList.parallelStream()
                           .filter( s -> s.contains("a"))
                           .collect( Collectors.toSet() );

Um den beiden zuletzt genannten Punkten entgegenzuwirken, kann sowohl in den Split- als auch in den Collect-Prozess durch das Bereitstellen von eigenen Klassen, sogenannten Spliteratoren bzw. Collectoren, eingegriffen werden. Diese sollten aber mit Bedacht eingesetzt werden, da man sich dadurch in die Tiefen der nebenläufigen Programmierung begibt, die einige Fallstricke beinhaltet.

Ein weiterer Punkt, den man im Auge behalten sollte, ist das Speicherlayout der Daten. Der Einsatz von Parallelisierung kann bei einem ungünstigen Speicherlayout unter Umständen zu vielen "Cache Misses" führen, so dass die Sicherstellung der Cache-Kohärenz zu einem Ausbremsen der Anwendung führt. Bei Java hat man nicht wirklich viele Möglichkeiten, diesem Effekt entgegen zu wirken. Hier kann aber die geplante Einführung von Value Types Abhilfe schaffen.

Die Klasse CompletableFuture kann als Erweiterung des Future-Patterns angesehen werden, das von Java 5 zusammen mit den Threadpools eingeführt wurde. Beim Future-Pattern wird ein Task (Aufgabe) in Form einer Callable-Klasse implementiert, dessen Instanz an einen Threadpool (Executor) übergeben und von diesem asynchron ausgeführt wird. Vom Threadpool erhält man bei der Übergabe ein Future-Objekt zurück, das stellvertretend für das Ergebnis des Tasks steht. Über dieses Stellvertreterobjekt kann zu einem späteren Zeitpunkt auf das Ergebnis zugegriffen werden (vgl. Abb.6).

Mit Hilfe des Future-Patterns kann lediglich immer nur ein Task asynchron ausgeführt werden. Wenn der Task zu Ende ist, muss aktiv das Ergebnis ausgelesen und dann unter Umständen der nächste asynchrone Verarbeitungsschritt gestartet werden. Die CompletableFuture-Klasse hebt diese Limitierung auf, in dem zusätzlich zum Future-API eine Schnittstelle für Task-Kompositionen (CompletionStage-Interface) bereitgestellt wird. Somit lassen sich komplexere Ablaufpfade spezifizieren, die dann nach Wahl nebenläufig in Threads aus einem Threadpool oder synchron vom aufrufenden Thread ausgeführt werden (vgl. Abb.7).

Wenn in einer solchen Ablaufkette ein Task beendet ist, werden der oder die anhängigen Tasks automatisch gestartet, wobei das Ergebnis immer an den Nachfolger übergeben wird. Hierbei können auch zwei Ergebnisse durch UND oder ODER verknüpft werden. Bei einer UND-Verknüpfung erhält der Nachfolgetask  beide Ergebnisse der Vorgänger, bei einer ODER-Verknüpfung erhält er das Ergebnis, das zuerst vorliegt.

Codebeispiel 5 zeigt eine Implementierung des in Abb.7 gezeigten Ablaufs, wobei die Aufgaben der Tasks hier recht einfach gehalten sind.

Codebeispiel 5: Implementierung einer komplexen Ablaufkette

CompletableFuture<String> task1 = CompletableFuture.supplyAsync(() -> "hello");
CompletableFuture<String> task2 = CompletableFuture.supplyAsync(() -> "world");
    
task1.thenCombineAsync(task2, (s1, s2) -> String.join(" ", s1, s2) ) 
     .thenApplyAsync( String::toUpperCase )
     .thenAcceptAsync( System.out::println );

Zuerst werden zwei Tasks definiert, durch die jeweils der String "hello" bzw. der String "world" erzeugt wird. Die beiden Tasks werden anschließend durch eine "und"-Methode (thenCombineAsync) verknüpft, wobei hier ein neuer String erzeugt wird, der die beiden Ergebnisse der Vorgänger durch ein Leerzeichen getrennt verkettet. Danach wird der String noch in Großbuchstaben umgewandelt und anschließend auf der Konsole ausgegeben. Jeder einzelne Task wird hierbei asynchron in einem eigenen Thread ausgeführt. Der Thread, in dessen Code der Ablaufpfad definiert wird, kann ungehindert weiterlaufen.

Die Klasse CompletableFuture bietet neben verschiedenen Verknüpfungsoperationen für Tasks auch Synchronisationsbarrieren (vgl. Abb.8). So kann z. B. mit der statischen Methode anyOf auf das erste Ergebnis von verschiedenen, asynchron ausgeführten spekulativen Berechnungen gewartet werden. Die Berechnungen können hierbei asynchron ablaufen. In Codebeispiel 6 werden drei parallele Aufrufe von getValueFrom durchgeführt. Man beachte, dass die Methode threadsicher sein muss und möglichst keine Synchronisierung bzw. Sequentialisierung beinhalten sollte. Sobald einer der Return-Werte vorliegt, wird er ausgegeben.

Codebeispiel 6: Synchronisation paralleler spekulativer Berechnungen

CompletableFuture<String> task1 = CompletableFuture.supplyAsync( 
                                       () -> getValueFrom("goedel.hs-kl.de") );
CompletableFuture<String> task2 = CompletableFuture.supplyAsync( 
                                       () -> getValueFrom("escher.hs-kl.de") );
CompletableFuture<String> task3 = CompletableFuture.supplyAsync( 
                                       () -> getValueFrom("bach.hs-kl.de") );
    
CompletableFuture.anyOf(task1, task2, task3)

    .thenAcceptAsync( System.out::println );

Schaut man sich das API der Klasse CompletableFuture näher an, so erscheint es auf den ersten Blick recht unübersichtlich. Das liegt daran, dass es jede Methode in verschiedenen Varianten (Überladungen) gibt. Prinzipiell kann bei jeder Task-Ausführung festgelegt werden, ob er asynchron ausgeführt wird und wenn ja, welcher Threadpool benutzt werden soll. Weiter gibt es Ausführungsmethoden, die einen Wert liefern, wie z. B. thenApply oder Ergebnisse nur konsumieren, wie z. B. thenAccept. Ein Thema, das wir hier komplett ausgeklammert haben, ist die Fehlermeldung bzw. -behandlung. Auch hierzu bietet die Klasse CompletableFuture einige Konzepte an. Hat man den Aufbau des APIs erstmal durchschaut, lassen sich mit der Klasse bequem komplexe asynchrone Abläufe definieren und starten sowie komfortabel benutzbare asynchrone APIs entwickeln.

Resümee

Nachdem in den früheren Versionen Java um zahlreiche Concurrency-Tools erweitert wurde, stehen ab der Version 8 auch mächtige Frameworks zur Verfügung, mit deren Hilfe Anwendungen durch Parallelisierung einfacher beschleunigt werden können. Zu nennen sind hier insbesondere die parallelen Streams und die Klassen CompletableFuture. Da der Einsatz dieser Frameworks recht einfach erscheint, ist ihr Einsatz in der täglichen Arbeit verlockend. Doch Vorsicht, ganz so unbedarft darf man die Frameworks nicht einsetzen. Ein grundlegendes Verständnis der nebenläufigen Programmierung und der Funktionsweise der Frameworks ist für deren gewinnbringenden Einsatz unerlässlich. Wir haben hier anhand einfacher Beispiele nur einige zu beachtende Aspekte ansprechen können und hoffen, dass wir Sie für das Thema sensibilisiert haben. Hält man sich an das Programmiermodell und besitzt man ein Verständnis der internen Abläufe, steht aber dem Einsatz der Parallelisierungsframeworks in der Praxis nichts mehr im Weg.