17-Java併發程式設計：執行緒間協作的兩種方式：wait、notify、notifyAll和Condition
Java併發程式設計：執行緒間協作的兩種方式：wait、notify、notifyAll和Condition
在前面我們將了很多關於同步的問題，然而在現實中，需要執行緒之間的協作。比如說最經典的生產者-消費者模型：當佇列滿時，生產者需要等待佇列有空間才能繼續往裡面放入商品，而在等待的期間內，生產者必須釋放對臨界資源（即佇列）的佔用權。因為生產者如果不釋放對臨界資源的佔用權，那麼消費者就無法消費佇列中的商品，就不會讓佇列有空間，那麼生產者就會一直無限等待下去。因此，一般情況下，當佇列滿時，會讓生產者交出對臨界資源的佔用權，並進入掛起狀態。然後等待消費者消費了商品，然後消費者通知生產者佇列有空間了。同樣地，當佇列空時，消費者也必須等待，等待生產者通知它佇列中有商品了。這種互相通訊的過程就是執行緒間的協作。
今天我們就來探討一下Java中執行緒協作的最常見的兩種方式：利用Object.wait()、Object.notify()和使用Condition
以下是本文目錄大綱：
一.wait()、notify()和notifyAll()
二.Condition
三.生產者-消費者模型的實現
一.wait()、notify()和notifyAll()
wait()、notify()和notifyAll()是Object類中的方法：
/** * Wakes up a single thread that is waiting on this object's * monitor. If any threads are waiting on this object, one of them * is chosen to be awakened. The choice is arbitrary and occurs at * the discretion of the implementation. A thread waits on an object's * monitor by calling one of the wait methods */ public final native void notify(); /** * Wakes up all threads that are waiting on this object's monitor. A * thread waits on an object's monitor by calling one of the * wait methods. */ public final native void notifyAll(); /** * Causes the current thread to wait until either another thread invokes the * {@link java.lang.Object#notify()} method or the * {@link java.lang.Object#notifyAll()} method for this object, or a * specified amount of time has elapsed. * <p> * The current thread must own this object's monitor. */ public final native void wait(long timeout) throws InterruptedException;
從這三個方法的文字描述可以知道以下幾點資訊：
1）wait()、notify()和notifyAll()方法是本地方法，並且為final方法，無法被重寫。
2）呼叫某個物件的wait()方法能讓當前執行緒阻塞，並且當前執行緒必須擁有此物件的monitor（即鎖）
3）呼叫某個物件的notify()方法能夠喚醒一個正在等待這個物件的monitor的執行緒，如果有多個執行緒都在等待這個物件的monitor，則只能喚醒其中一個執行緒；
4）呼叫notifyAll()方法能夠喚醒所有正在等待這個物件的monitor的執行緒；
有朋友可能會有疑問：為何這三個不是Thread類宣告中的方法，而是Object類中宣告的方法（當然由於Thread類繼承了Object類，所以Thread也可以呼叫者三個方法）？其實這個問題很簡單，由於每個物件都擁有monitor（即鎖），所以讓當前執行緒等待某個物件的鎖，當然應該通過這個物件來操作了。而不是用當前執行緒來操作，因為當前執行緒可能會等待多個執行緒的鎖，如果通過執行緒來操作，就非常複雜了。
上面已經提到，如果呼叫某個物件的wait()方法，當前執行緒必須擁有這個物件的monitor（即鎖），因此呼叫wait()方法必須在同步塊或者同步方法中進行（synchronized塊或者synchronized方法）。
呼叫某個物件的wait()方法，相當於讓當前執行緒交出此物件的monitor，然後進入等待狀態，等待後續再次獲得此物件的鎖（Thread類中的sleep方法使當前執行緒暫停執行一段時間，從而讓其他執行緒有機會繼續執行，但它並不釋放物件鎖）；
notify()方法能夠喚醒一個正在等待該物件的monitor的執行緒，當有多個執行緒都在等待該物件的monitor的話，則只能喚醒其中一個執行緒，具體喚醒哪個執行緒則不得而知。
同樣地，呼叫某個物件的notify()方法，當前執行緒也必須擁有這個物件的monitor，因此呼叫notify()方法必須在同步塊或者同步方法中進行（synchronized塊或者synchronized方法）。
nofityAll()方法能夠喚醒所有正在等待該物件的monitor的執行緒，這一點與notify()方法是不同的。
這裡要注意一點：notify()和notifyAll()方法只是喚醒等待該物件的monitor的執行緒，並不決定哪個執行緒能夠獲取到monitor。
舉個簡單的例子：假如有三個執行緒Thread1、Thread2和Thread3都在等待物件objectA的monitor，此時Thread4擁有物件objectA的monitor，當在Thread4中呼叫objectA.notify()方法之後，Thread1、Thread2和Thread3只有一個能被喚醒。注意，被喚醒不等於立刻就獲取了objectA的monitor。假若在Thread4中呼叫objectA.notifyAll()方法，則Thread1、Thread2和Thread3三個執行緒都會被喚醒，至於哪個執行緒接下來能夠獲取到objectA的monitor就具體依賴於作業系統的排程了。
上面尤其要注意一點，一個執行緒被喚醒不代表立即獲取了物件的monitor，只有等呼叫完notify()或者notifyAll()並退出synchronized塊，釋放物件鎖後，其餘執行緒才可獲得鎖執行。
下面看一個例子就明白了：
public class Test {
public static Object object = new Object();
public static void main(String[] args) {
Thread1 thread1 = new Thread1();
Thread2 thread2 = new Thread2();
thread1.start();
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
thread2.start();
}
static class Thread1 extends Thread{
@Override
public void run() {
synchronized (object) {
try {
object.wait();
} catch (InterruptedException e) {
}
System.out.println("執行緒"+Thread.currentThread().getName()+"獲取到了鎖");
}
}
}
static class Thread2 extends Thread{
@Override
public void run() {
synchronized (object) {
object.notify();
System.out.println("執行緒"+Thread.currentThread().getName()+"呼叫了object.notify()");
}
System.out.println("執行緒"+Thread.currentThread().getName()+"釋放了鎖");
}
}
}無論執行多少次，執行結果必定是：
執行緒Thread-1呼叫了object.notify()
執行緒Thread-1釋放了鎖
執行緒Thread-0獲取到了鎖二.Condition
Condition是在java 1.5中才出現的，它用來替代傳統的Object的wait()、notify()實現執行緒間的協作，相比使用Object的wait()、notify()，使用Condition1的await()、signal()這種方式實現執行緒間協作更加安全和高效。因此通常來說比較推薦使用Condition，在阻塞佇列那一篇博文中就講述到了，阻塞佇列實際上是使用了Condition來模擬執行緒間協作。
- Condition是個介面，基本的方法就是await()和signal()方法；
- Condition依賴於Lock介面，生成一個Condition的基本程式碼是lock.newCondition()
- 呼叫Condition的await()和signal()方法，都必須在lock保護之內，就是說必須在lock.lock()和lock.unlock之間才可以使用
Conditon中的await()對應Object的wait()；
Condition中的signal()對應Object的notify()；
Condition中的signalAll()對應Object的notifyAll()。
三.生產者-消費者模型的實現
1.使用Object的wait()和notify()實現：
public class Test {
private int queueSize = 10;
private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
public static void main(String[] args) {
Test test = new Test();
Producer producer = test.new Producer();
Consumer consumer = test.new Consumer();
producer.start();
consumer.start();
}
class Consumer extends Thread{
@Override
public void run() {
consume();
}
private void consume() {
while(true){
synchronized (queue) {
while(queue.size() == 0){
try {
System.out.println("佇列空，等待資料");
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
queue.notify();
}
}
queue.poll(); //每次移走隊首元素
queue.notify();
System.out.println("從佇列取走一個元素，佇列剩餘"+queue.size()+"個元素");
}
}
}
}
class Producer extends Thread{
@Override
public void run() {
produce();
}
private void produce() {
while(true){
synchronized (queue) {
while(queue.size() == queueSize){
try {
System.out.println("佇列滿，等待有空餘空間");
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
queue.notify();
}
}
queue.offer(1); //每次插入一個元素
queue.notify();
System.out.println("向佇列取中插入一個元素，佇列剩餘空間："+(queueSize-queue.size()));
}
}
}
}
}2.使用Condition實現
public class Test {
private int queueSize = 10;
private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
private Lock lock = new ReentrantLock();
private Condition notFull = lock.newCondition();
private Condition notEmpty = lock.newCondition();
public static void main(String[] args) {
Test test = new Test();
Producer producer = test.new Producer();
Consumer consumer = test.new Consumer();
producer.start();
consumer.start();
}
class Consumer extends Thread{
@Override
public void run() {
consume();
}
private void consume() {
while(true){
lock.lock();
try {
while(queue.size() == 0){
try {
System.out.println("佇列空，等待資料");
notEmpty.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.poll(); //每次移走隊首元素
notFull.signal();
System.out.println("從佇列取走一個元素，佇列剩餘"+queue.size()+"個元素");
} finally{
lock.unlock();
}
}
}
}
class Producer extends Thread{
@Override
public void run() {
produce();
}
private void produce() {
while(true){
lock.lock();
try {
while(queue.size() == queueSize){
try {
System.out.println("佇列滿，等待有空餘空間");
notFull.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
queue.offer(1); //每次插入一個元素
notEmpty.signal();
System.out.println("向佇列取中插入一個元素，佇列剩餘空間："+(queueSize-queue.size()));
} finally{
lock.unlock();
}
}
}
}
}參考資料：
《Java程式設計思想》