Java Thread Pool Klasse #

Die Threads selbst zu verwalten ist nicht immer nötig und für kleinere Aufgaben auch oft zu teuer. Es gibt in Java den so genannten Thread Pool, dem Aufgaben übergeben werden können. Diese Aufgaben werden ausgeführt und es wird sofort ein Future<Ergebnis> zurückgegeben. Dieser Future enthält dann irgendwann das Ergebnis der Aufgabe oder “null”, falls kein Ergebnis gewünscht wurde.

ForkJoinPool threadPool = new ForkJoinPool();

Future<Integer> future = threadPool.submit(() -> {
int value = ...;
// long calculation
return value;
});

int result = future.get();

Es können dem ForkJoinPool aber auch ein RecursiveTask<Ergebnis> oder RecursiveAction übergeben werden.

class CountTask extends RecursiveTask<Integer> {
   // Constructor
   @Override
   protected Integer compute() {
      // if no or single element => return result
      // split into two parts
      CountTask left = new CountTask(lower, middle);
      CountTask right = new CountTask(middle, upper);
      left.fork(); right.fork();
      return right.join() + left.join();
   }
}

int result = threadPool.invoke(new CountTask(2, N)); //invoke blockiert und submit nicht

Der Vorteil der obigen Struktur ist die mögliche Rekursion. Es darf aber nicht davon ausgegangen werden, dass die Aufgaben der Reihe nach bearbeitet werden. Es dürfen also keine Abhängigkeiten zwischen verschiedenen Tasks bestehen.

Ab Java 8 müsste nicht mehr ein eigener ForkJoinPool instanziert werden, denn ein ForkJoinPool könnte auch so aufgerufen werden:

ForkJoinPool.commonPool().submit(...);
ForkJoinPool.commonPool().invoke(...);

new CountTask(2, N).invoke() //blockierend

Moderne Asynchronität - CompletableFuture #

Diese laufen auf dem ForkjoinPool.commonPool() und werden so augerufen:

CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> longOperation());
process(future.get()); //join() wirft nur eine unchecked exception

Eine Continuation ist wie folgt möglich:

future.thenApply(result -> "My result: " + result); //result ist "test"
future.thenAccept(result -> System.out.println(result)); //result ist "My result: test", thenAccept(...) liefert ein Future<Void> und thenApply(...) ein Future<Type>

Es gibt noch weitere interessante Funktionen wie zum Beispiel:

CompletableFuture.allOf(future1, future2).thenAccept(...);
CompletableFuture.any(future1, future2).thenAccept(...);

Irgendjemand muss aber einmal future.join() oder future.get() aufrufen, denn sonst kann es passieren, dass das Programm endet und die ThreadPool-Threads gekillt werden. Das passiert, weil die Threads im Pool als Daemon markiert sind.