線程間的協(xié)作

理解了線程會(huì)相互沖突以及該如何防止這種沖突之后，下一步就該學(xué)習(xí)怎樣讓線程協(xié)同工作了。要做到這一點(diǎn)，關(guān)鍵是要讓線程能相互"協(xié)商(handshaking)"。而這個(gè)任務(wù)要由Object的wait( )和notify( )來(lái)完成。

wait與notify

首先要強(qiáng)調(diào)，線程sleep( )的時(shí)候并不釋放對(duì)象的鎖，但是wait( )的時(shí)候卻會(huì)釋放對(duì)象的鎖。也就是說(shuō)在線程wait( )期間，別的線程可以調(diào)用它的synchronized方法。當(dāng)線程調(diào)用了某個(gè)對(duì)象wait( )方法之后，它就中止運(yùn)行并釋放那個(gè)對(duì)象鎖了。

Java有兩種wait( )。第一種需要一個(gè)以毫秒記的時(shí)間作參數(shù)，它的意思和sleep( )一樣，都是:"暫停一段時(shí)間。"區(qū)別在于:

1. wait( )會(huì)釋放對(duì)象的鎖。
2. 除了時(shí)間到了，wait( )還可以用notify( )或notifyAll( )來(lái)中止

第二種wait( )不需要任何參數(shù)；它的用途更廣。線程調(diào)用了這種wait( )之后，會(huì)一直等下去，直到(有別的線程調(diào)用了這個(gè)對(duì)象的)notify( )或notifyAll( )。

和sleep( )屬于Thread不同，wait( )， notify( )， 和notifyAll( )是根Object的方法。雖然這樣做法(把專為多線程服務(wù)的方法放到通用的根類里面)看上去有些奇怪，但卻是必要的。因?yàn)樗鼈兯倏氐氖敲總€(gè)對(duì)象都會(huì)有的鎖。所以結(jié)論就是，你可以在類的synchronized方法里調(diào)用wait( )，至于它繼不繼承Thread，實(shí)沒(méi)實(shí)現(xiàn)Runnable已經(jīng)無(wú)所謂了。實(shí)際上你也只能在synchronized方法里或synchronized段里調(diào)用wait( )，notify( )或notifyAll( )(sleep( )則沒(méi)有這個(gè)限制，因?yàn)樗粚?duì)鎖進(jìn)行操作)。如果你在非synchronized方法里調(diào)用了這些方法，程序還是可以編譯的，但是一運(yùn)行就會(huì)出一個(gè)IllegalMonitorStateException。這個(gè)異常帶著一個(gè)挺讓人費(fèi)解的"current thread not owner"消息。這個(gè)消息的意思是，如果線程想調(diào)用對(duì)象的wait( )， notify( )，或notifyAll( )方法，必須先"擁有"(得到)這個(gè)對(duì)象的鎖。

你可以讓另一個(gè)對(duì)象來(lái)操控這個(gè)對(duì)象的鎖。要想這么做，第一步是先獲取對(duì)象的鎖。比方說(shuō)要想調(diào)用對(duì)象x的notify( )，可以在x的synchronized段里：

synchronized(x) {
x.notify();
}

通常情況下，如果條件是由方法之外的其他力量所控制的(最常見的就是要由其他線程修改)，那么你就應(yīng)該用wait( )。你總不希望閑著的時(shí)候還讓線程不停地作測(cè)試吧；這種"瞎忙活"是對(duì)CPU資源的巨大浪費(fèi)。所以wait( )能讓你在等待世道改變的同時(shí)讓線程休眠，當(dāng)(其他線程調(diào)用了對(duì)象的)notify( )或notifyAll( )的時(shí)候，線程自會(huì)醒來(lái)，然后檢查條件是不是改變了。所以說(shuō)wait( )提供了一種同步線程間的活動(dòng)的方法。

舉個(gè)例子，假設(shè)有這么一間飯店，它有一個(gè)廚師和一個(gè)服務(wù)員。服務(wù)員必須等廚師燒菜。廚師做完一道菜就通知服務(wù)員，服務(wù)員上完菜接著等。這是一個(gè)很精彩的線程合作的例子：廚師就是producer，而服務(wù)員就是consumer。下面我們用代碼來(lái)為這個(gè)故事建模：

