JAVA-B4
Parallel Computing (Parallelism)
Parallelism
Multithreading
Klassische Threads
Threads & Runnables
Overview
Thread (Begriff)
DruckThread
DruckRunnable
ThreadDemo
Output
Thread Lifecycle
Overview
start()
run()
join()
stop()?
StartRunJoinDemo Example
Von allein endende Threads
Threads mit while(criteria)
Daemon-Threads
Overview
ThreadDaemonDemo
Daemon (Summary)
Thread-Priority
Overview
t.setPriority(..)
MachineThread Klasse (Example)
ThreadPriorityDemo (Example)
Thread Notification
Grundlagen
wait
notify
notifyAll
sleep
sleep vs. wait
Thread Notification Demo
sychronized(monitor)
Notification/Monitor Zusammenfassung
Concurrent API
Vom Thread zur Concurrent API
Was ist Concurrency?
Was ist Concurrency?
Problemstellungen bei herkömmlichen Threads
Features der Concurrent API
Warum: Bedeutung des Umstiegs
Themenüberblick
Technischer Hintergrund
Thread.sleep(0, 500_000)
System.nanoTime()
Executors (Übersicht) #310
ExecutorService (Page)
Begriffsbestimmung
Excetutor
.execute()
ExecutorService (Begriff)
ExecutorService Interface Methoden
Mehrere Callables und Runnables auf einmal starten
Kontrolliertes Beenden
Executors-Utilily Klasse
newCachedThreadPool()
newSingleThreadExecutor()
newFixedThreadPool(int nThreads)
newWorkStealingPool()
newScheduledThreadPool(int corePoolSize)
Beispiel
ForkJoinPool, RecursiveTask, Executors Example#09
A | B1,B2,B3 | C mit Summenbildung.
Executor vs. ExecutorService Example#10
Executors.newFixedThreadPool(2)
Beispiel
Zusammenfassung Executors
Callable, Future #330
Callable Future (Begriff)
Callable Interface (Begriff)
Begriff
Callable erzeugen
Callable verwenden
Future
Beispiel
Future-Listen mit invokeAll
Was sind Future-Listen?
Wie erzeuge ich Future-Listen?
Wann blockt eine Future-Liste?
Beispiel mit 12 Callables in 4 Gruppen
Executors (#350) @! Bis auf Hinweis doppelt
Overview
Executor (Begriff)
ExecutorService (Begriff)
Executors.new~ Factory Methoden
Executors für ThreadPoolExecutor mit fixed Pool
ScheduledExecutorService
ScheduledExecutorService
Superklasse ExecutorService
schedule(*)
scheduleAtFixedRate(*)
scheduleWithFixedDelay(*)
ScheduledExecutorService (xmpl)
ProductClock(Main, Bsp.)
AbstractMachine**
AbstractMaterial**
Coal/Iron/Steel**
Coal-/Iron-/Steel-Machine**
Material**
ProductionBus-Machine**
Storage T**
ThreadPoolExecutor (Detail)
Executor Implementierung
ThreadPoolExecutor (Term)
ThreadPoolExecutor Beispiele
"FixedSizeExecutorExample" EXAMPLE C20
"CachedExecutorExample" EXAMPLE C21
SynchronousQueue
"ScheduledExecutorExample" EXAMPLE C22
"CustomSingleThreadExecutorRunnableComponent" EXAMPLE C23
"CustomPolicyRejectedExecutorExample" EXAMPLE C24
"PrioritizedCombinerRunnableComponent" EXAMPLE C30
"BufferingRejectedTasksExample" EXAMPLE C31
Java RejectedExecutionHandler
ThreadPoolExecutor FAQ
Synchronizer
Synchronizer (overview)
Synchronizer Begriff
CountDownLatch
CountDownLatch Example
CyclicBarrier
CyclicBarrier Example (@Zeit ausgeben!)
Phaser
Phaser Example(@Erl!)
Exchanger
Exchanger ping/pong
Semaphore
CompletableFuture
async!!!
SemaphoreCompletableFutureDemo (@!?)
Monitoring
JMX
ThreadPoolExecutor "live" überwachen
getActiveCount()
getPoolSize()
getCompletedTaskCount()
getQueue().size()
Concurrency
Example:DeterministicCopy @todo FILE LINK FEHLT!
Example:NonDeterministicCopy @todo FILE LINK FEHLT!
Kapitel java.util.concurrent.Flow @!fehltReactive Flow @!fehltPublisher TSubscriber TSubscription TProcressor TProcressor T
Overview
Callable & Future | Overview
Übersicht: Executor-Interfaces und Klassen ersetzen in Java Concurrency das herkömmliche thread.start() sowie mitunter hierbei zugleich auch das Problem, wie man diesen Start für viele Threads und auch deren Ende eigentlich kontrolliert steuert.
Das wichtigste Grundprinzip eines Executors vorab:
Man erzeugt diese über die Factory-Methoden von Executors.*
Callable & Future
Callable & Future | Begriffsbestimmung
Das Interface Callable
Ein Callable<T> ist ein funktionales Interface in Java, das einen Task mit Rückgabewert beschreibt:
public interface Callable<V> {
V call() throws Exception;
}
Es ist funktional: Man kann Lambda-Ausdrücke dafür verwenden.
Es liefert bei einer Ausführung dem im Typ-Parameter angegeben Datentyp mit einem Wert oder vom Prinzip her auch null.
Es kann checked Exceptions auswerten mit throw Exception.
Callable T ist (wie Runnable) ein funktionales Interface
Warum reicht ein Lambda-Ausdruck? Es hat genau genau eine einzige abstrakte Methode call(). Damit erfüllt es die es die Kriterien für die Lambda-Nutzung in Java.
Für die Deklaration eines Callables reicht ein Lambda-Ausdruck.
Callable<String> task = () -> "Ich bin ein Rückgabewert";
Für ein Callable muss man also keine Implementierungsklasse für das Interface suchen oder erstellen. Grundsätzlich ist das aber möglich. Sie muss nur eine call-Methode haben.
Die Java Virtual Machine oder auch Graal Virtual Machine erzeugt intern eine synthetische Instanz einer anonymen Klasse, die das Interface implementiert.
Man erhält also tatsächlich ein Objekt und damit eine Instanz einer Klasse, nur wurde diese Klasse als anonyme Klasse nur für dieses eine Objekt erzeugt.
Es gibt eine ganze Reihe von Unterschieden
Merkmal Runnable Callable<T>
Methode run() call()
Rückgabe void beliebiger Typ <T>
Exception keine throws Exception möglich
deklarierten
Exceptions
Nutzung Thread, Executor ExecutorService, Future
Typische Verwendung Task ohne Ergebnis Rechen-Task, IO-Abfrage, etc
Ein Future steht für ein Ergebnis in der Zukunft
Sobald ein Callable an einen ExecutorService übergeben wurde, erhält man ein Future<T> zurück.
Dieses Objekt repräsentiert das zukünftige Ergebnis der Berechnung, das noch nicht fertig ist.
Man kann auf dieses Ergebnis mit .get() zugreifen – entweder blockierend oder mit Timeout.
Alternativ erlaubt .isDone() eine Prüfung, ob die Berechnung bereits abgeschlossen ist.
Mit .cancel(true) kann ein laufender Task sogar abgebrochen werden – vorausgesetzt, er reagiert auf Unterbrechungen.
Der Future eignet sich besonders für komplexe Berechnungen, Web-Abfragen, Dateioperationen und alle Tasks, deren Ergebnis später benötigt wird.
Example
Callable & Future | Example
Den Unterschied zwischen einem Executor und einem ExecutorServices demonstriert unser Beispiel.
In beiden Fällen erfolgt der Zugriff auf die API über die
java.util.concurrent.Executor
Klasse, was es recht einfach macht, sich das zu merken.
package com.stuelken.java.b4.parallel.c11.concurrentcallablefuture;
import java.util.concurrent.*;
/**
* @author t2m
*/
public class CallableFutureDemo {
public static void main(String[] args) throws Exception {
System.out.println("main: Beginn");
// {#1} ExecutorService mit einem Pool
ExecutorService executor = Executors.newFixedThreadPool(2);
// {#2} Callable definieren – liefert String zurück
// List<Callable<String> tasks
Callable<String> task01 = () -> {
System.out.println("task01 () begonnen");
Thread.sleep(3000);
System.out.println("task01 () beendet");
return "Ergebnis vom Task01 auf Thread " + Thread.currentThread().getName();
};
Callable<String> task02 = () -> {
System.out.println("task02 () begonnen");
Thread.sleep(2000);
System.out.println("task02 () beendet");
return "Ergebnis vom Task02 auf Thread " + Thread.currentThread().getName();
};
Callable<String> task03 = () -> {
System.out.println("task03 () begonnen");
Thread.sleep(1000);
System.out.println("task03 () beendet");
return "Ergebnis vom Task03 auf Thread " + Thread.currentThread().getName();
};
// {#3} Task ausführen lassen und ein Future erhalten
Future<String> future01 = executor.submit(task01);
Future<String> future02 = executor.submit(task02);
Future<String> future03 = executor.submit(task03);
System.out.println("main: Task wurde eingereicht");
// {#4} Ergebnis synchron abrufen
String result01 = future01.get();
String result02 = future02.get();
String result03 = future03.get();
System.out.println("main: Ergebnis 01 erhalten " + result01);
System.out.println("main: Ergebnis 02 erhalten → " + result02);
System.out.println("main: Ergebnis 03 erhalten → " + result03);
// {#5} Pool schließen
executor.shutdown();
System.out.println("main: Ende");
}
}
// @formatting:off
/*
main: Beginn
main: Task wurde eingereicht
task01 () begonnen
task02 () begonnen
task02 () beendet
task03 () begonnen
task01 () beendet
task03 () beendet
main: Ergebnis 01 erhalten Ergebnis vom Task01 auf Thread pool-1-thread-1
main: Ergebnis 02 erhalten ? Ergebnis vom Task02 auf Thread pool-1-thread-2
main: Ergebnis 03 erhalten ? Ergebnis vom Task03 auf Thread pool-1-thread-2
main: Ende
*/
Warum startet Task 3 mit Verzögerung?
Wie es bei nebenläufiger Programmierung immer passieren kann: Die Entscheidung, wann ein Runnable oder hier ein Callable beginnen darf, entscheidet die Runtime.
Es kann also passieren, dass task01 und task02 starten, dann aber der task02 erst einmal seinen eigenen sleep() Befehl ausführt und dahingehend der Start von task03 tatsächlich warten muss.
Der sleep() Befehl eines jeden dieser 3 Tasks blockiert, wenn er ausgeführt wird.
Die Tasks werden also in diesem Beispiel nacheinander gestartet und damit nicht zeitgleich.
Wird in Reihenfolge gestartet?
Nein: Die Reihenfolge von .submit() entscheidet NICHT über den Startzeitpunkt eines Tasks sondern nur über den Zeitpunkt, wo der Task in die Task eingereiht wurde.
Aufgrund von newFixedThreadPool(2) werden maximal 2 Threads gleichzeitig ausgeführt. Deshalb greift sich der Executor zu Beginn beliebt zwei dieser Callables und startet. Es muss aber ein Callable beendet sein, bevor er den 3. Task starten kann.
Task 3 ist also in der Wartschlange, englisch Queue. Er muss aber warten, bis wieder ein Thread frei wird.
Merke: Es spielt keine Rolle, wie lang oder kurz ein Thread arbeitet. Wenn der Pool nur 2 Plätze hier im Beispiel frei hat, muss der 3. Task warten, bis er an der Reihe ist.
Laufen die Task nebeneinander oder hintereinander?
Sie laufen nebeneinander parallel, limitiert aber durch die maximale Anzahl, hier 2.
Könnten alle 3 auch zeitgleich starten?
Ja. Wollte man erreichen, dass alle drei Tasks sofort parallel starten, müsste man Executors.newFiexThreadPool(3) verwenden.
Man kann auch Callable T Instanzen mit invokeAll() auf einmal starten, um deren Ergebnissse als Liste zu erhalten.
Wann wird geblockt?
Future.get() blockiert
die Ausführung (hier im MainThread). So lange man also
den Wert eines Future nicht abruft blockiert auch die
Ausführung nicht.
Ein ähnliches Prinzip kennt man auch in JavaScript mit async.
Könnte man die Anzahl der Threads nachträglich erhöhen?
Ja, aber nicht mit jedem Executor. Executors.newFixedThreadPool(...) bewirkt, dass die Anzahl der Threads fix ist und sich auch nicht während der Ausführung verändern lässt.
Mit Hilfe eines ThreadPoolExecutor wäre das aber möglich. Das ist ein eigenständiger Themenblock.
Custom Callable Class
Callables & | Custom Callable Class
Man kann statt funktionalen Ausdrücken für Callables auch bewusst eigene Klassen implementieren, um beispielsweise dann auch Werte in deren Eigenschaften verwalten zu können.
Maßgebend ist hierbei, dass es eine call-Methode gibt welche im Falle von Callable T mit T als String dann einen String Wert liefert.
Callables müssen auch über call() Exception-Instanzen
als Ausnahmen auswerfen können und deshalb
throws Exception angeben.
Merke: Callables können Ausnahmen auswerfen und haben Rückgabewerte, Runnables haben nur void als Rückgabetyp, also nichts, und können auch keine geplanten kontrollierten Ausnahmen auswerfen.
(Auszug)
// Custom-Callable-Klasse
class CallableContainer {
public String call() throws Exception {
return "Hallo aus Methodenreferenz";
}
}
...
// Referenz
Callable<String> task = new CallableContainer()::call;
// Schleife
List<Callable<String>> tasks = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
tasks.add(new CallableContainer(i)::call);
}
UIO3 Es ist einfacher als Du denkst.
Stelle noch heute Deine Anfrage.