| Titel | Inhalt | Suchen | Index | API | Go To Java 2, Zweite Auflage, Handbuch der Java-Programmierung |
| << | < | > | >> | Kapitel 22 - Multithreading | |
Wenn man sich mit Nebenläufigkeit beschäftigt, muß man sich in aller Regel auch mit Fragen der Synchronisation nebenläufiger Prozesse beschäftigen. In Java erfolgt die Kommunikation zweier Threads auf der Basis gemeinsamer Variablen, die von beiden Threads erreicht werden können. Führen beide Prozesse Änderungen auf den gemeinsamen Daten durch, so müssen sie synchronisiert werden, denn andernfalls können undefinierte Ergebnisse entstehen.
Wir wollen uns als einleitendes Beispiel ein kleines Programm ansehen, bei dem zwei Threads einen gemeinsamen Zähler hochzählen:
001 /* Listing2209.java */
002
003 public class Listing2209
004 extends Thread
005 {
006 static int cnt = 0;
007
008 public static void main(String[] args)
009 {
010 Thread t1 = new Listing2209();
011 Thread t2 = new Listing2209();
012 t1.start();
013 t2.start();
014 }
015
016 public void run()
017 {
018 while (true) {
019 System.out.println(cnt++);
020 }
021 }
022 }
|
Listing2209.java |
Läßt man das Programm eine Weile laufen, könnte es
beispielsweise zu folgender Ausgabe kommen:
0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
33 <-- Nanu? Wo ist die 32?
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
32 <-- Ach so, hier!
58
59
Beide Prozesse greifen unsynchronisiert auf die gemeinsame Klassenvariable cnt zu. Da die Operation System.out.println(cnt++); nicht atomar ist, kommt es zu dem Fall, daß die Operation mitten in der Ausführung unterbrochen wird und der Scheduler mit dem anderen Thread fortfährt. Erst später, wenn der unterbrochene Prozeß wieder Rechenzeit erhält, kann er seinen vor der Unterbrechung errechneten Zählerwert von 32 ausgeben. Sein Pendant war in der Zwischenzeit allerdings bereits bis 56 fortgefahren. Um diese Art von Inkonsistenzen zu beseitigen, bedarf es der Synchronisation der beteiligten Prozesse.
Zur Synchronisation nebenläufiger Prozesse hat Java das Konzept des Monitors implementiert. Ein Monitor ist die Kapselung eines kritischen Bereichs (also eines Programmteils, der nur von jeweils einem Prozeß zur Zeit durchlaufen werden darf) mit Hilfe einer automatisch verwalteten Sperre. Diese Sperre wird beim Betreten des Monitors gesetzt und beim Verlassen wieder zurückgenommen. Ist sie beim Eintritt in den Monitor bereits von einem anderen Prozeß gesetzt, so muß der aktuelle Prozeß warten, bis der Konkurrent die Sperre freigegeben und den Monitor verlassen hat.
Das Monitor-Konzept wird mit Hilfe des in die Sprache integrierten Schlüsselworts synchronized realisiert. Durch synchronized kann entweder eine komplette Methode oder ein Block innerhalb einer Methode geschützt werden. Der Eintritt in den so deklarierten Monitor wird durch das Setzen einer Sperre auf einer Objektvariablen erreicht. Bezieht sich synchronized auf eine komplette Methode, wird als Sperre der this-Pointer verwendet, andernfalls ist eine Objektvariable explizit anzugeben.
Wir wollen uns diese Art der Verwendung an einem Beispiel ansehen, welches das oben demonstrierte Synchronisationsproblem löst. Die naheliegende Lösung, die Anweisung System.out.println(cnt++); durch einen synchronized-Block auf der Variablen this zu synchronisieren, funktioniert leider nicht. Da der Zeiger this für jeden der beiden Threads, die ja unterschiedliche Instanzen repräsentieren, neu vergeben wird, wäre für jeden Thread der Eintritt in den Monitor grundsätzlich erlaubt.
Statt dessen verwenden wir die (in Abschnitt 43.2.2 erläuterte) Methode getClass, die uns ein Klassenobjekt beschafft (ein und dasselbe für alle Instanzen), mit dem wir die Klassenvariable cnt schützen können:
001 /* Listing2210.java */
002
003 public class Listing2210
004 extends Thread
005 {
006 static int cnt = 0;
007
008 public static void main(String[] args)
009 {
010 Thread t1 = new Listing2210();
011 Thread t2 = new Listing2210();
012 t1.start();
013 t2.start();
014 }
015
016 public void run()
017 {
018 while (true) {
019 synchronized (getClass()) {
020 System.out.println(cnt++);
021 }
022 }
023 }
024 }
|
Listing2210.java |
Nun werden alle Zählerwerte in aufsteigender Reihenfolge ausgegeben.
Ein anderer Fall ist der, bei dem der Zugriff auf ein Objekt selbst synchronisiert werden muß, weil damit zu rechnen ist, daß mehr als ein Thread zur gleichen Zeit das Objekt verwenden will.
Im folgenden werden die potentiellen Probleme am Beispiel eines Zählerobjekts erläutert, dessen Aufgabe es ist, einen internen Zähler zu kapseln, auf Anforderung den aktuellen Zählerstand zu liefern und den internen Zähler zu inkrementieren. Hierbei handelt es sich um eine Aufgabe, die beispielsweise in der Datenbankprogrammierung sehr häufig vorkommt, um Schlüsselnummern zu generieren.
Typischerweise wird das Synchronisationsproblem dadurch verschärft, daß die Verwendung des Zählers einige vergleichsweise langsame Festplattenzugriffe erforderlich macht. In unserem Beispiel wird der Zähler von fünf Threads verwendet. Die Langsamkeit und damit die Wahrscheinlichkeit, daß der Scheduler die Zugriffsoperation unterbricht, wird in unserem Beispiel durch eine Sequenz eingestreuter Fließkommaoperationen erhöht:
001 /* Listing2211.java */
002
003 class Counter2211
004 {
005 int cnt;
006
007 public Counter2211(int cnt)
008 {
009 this.cnt = cnt;
010 }
011
012 public int nextNumber()
013 {
014 int ret = cnt;
015 //Hier erfolgen ein paar zeitaufwendige Berechnungen, um
016 //so zu tun, als sei das Errechnen des Nachfolgezählers
017 //eine langwierige Operation, die leicht durch den
018 //Scheduler unterbrochen werden kann.
019 double x = 1.0, y, z;
020 for (int i= 0; i < 1000; ++i) {
021 x = Math.sin((x*%i35)*1.13);
022 y = Math.log(x+10.0);
023 z = Math.sqrt(x+y);
024 }
025 //Jetzt ist der Wert gefunden
026 cnt++;
027 return ret;
028 }
029 }
030
031 public class Listing2211
032 extends Thread
033 {
034 private String name;
035 private Counter2211 counter;
036
037 public Listing2211(String name, Counter2211 counter)
038 {
039 this.name = name;
040 this.counter = counter;
041 }
042
043 public static void main(String[] args)
044 {
045 Thread[] t = new Thread[5];
046 Counter2211 cnt = new Counter2211(10);
047 for (int i = 0; i < 5; ++i) {
048 t[i] = new Listing2211("Thread-"+i,cnt);
049 t[i].start();
050 }
051 }
052
053 public void run()
054 {
055 while (true) {
056 System.out.println(counter.nextNumber()+" for "+name);
057 }
058 }
059 }
|
Listing2211.java |
Das Ergebnis des Programms ist - wie nicht anders zu erwarten - schlecht,
denn es werden sehr viele doppelte Schlüssel produziert. Ein
Beispiellauf brachte bereits in den ersten 15 Aufrufen 6 doppelte
Zählerwerte:
10 for Thread-2
11 for Thread-4
10 for Thread-0
10 for Thread-1
11 for Thread-2
11 for Thread-3
12 for Thread-4
13 for Thread-0
14 for Thread-1
15 for Thread-2
16 for Thread-3
17 for Thread-4
18 for Thread-0
19 for Thread-1
20 for Thread-2
Auch hier gibt es eine einfache Lösung für das Synchronisationsproblem. Eine einfache Markierung der Methode nextNumber als synchronized macht diese zu einem Monitor und sorgt dafür, daß der komplette Code innerhalb der Methode als atomares Programmfragment behandelt wird. Eine Unterbrechung des kritischen Abschnitts durch einen anderen Thread ist dann nicht mehr möglich:
001 public synchronized int nextNumber()
002 {
003 int ret = cnt;
004 //Hier erfolgen ein paar zeitaufwendige Berechnungen, um so
005 //zu tun, als sei das Errechnen des Nachfolgezählerstandes
006 //eine langwierige Operation, die leicht durch den
007 //Scheduler unterbrochen werden kann.
008 double x = 1.0, y, z;
009 for (int i= 0; i < 1000; ++i) {
010 x = Math.sin((x*%i35)*1.13);
011 y = Math.log(x+10.0);
012 z = Math.sqrt(x+y);
013 }
014 //Jetzt ist der Wert gefunden
015 cnt++;
016 return ret;
017 }
|
Durch das synchronized-Attribut wird beim Aufruf der Methode die Instanzvariable this gesperrt und damit der Zugriff für andere Threads unmöglich gemacht. Erst nach Verlassen der Methode und Entsperren von this kann nextNumber wieder von anderen Threads aufgerufen werden.
 |