//: c13:Restaurant.java
// The producer-consumer approach to thread cooperation.
import com.bruceeckel.simpletest.*;
class Order {
private static int i = 0;
private int count = i++;
public Order() {
if(count == 10) {
System.out.println("Out of food, closing");
System.exit(0);
}
}
public String toString() { return "Order " + count; }
}
class WaitPerson extends Thread {
private Restaurant restaurant;
public WaitPerson(Restaurant r) {
restaurant = r;
start();
}
public void run() {
while(true) {
while(restaurant.order == null)
synchronized(this) {
try {
wait();
} catch(InterruptedException e) {
throw new RuntimeException(e);
}
}
System.out.println(
"Waitperson got " + restaurant.order);
restaurant.order = null;
}
}
}
class Chef extends Thread {
private Restaurant restaurant;
private WaitPerson waitPerson;
public Chef(Restaurant r, WaitPerson w) {
restaurant = r;
waitPerson = w;
start();
}
public void run() {
while(true) {
if(restaurant.order == null) {
restaurant.order = new Order();
System.out.print("Order up! ");
synchronized(waitPerson) {
waitPerson.notify();
}
}
try {
sleep(100);
} catch(InterruptedException e) {
throw new RuntimeException(e);
}
}
}
}
public class Restaurant {
private static Test monitor = new Test();
Order order; // Package access
public static void main(String[] args) {
Restaurant restaurant = new Restaurant();
WaitPerson waitPerson = new WaitPerson(restaurant);
Chef chef = new Chef(restaurant, waitPerson);
monitor.expect(new String[] {
"Order up! Waitperson got Order 0",
"Order up! Waitperson got Order 1",
"Order up! Waitperson got Order 2",
"Order up! Waitperson got Order 3",
"Order up! Waitperson got Order 4",
"Order up! Waitperson got Order 5",
"Order up! Waitperson got Order 6",
"Order up! Waitperson got Order 7",
"Order up! Waitperson got Order 8",
"Order up! Waitperson got Order 9",
"Out of food, closing"
}, Test.WAIT);
}
} ///:~

Order是一個(gè)會(huì)自己計(jì)數(shù)的類，同時(shí)它還負(fù)責(zé)中止程序；數(shù)到10的時(shí)候，它會(huì)調(diào)用System.exit( )。

WaitPerson必須知道它是為哪家Restaurant服務(wù)的，所以他得從那家飯店的"帳臺(tái)"，也就是restaurant.order，拿單子。WaitPerson的run( )調(diào)用了wait( )方法，所以除非Chef用notify( )叫醒他，否則他會(huì)一直等下去。這是個(gè)非常簡(jiǎn)單的程序，所以我們知道只會(huì)有一個(gè)線程在等WaitPerson的鎖:就是WaitPerson線程自己，所以notify( )是安全的。如果情況比較復(fù)雜，有很多線程在等一個(gè)對(duì)象的鎖，那么你可能就不知道該叫醒哪個(gè)線程了。遇到這種情況，可以用notifyAll( )，這個(gè)方法會(huì)叫醒所有在等這個(gè)對(duì)象的鎖的線程。這樣線程就能自行判斷本次喚醒是否與自己有關(guān)。

請(qǐng)注意，wait( )所在的while循環(huán)，它的測(cè)試條件正是wait要等的東西。乍看上去真有點(diǎn)奇怪——如果你等的是訂單，那么醒過(guò)來(lái)之后肯定會(huì)有訂單，難道不是嗎？問(wèn)題在于在多線程環(huán)境下，還沒(méi)等WaitPerson醒過(guò)來(lái)，訂單就可能被其他線程給搶走了。所以唯一安全的做法就是套用下面這個(gè)wait( )定式:

while(conditionIsNotMet)
wait( );

用管道進(jìn)行線程間的I/O操作

在很多情況下，線程也可以利用I/O來(lái)進(jìn)行通信。多線程類庫(kù)會(huì)提供一種"管道(pipes)"來(lái)實(shí)現(xiàn)線程間的I/O。對(duì)Java I/O類庫(kù)而言，這個(gè)類就是PipedWriter(可以讓線程往管道里寫數(shù)據(jù))和PipedReader(讓另一個(gè)線程從這個(gè)管道里讀數(shù)據(jù))。你可以把它理解成"producer-consumer"問(wèn)題的一個(gè)變型，而管道則提供了一個(gè)現(xiàn)成的解決方案。

下面我們來(lái)舉一個(gè)簡(jiǎn)單的例子。兩個(gè)線程利用管道來(lái)進(jìn)行通信：

