3.6Datensynchronisation durch besondere Concurrency-Klassen *
Arbeiten mehrere Threads zusammen, so wollen sie in der Regel Daten austauschen und sich an bestimmten Bedingungen synchronisieren. Die Java-API bietet für diese Zusammenarbeit eine Reihe von Klassen:
Semaphore: Erlaubt eine maximale Anzahl Threads in einem Programmblock.
CyclicBarrier: Eine Menge von Threads wartet aufeinander, um zu einem gemeinsamen Punkt zu kommen.
CountDownLatch: Ein oder mehrere Threads warten auf eine Bedingung. Ist sie erfüllt, können die Threads fortfahren.
Exchanger: Zwei Threads treffen sich und tauschen Daten aus.
3.6.1Semaphor
Ein Semaphor[ 36 ](Semaphoren wurden 1968 vom niederländischen Informatiker Edsger Wybe Dijkstra eingeführt, also 10 Jahre vor Hoares Monitoren.) stellt sicher, dass nur eine bestimmte Anzahl von Threads auf ein Programmstück zugreift, und zählt damit zur Technik der Sperrmechanismen. Es lassen sich zwei Typen von Semaphoren unterscheiden:
Binäre Semaphoren lassen höchstens einen Thread auf ein Programmstück zu. Das bekannte Paar await(…)/signalXXX() bei Condition bzw. wait(…)/notifyXXX() von Object bietet sich für binäre Semaphoren an.
Allgemeine Semaphoren erlauben eine begrenzte Anzahl an Threads in einem kritischen Abschnitt. Das Semaphor verwaltet intern eine Menge so genannter Erlaubnisse (engl. permits).
Die Klasse Semaphore
Für allgemeine Semaphoren mit einer maximalen Anzahl Threads im Programmstück deklariert die Java-Bibliothek die Klasse java.util.concurrent.Semaphore.
Die wichtigen Eigenschaften der Semaphore-Klasse sind der Konstruktor und die Methoden zum Betreten und Verlassen des kritischen Abschnitts. Intern vermerkt das Semaphor jedes Betreten und lässt Threads warten, wenn das gesetzte Maximum erreicht ist, bis ein anderer Thread das Programmsegment verlässt.
implements Serializable
Semaphore(int permits)
Das neue Semaphor, das bestimmt, wie viele Threads in einem Block sein dürfen.void acquire()
Versucht, in den kritischen Block einzutreten. Wenn der gerade belegt ist, wird gewartet. Vermindert die Menge der Erlaubnisse um eins.void release()
Verlässt den kritischen Abschnitt und legt eine Erlaubnis zurück.
Allgemeine Semaphoren vereinfachen das Konsumenten-Produzenten-Problem, da eine bestimmte Anzahl von Threads in einem Block erlaubt ist. Die verbleibende Größe des Puffers ist somit automatisch die maximale Anzahl von Produzenten, die sich nebenläufig im Einfügeblock befinden können.
Unser Beispiel soll mit einem Semaphor arbeiten, das nur zwei Threads gleichzeitig in den kritischen Abschnitt lässt:
Listing 3.33com/tutego/insel/thread/concurrent/SemaphoreDemo.java, Teil 1
import java.util.concurrent.Semaphore;
public class SemaphoreDemo {
static Semaphore semaphore = new Semaphore( 2 );
Der kritische Abschnitt besteht aus zwei Operationen: einer Ausgabe auf dem Bildschirm und einer Wartezeit von 2 Sekunden. Er ist in einem Runnable eingebettet:
Listing 3.34com/tutego/insel/thread/concurrent/SemaphoreDemo.java, Teil 2
@Override public void run() {
while ( true ) {
try {
semaphore.acquire();
try {
System.out.println( "Thread=" + Thread.currentThread().getName() +
", Available Permits=" + semaphore.availablePermits() );
TimeUnit.SECONDS.sleep( 2 );
}
finally {
semaphore.release();
}
}
catch ( InterruptedException e ) {
e.printStackTrace();
Thread.currentThread().interrupt();
break;
}
}
}
};
Der kritische Abschnitt beginnt mit dem acquire() und endet mit release(). Wichtig ist die richtige Ausnahmebehandlung. Fünf Dinge müssen beachtet werden und formen den Quellcode:
Das acquire() kann eine InterruptedException auslösen, was eine geprüfte Ausnahme ist. Wir müssen sie folglich behandeln.
Da acquire() sowie sleep() eine InterruptedException auslösen können, ist die Frage, wie damit umzugehen ist. Im Prinzip signalisiert die Ausnahme die Bitte, das Warten zu beenden. Das wollen wir berücksichtigen, und wir brechen daher aus der Endlosschleife aus. Details zu der Ausnahme und zur interrupt()-Methode enthält Abschnitt 3.2.8, »Einen Thread höflich mit Interrupt beenden«.
Immer dann, wenn ein acquire() erfolgreich war, muss auch ein release() folgen. Das release() ist im finally sehr gut aufgehoben, denn wir wollen in jedem Fall die Semaphoren wieder freigeben, auch wenn irgendwie eine andere RuntimeException auftauchen sollte.
Ein release() darf nur dann erfolgen, wenn es ein zugehöriges acquire() gibt. Eine Programmierung wie try { semaphore.acquire(); … } finally { semaphore.release(); } ist unsicher, denn wenn ein acquire() wirklich eine Ausnahme erzeugt, wird fälschlicherweise ein release() ausgelöst.
Ein release() erzeugt keine Ausnahme, daher muss auch nichts behandelt werden.
Drei Threads sollen sich koordinieren:
Listing 3.35com/tutego/insel/thread/concurrent/SemaphoreDemo.java, Teil 3
new Thread( r ).start();
new Thread( r ).start();
new Thread( r ).start();
}
}
Nach dem Starten ist gut zu beobachten, wie jeweils zwei Threads im Abschnitt sind (eine Leerzeile symbolisiert die Wartezeit):
Thread=Thread-1, Available Permits=0
Thread=Thread-2, Available Permits=0
Thread=Thread-0, Available Permits=0
Thread=Thread-2, Available Permits=0
Thread=Thread-0, Available Permits=0
Fair und unfair
In der Ausgabe ist zu sehen, dass Thread 0, 1 und 2 zwar ihre Aufgaben ausführen können, aber plötzlich eine Sequenz 0, 2, 0 entsteht. Unser Gerechtigkeitssinn sagt uns jedoch, dass Thread 1 wieder an die Reihe kommen müsste. Wie ist das möglich? Die Antwort lautet, dass das acquire() nicht berücksichtigt, wer am längsten wartet, sondern dass es sich aus der Liste der Wartenden einen beliebigen Thread auswählt (wir kennen das von notify() her und dem Betreten eines synchronized-Blocks). Um ein faires Verhalten zu realisieren, wird die Fairness einfach über den Konstruktor von Semaphore angegeben. Ändern wir im Programm folgende Zeile:
Nun bekommen wir eine Ausgabe wie die folgende:
Thread=Thread-1, Available Permits=0
Thread=Thread-2, Available Permits=0
Thread=Thread-0, Available Permits=0
Thread=Thread-1, Available Permits=0
Thread=Thread-2, Available Permits=0
Thread=Thread-0, Available Permits=0
Thread=Thread-1, Available Permits=0
3.6.2Barrier und Austausch
Der Punkt, an dem alle Threads zusammenkommen und sich die Ergebnisse zusammenlegen lassen, heißt auf Englisch barrier. Die Klasse CyclicBarrier im Paket java.util.concurrent realisiert eine solche Barriere. Der Vorteil gegenüber join(…) besteht in der Tatsache, dass der für die Abarbeitung verwendete Thread nicht enden muss – und ein Thread im Thread-Pool endet eigentlich nicht –, sondern dass er mit awaitXXX(…) signalisiert, dass die Arbeit erledigt ist und er auf neue Arbeit wartet. Das folgende Beispiel zeigt anhand eines nebenläufigen Summierers die Funktionsweise:
Listing 3.36com/tutego/insel/thread/concurrent/ArraySummer.java, ArraySummer
public static void main( String[] args ) {
int[] array = new int[ 1000 ];
Random r = new Random();
for ( int i = 0; i < array.length; i++ )
array[ i ] = Math.abs( r.nextInt() / 2 );
parallSummer( array );
}
public static void parallSummer( int[] array ) {
int prozessors = 2; // Runtime.getRuntime().availableProcessors();
final List<Long> longs = new ArrayList<>();
Runnable merger = new Runnable() {
@Override public void run() {
long sum = 0;
for ( long i : longs )
sum += i;
System.out.println( sum );
}
};
CyclicBarrier barrier = new CyclicBarrier( prozessors, merger );
for ( int part = 0; part < prozessors; part++ )
new Thread( new AtomarSummer( barrier, array, prozessors, part,
longs)).start();
}
}
Listing 3.37com/tutego/insel/thread/concurrent/ArraySummer.java, AtomarSummer
private final CyclicBarrier barrier;
private final int[] array;
private final List<Long> longs;
private int start, end;
public AtomarSummer( CyclicBarrier barrier, int[] array, int maxPart,
int currentPart, List<Long> longs ) {
this.barrier = barrier;
this.array = array;
this.longs = longs;
start = (int) ((double) array.length / maxPart * currentPart);
end = (int) ((double) array.length / maxPart * (currentPart + 1) – 1);
}
@Override public void run() {
long sum = 0;
for ( int i = start; i < end; i++ )
sum += array[ i ];
longs.add( sum );
try {
barrier.await();
}
catch ( InterruptedException | BrokenBarrierException e ) {
e.printStackTrace();
}
}
}
3.6.3Stop and go mit Exchanger
Die generische Klasse java.util.concurrent.Exchanger dient ebenfalls dem Zusammenkommen von Threads, die jedoch bei ihrem Rendezvous Daten austauschen können. Ein üblicher Fall betrifft das Füllen von Puffern, etwa wenn ein Thread Daten vom Datensystem liest und ein anderer Thread die Daten über das Netzwerk weiterschickt. Ein Dateisystem-Thread füllt den Puffer, und wenn er komplett gefüllt ist, trifft er sich mit einem anderen Netzwerk-Thread, dem er den vollen Puffer gibt und von dem er wieder einen leeren Puffer empfängt. Der Netzwerk-Thread kann dann den Inhalt des Puffers wieder »verbrauchen« und der erste Thread den Puffer wieder mit Daten vom Dateisystem füllen.