|
Diese Art des Zugriffschutzes wird in Java von vielen Klassen verwendet, um ihre Methoden thread-sicher zu machen. Nach Aussage der Sprachspezifikation kann davon ausgegangen werden, daß die gesamte Java-Klassenbibliothek in diesem Sinne thread-sicher ist. |
|
|
|
Neben dem Monitorkonzept stehen mit den Methoden wait und notify der Klasse Object noch weitere Synchronisationsprimitive zur Verfügung. Zusätzlich zu der bereits erwähnten Sperre, die einem Objekt zugeordnet ist, besitzt ein Objekt nämlich auch noch eine Warteliste. Dabei handelt es sich um eine (möglicherweise leere) Menge von Threads, die vom Scheduler unterbrochen wurden und auf ein Ereignis warten, um fortgesetzt werden zu können.
Sowohl wait als auch notify dürfen nur aufgerufen werden, wenn das Objekt bereits gesperrt ist, also nur innerhalb eines synchronized-Blocks für dieses Objekt. Ein Aufruf von wait nimmt die bereits gewährten Sperren (temporär) zurück und stellt den Prozeß, der den Aufruf von wait verursachte, in die Warteliste des Objekts. Dadurch wird er unterbrochen und im Scheduler als wartend markiert. Ein Aufruf von notify entfernt einen (beliebigen) Prozeß aus der Warteliste des Objekts, stellt die (temporär) aufgehobenen Sperren wieder her und führt ihn dem normalen Scheduling zu. wait und notify sind damit für elementare Synchronisationsaufgaben geeignet, bei denen es weniger auf die Kommunikation als auf die Steuerung der zeitlichen Abläufe ankommt.
Das folgende Beispiel demonstriert den Einsatz von wait und notify an einem Producer/Consumer-Beispiel. Ein Prozeß arbeitet dabei als Produzent, der Fließkommazahlen »herstellt«, und ein anderer als Konsument, der die produzierten Daten verbraucht. Die Kommunikation zwischen beiden erfolgt über ein gemeinsam verwendetes Vector-Objekt, das die produzierten Elemente zwischenspeichert und als Medium für die wait-/notify-Aufrufe dient:
001 /* Listing2213.java */
002
003 import java.util.*;
004
005 class Producer2213
006 extends Thread
007 {
008 private Vector v;
009
010 public Producer2213(Vector v)
011 {
012 this.v = v;
013 }
014
015 public void run()
016 {
017 String s;
018
019 while (true) {
020 synchronized (v) {
021 s = "Wert "+Math.random();
022 v.addElement(s);
023 System.out.println("Produzent erzeugte "+s);
024 v.notify();
025 }
026 try {
027 Thread.sleep((int)(100*Math.random()));
028 } catch (InterruptedException e) {
029 //nichts
030 }
031 }
032 }
033 }
034
035 class Consumer2213
036 extends Thread
037 {
038 private Vector v;
039
040 public Consumer2213(Vector v)
041 {
042 this.v = v;
043 }
044
045 public void run()
046 {
047 while (true) {
048 synchronized (v) {
049 if (v.size() < 1) {
050 try {
051 v.wait();
052 } catch (InterruptedException e) {
053 //nichts
054 }
055 }
056 System.out.print(
057 " Konsument fand "+(String)v.elementAt(0)
058 );
059 v.removeElementAt(0);
060 System.out.println(" (verbleiben: "+v.size()+")");
061 }
062 try {
063 Thread.sleep((int)(100*Math.random()));
064 } catch (InterruptedException e) {
065 //nichts
066 }
067 }
068 }
069 }
070
071 public class Listing2213
072 {
073 public static void main(String[] args)
074 {
075 Vector v = new Vector();
076
077 Producer2213 p = new Producer2213(v);
078 Consumer2213 c = new Consumer2213(v);
079 p.start();
080 c.start();
081 }
082 }
|
Listing2213.java |
Um die Arbeitsverteilung zwischen den Prozessen etwas interessanter
zu gestalten, werden beide gezwungen, nach jedem Schritt eine kleine
Pause einzulegen. Da die Wartezeit zufällig ausgewählt wird,
kann es durchaus dazu kommen, daß der Produzent eine größere
Anzahl an Elementen anhäuft, die der Konsument noch nicht abgeholt
hat. Der umgekehrte Fall ist natürlich nicht möglich, da
der Konsument warten muß, wenn keine Elemente verfügbar
sind. Eine Beispielsitzung könnte etwa so aussehen:
Produzent erzeugte Wert 0.09100924684649958
Konsument fand Wert 0.09100924684649958 (verbleiben: 0)
Produzent erzeugte Wert 0.5429652807455857
Konsument fand Wert 0.5429652807455857 (verbleiben: 0)
Produzent erzeugte Wert 0.6548096532111007
Konsument fand Wert 0.6548096532111007 (verbleiben: 0)
Produzent erzeugte Wert 0.02311095955845288
Konsument fand Wert 0.02311095955845288 (verbleiben: 0)
Produzent erzeugte Wert 0.6277057416210464
Konsument fand Wert 0.6277057416210464 (verbleiben: 0)
Produzent erzeugte Wert 0.6965546173953919
Produzent erzeugte Wert 0.6990053250441516
Produzent erzeugte Wert 0.9874467815778902
Produzent erzeugte Wert 0.12110075531692543
Produzent erzeugte Wert 0.5957795111549329
Konsument fand Wert 0.6965546173953919 (verbleiben: 4)
Produzent erzeugte Wert 0.019655027417308846
Konsument fand Wert 0.6990053250441516 (verbleiben: 4)
Konsument fand Wert 0.9874467815778902 (verbleiben: 3)
Produzent erzeugte Wert 0.14247583735074354
Konsument fand Wert 0.12110075531692543 (verbleiben: 3)
|
|
Durch eine konstante Pause nach jedem produzierten Element könnte der Produzent bewußt langsamer gemacht werden. Der schnellere Konsument würde dann einen Großteil seiner Zeit damit verbringen, festzustellen, daß keine Elemente verfügbar sind. Zwar würde das Beispiel (in leicht modifizierter Form) auch ohne den Einsatz von wait/notify funktionieren. Durch ihre Verwendung aber ist der Konsumentenprozeß nicht gezwungen, aktiv zu warten, sondern wird vom Produzenten benachrichtigt, wenn ein neues Element verfügbar ist. Der Rechenzeitbedarf reduziert sich dadurch auf einen Bruchteil dessen, was andernfalls benötigt würde. |
|
|
|
Sollen wie im vorigen Beispiel zwei Threads so miteinander verbunden werden, daß einer von beiden Daten erzeugt, die der andere verarbeitet, gibt es die Möglichkeit, beide mit Hilfe einer Pipe zu synchronisieren. Dabei werden die beiden Threads über einen ByteStream miteinander verbunden, der von einem Thread geschrieben und von dem anderen gelesen wird.
Dieses Piping-Konzept wird in Java durch die Klassen PipedInputStream und PipedOutputStream realisiert. Beide Klassen werden immer paarweise und immer in getrennten Threads verwendet. Daten, die der eine Thread in den PipedOutputStream schreibt, kann der andere aus dem angebundenen PipedInputStream lesen.
Da die Kommunikation gepuffert erfolgt, kann der schreibende Thread in einem gewissen Rahmen mehr Daten produzieren, als der lesende verarbeiten kann. Ist der Puffer voll (im JDK 1.2 ist er 1024 Byte groß), wird der schreibende Thread angehalten, bis der lesende ausreichend Zeichen gelesen hat. Greift der lesende Thread auf eine Pipe zu, die nicht genügend Daten enthält, muß er warten, bis ein anderer Thread die erforderliche Anzahl an Bytes hineingeschrieben hat.
Wir können das Producer-/Consumer-Beispiel unter Verwendung von PipedInputStream und PipedOutputStream vereinfachen, denn die gesamte Synchronisationsarbeit wird automatisch beim Aufruf der read- und write-Methoden erledigt.
Das folgende Listing zeigt zwei Threads, die über eine Pipe einzelne Datenbytes miteinander austauschen. Die Pipe wird vom Hauptprogramm erzeugt, indem zunächst der PipedInputStream angelegt und an den danach erzeugten PipedOutputStream übergeben wird. Alternativ hätte die Verbindung auch hergestellt werden können, indem einer der beiden Streams an die connect-Methode des anderen übergeben worden wäre.
001 /* Listing2214.java */
002
003 import java.io.*;
004
005 class Producer2214
006 extends Thread
007 {
008 private PipedOutputStream pipe;
009
010 public Producer2214(PipedOutputStream pipe)
011 {
012 this.pipe = pipe;
013 }
014
015 public void run()
016 {
017 while (true) {
018 byte b = (byte)(Math.random() * 128);
019 try {
020 pipe.write(b);
021 System.out.println("Produzent erzeugte " + b);
022 } catch (IOException e) {
023 System.err.println(e.toString());
024 }
025 try {
026 Thread.sleep((int)(100*Math.random()));
027 } catch (InterruptedException e) {
028 //nichts
029 }
030 }
031 }
032 }
033
034 class Consumer2214
035 extends Thread
036 {
037 private PipedInputStream pipe;
038
039 public Consumer2214(PipedInputStream pipe)
040 {
041 this.pipe = pipe;
042 }
043
044 public void run()
045 {
046 while (true) {
047 try {
048 byte b = (byte)pipe.read();
049 System.out.println(" Konsument fand " + b);
050 } catch (IOException e) {
051 System.err.println(e.toString());
052 }
053 try {
054 Thread.sleep((int)(100*Math.random()));
055 } catch (InterruptedException e) {
056 //nichts
057 }
058 }
059 }
060 }
061
062 public class Listing2214
063 {
064 public static void main(String[] args)
065 throws Exception
066 {
067 PipedInputStream inPipe = new PipedInputStream();
068 PipedOutputStream outPipe = new PipedOutputStream(inPipe);
069 Producer2214 p = new Producer2214(outPipe);
070 Consumer2214 c = new Consumer2214(inPipe);
071 p.start();
072 c.start();
073 }
074 }
|
Listing2214.java |
|
|
|
|
Piping gibt es auch für zeichenbasierte Kommunikation. Dazu werden die Klassen PipedWriter und PipedReader verwendet. Bis auf den Unterschied, daß Zeichen anstelle von Bytes ausgetauscht werden, entspricht ihre Arbeitsweise genau den hier vorgestellten Byte-Pipes. |
|
|
|
| Titel | Inhalt | Suchen | Index | API | Go To Java 2, Zweite Auflage, Addison Wesley, Version 2.0 |
| << | < | > | >> | © 2000 Guido Krüger, http://www.gkrueger.com | |