//: c13:PipedIO.java
// Using pipes for inter-thread I/O
import java.io.*;
import java.util.*;
class Sender extends Thread {
private Random rand = new Random();
private PipedWriter out = new PipedWriter();
public PipedWriter getPipedWriter() { return out; }
public void run() {
while(true) {
for(char c = ‘A‘; c <= ‘z‘; c++) {
try {
out.write(c);
sleep(rand.nextInt(500));
} catch(Exception e) {
throw new RuntimeException(e);
}
}
}
}
}
class Receiver extends Thread {
private PipedReader in;
public Receiver(Sender sender) throws IOException {
in = new PipedReader(sender.getPipedWriter());
}
public void run() {
try {
while(true) {
// Blocks until characters are there:
System.out.println("Read: " + (char)in.read());
}
} catch(IOException e) {
throw new RuntimeException(e);
}
}
}
public class PipedIO {
public static void main(String[] args) throws Exception {
Sender sender = new Sender();
Receiver receiver = new Receiver(sender);
sender.start();
receiver.start();
new Timeout(4000, "Terminated");
}
} ///:~

Sender和Receiver表示兩個(gè)需要相互通信才能完成任務(wù)的線程。Sender的PipedWriter是一個(gè)獨(dú)立的對(duì)象；但是Receiver的PipedReader必須與構(gòu)造函數(shù)里的PipedWriter相關(guān)聯(lián)。Sender把數(shù)據(jù)寫入Writer，然后休眠一段時(shí)間。Receiver則不sleep( )或wait( )，它read( )不到數(shù)據(jù)的時(shí)候，自動(dòng)就阻塞了。這樣一來(lái)你就能不用wait( )循環(huán)就獲得producer-consumer的效果了。

注意，main( )先創(chuàng)建了sender和receiver，再調(diào)用了start( )。如果你沒(méi)創(chuàng)建完對(duì)象就啟動(dòng)線程，那么管道在不同的平臺(tái)上的行為就有可能會(huì)不一致。

更復(fù)雜的協(xié)同

這里只講了最基本的協(xié)同方式(即通常籍由wait( )，notify( )/notifyAll( )來(lái)實(shí)現(xiàn)的producer-consumer模式)。它已經(jīng)能解決絕大多數(shù)的線程協(xié)同問(wèn)題了，但是在高級(jí)的教科書里還有很多更復(fù)雜協(xié)同方式(特別是本章末尾所推薦的，由Doug Lea編著的Concurrent Programming in Java, 2nd Edition)

死鎖

由于線程能被阻塞，更由于synchronized方法能阻止其它線程訪問(wèn)本對(duì)象，因此有可能會(huì)出現(xiàn)如下這種情況：線程一在等線程二(釋放某個(gè)對(duì)象)，線程二又在等線程三，這樣依次排下去直到有個(gè)線程在等線程一。這樣就形成了一個(gè)環(huán)，每個(gè)線程都在等對(duì)方釋放資源，而它們誰(shuí)都不能運(yùn)行。這就是所謂的死鎖(deadlock)。

如果程序一運(yùn)行就死鎖，那倒也簡(jiǎn)單了。你可以馬上著手解決這個(gè)問(wèn)題。但真正的麻煩在于，程序看上去能正常運(yùn)行，但是卻潛伏著會(huì)引起死鎖的隱患?；蛟S你認(rèn)為這里根本就不可能會(huì)有死鎖，而bug也就這樣潛伏下來(lái)了。直到有一天，讓某個(gè)用戶給撞上了(而且這種bug還很可能是不可重復(fù)的)。所以對(duì)并發(fā)編程來(lái)說(shuō)，防止死鎖是設(shè)計(jì)階段的一個(gè)重要任務(wù)。

下面我們來(lái)看看由Dijkstra發(fā)現(xiàn)的經(jīng)典的死鎖場(chǎng)景：哲學(xué)家吃飯問(wèn)題。原版的故事里有五個(gè)哲學(xué)家(不過(guò)我們的例程里允許有任意數(shù)量)。這些哲學(xué)家們只做兩件事，思考和吃飯。他們思考的時(shí)候，不需要任何共享資源，但是吃飯的時(shí)候，就必須坐到餐桌旁。餐桌上的餐具是有限的。原版的故事里，餐具是叉子，吃飯的時(shí)候要用兩把叉子把面條從碗里撈出來(lái)。但是很明顯，把叉子換成筷子會(huì)更合理，所以：一個(gè)哲學(xué)家需要兩根筷子才能吃飯。

現(xiàn)在引入問(wèn)題的關(guān)鍵：這些哲學(xué)家很窮，只買得起五根筷子。他們坐成一圈，兩個(gè)人的中間放一根筷子。哲學(xué)家吃飯的時(shí)候必須同時(shí)得到左手邊和右手邊的筷子。如果他身邊的任何一位正在使用筷子，那他只有等著。

這個(gè)問(wèn)題之所以有趣就在于，它演示了這么一個(gè)程序，它看上去似乎能正常運(yùn)行，但是卻容易引起死鎖。你可以自己試試，用命令行參數(shù)調(diào)節(jié)哲學(xué)家的數(shù)量和思考的時(shí)間。如果有很多哲學(xué)家，而且/或者他們思考的時(shí)間很長(zhǎng)，或許你永遠(yuǎn)也碰不到死鎖，但是死鎖的可能性總還是在的。默認(rèn)的命令行參數(shù)會(huì)讓它很快地死鎖：

//: c13:DiningPhilosophers.java
// Demonstrates how deadlock can be hidden in a program.
// {Args: 5 0 deadlock 4}
import java.util.*;
class Chopstick {
private static int counter = 0;
private int number = counter++;
public String toString() {
return "Chopstick " + number;
}
}
class Philosopher extends Thread {
private static Random rand = new Random();
private static int counter = 0;
private int number = counter++;
private Chopstick leftChopstick;
private Chopstick rightChopstick;
static int ponder = 0; // Package access
public Philosopher(Chopstick left, Chopstick right) {
leftChopstick = left;
rightChopstick = right;
start();
}
public void think() {
System.out.println(this + " thinking");
if(ponder > 0)
try {
sleep(rand.nextInt(ponder));
} catch(InterruptedException e) {
throw new RuntimeException(e);
}
}
public void eat() {
synchronized(leftChopstick) {
System.out.println(this + " has "
+ this.leftChopstick + " Waiting for "
+ this.rightChopstick);
synchronized(rightChopstick) {
System.out.println(this + " eating");
}
}
}
public String toString() {
return "Philosopher " + number;
}
public void run() {
while(true) {
think();
eat();
}
}
}
public class DiningPhilosophers {
public static void main(String[] args) {
if(args.length < 3) {
System.err.println("usage:\n" +
"java DiningPhilosophers numberOfPhilosophers " +
"ponderFactor deadlock timeout\n" +
"A nonzero ponderFactor will generate a random " +
"sleep time during think().\n" +
"If deadlock is not the string " +
"‘deadlock‘, the program will not deadlock.\n" +
"A nonzero timeout will stop the program after " +
"that number of seconds.");
System.exit(1);
}
Philosopher[] philosopher =
new Philosopher[Integer.parseInt(args[0])];
Philosopher.ponder = Integer.parseInt(args[1]);
Chopstick
left = new Chopstick(),
right = new Chopstick(),
first = left;
int i = 0;
while(i < philosopher.length - 1) {
philosopher[i++] =
new Philosopher(left, right);
left = right;
right = new Chopstick();
}
if(args[2].equals("deadlock"))
philosopher[i] = new Philosopher(left, first);
else // Swapping values prevents deadlock:
philosopher[i] = new Philosopher(first, left);
// Optionally break out of program:
if(args.length >= 4) {
int delay = Integer.parseInt(args[3]);
if(delay != 0)
new Timeout(delay * 1000, "Timed out");
}
}
} ///:~

Chopstick和Philosopher都包含一個(gè)能自動(dòng)遞增的static counter。每個(gè)Philosopher有兩個(gè)reference，一個(gè)表示左邊那根Chopstick，另一個(gè)表示右邊那根；Philosopher吃飯之前必須拿到這兩根筷子。

static的ponder表示哲學(xué)家要花多長(zhǎng)時(shí)間思考。如果這個(gè)值非零，則think( )會(huì)用它來(lái)生成一個(gè)隨機(jī)的休眠時(shí)間。我們用這種方法證明，如果線程(也就是哲學(xué)家)花在其它事情上(思考)的時(shí)間多了，那么它們使用共享資源(筷子)的機(jī)會(huì)就少了，因而程序就不太容易死鎖了，但事實(shí)并非如此。

請(qǐng)看eat( )。Philosopher先synchronized 左邊那根筷子。如果得不到，他就等，這時(shí)他處于阻塞狀態(tài)。得到左邊那根筷子之后，他又用相同的方法去申請(qǐng)右邊的筷子。吃完之后，他也是先放左邊的，再放右邊的。

Philosopher在run( )里不停的思考和吃飯。

main( )至少需要三個(gè)參數(shù)，如果沒(méi)給夠，它就打印使用的信息。第三個(gè)參數(shù)可以是"deadlock"字符串，這時(shí)程序會(huì)啟動(dòng)會(huì)引發(fā)死鎖的版本。任何其它字符串都會(huì)運(yùn)行其非死鎖的版本。最后一個(gè)參數(shù)(可選的)是一個(gè)時(shí)限，程序運(yùn)行這段時(shí)間之后(以秒為單位，不論是否死鎖都會(huì))都會(huì)退出。為了能自動(dòng)地進(jìn)行測(cè)試，這個(gè)時(shí)限參數(shù)是必須的。

除了創(chuàng)建Philosopher數(shù)組，設(shè)置ponder的值，main( )還創(chuàng)建了兩個(gè)Chopstick對(duì)象，第一個(gè)對(duì)象放在first變量里。除了最后一個(gè)對(duì)象，其他Philosopher的初始化過(guò)程都一樣：先新建一個(gè)Philosopher，把left和right Chopstick傳給他，然后把right的Chopstick傳到left，再為right創(chuàng)建一個(gè)新的Chopstick。下一個(gè)Philosopher再繼續(xù)用這兩根筷子。

在會(huì)死鎖版本里，我們把先前存在first里的那根筷子放在最后那個(gè)Philosopher的左邊。由于最后一位Philosopher正好坐在第一位的旁邊，這樣他們就必須共享first筷子了。這種安排有可能會(huì)造成這種情況:每個(gè)哲學(xué)家手里都攥著一根筷子，他在等他旁邊那位放下手里的筷子，這樣他才能吃飯，但程序已經(jīng)死鎖了。

試著用各種命令行參數(shù)來(lái)運(yùn)行程序，看看程序會(huì)怎樣運(yùn)行，特別是要注意在哪些情況下程序不會(huì)死鎖。

在告訴你如何修補(bǔ)這個(gè)問(wèn)題之前，先了解一下只有在下述四個(gè)條件同時(shí)滿足的情況下，死鎖才會(huì)發(fā)生:

1. 互斥:也許線程會(huì)用到很多資源，但其中至少要有一項(xiàng)是不能共享的。這里，一根筷子同一時(shí)刻只能供一個(gè)哲學(xué)家使用。
2. 至少要有一個(gè)進(jìn)程會(huì)在占用一項(xiàng)資源的同時(shí)還在等另一項(xiàng)正被其它進(jìn)程所占用的資源。也就是說(shuō)，要想讓死鎖發(fā)生，哲學(xué)家必須攥著一根筷子等另一根。
3. (調(diào)度系統(tǒng)或其他進(jìn)程)不能從進(jìn)程里搶資源。所有進(jìn)程都必須正常的釋放資源。我們的哲學(xué)家都彬彬有禮，不會(huì)從他的鄰座手里搶筷子。
4. 必需要有等待的環(huán)。一個(gè)進(jìn)程在一個(gè)已經(jīng)被另一進(jìn)程搶占了的資源，而那個(gè)進(jìn)程又在等另一個(gè)被第三個(gè)進(jìn)程搶占了的資源，以此類推，直到有個(gè)進(jìn)程正在等被第一個(gè)進(jìn)程搶占了的資源，這樣就形成了癱瘓性的阻塞了。這里，由于每個(gè)哲學(xué)家都是先左后右的拿筷子，所以有可能會(huì)造成等待的環(huán)。在例程中，我們修改了最后一位哲學(xué)家的構(gòu)造函數(shù)，讓他先右后左地拿筷子，從而破解了死鎖。

由于死鎖要同時(shí)滿足這四個(gè)條件，所用只要去掉其中一個(gè)就能防止死鎖。對(duì)于這個(gè)程序，預(yù)防死鎖最簡(jiǎn)單的辦法是破掉第四條。之所以會(huì)死鎖，是因?yàn)槊總€(gè)哲學(xué)家都以固定的次序拿筷子:先左后右。因此就有可能會(huì)發(fā)生每個(gè)人的左手都捏著一根筷子，然后等右邊那根，這樣就形成了等待環(huán)了。如果初始化的時(shí)候讓最后那位哲學(xué)家先右后左的拿筷子，那他就不會(huì)再礙著他左邊那位去拿右邊的筷子了，因此等待環(huán)就被破了。這只是這個(gè)問(wèn)題的解決方法之一，你還可以用破解其它條件的辦法來(lái)解決這個(gè)問(wèn)題(要想學(xué)得更細(xì)，可以去看更高級(jí)的教課書)。

Java語(yǔ)言沒(méi)有提供任何能預(yù)防死鎖的機(jī)制，所以只能靠你來(lái)設(shè)計(jì)了。對(duì)于那些排錯(cuò)的人來(lái)說(shuō)，這可不是什么好消息。