計算機程序的思惟邏輯 (67) - 線程的基本協做機制 (上)

本系列文章經補充和完善，已修訂整理成書《Java編程的邏輯》（馬俊昌著），由機械工業出版社華章分社出版，於2018年1月上市熱銷，讀者好評如潮！各大網店和書店有售，歡迎購買：京東自營連接

上節介紹了多線程之間競爭訪問同一個資源的問題及解決方案synchronized，咱們提到，多線程之間除了競爭，還常常須要相互協做，本節就來介紹Java中多線程協做的基本機制wait/notify。

都有哪些場景須要協做？wait/notify是什麼？如何使用？實現原理是什麼？協做的核心是什麼？如何實現各類典型的協做場景？因爲內容較多，咱們分爲上下兩節來介紹。

咱們先來看看都有哪些協做的場景。

協做的場景

多線程之間須要協做的場景有不少，好比說：

• 生產者/消費者協做模式：這是一種常見的協做模式，生產者線程和消費者線程經過共享隊列進行協做，生產者將數據或任務放到隊列上，而消費者從隊列上取數據或任務，若是隊列長度有限，在隊列滿的時候，生產者須要等待，而在隊列爲空的時候，消費者須要等待。
• 同時開始：相似運動員比賽，在聽到比賽開始槍響後同時開始，在一些程序，尤爲是模擬仿真程序中，要求多個線程能同時開始。
• 等待結束：主從協做模式也是一種常見的協做模式，主線程將任務分解爲若干個子任務，爲每一個子任務建立一個線程，主線程在繼續執行其餘任務以前須要等待每一個子任務執行完畢。
• 異步結果：在主從協做模式中，主線程手工建立子線程的寫法每每比較麻煩，一種常見的模式是將子線程的管理封裝爲異步調用，異步調用立刻返回，但返回的不是最終的結果，而是一個通常稱爲Promise或Future的對象，經過它能夠在隨後得到最終的結果。
• 集合點：相似於學校或公司組團旅遊，在旅遊過程當中有若干集合點，好比出發集合點，每一個人從不一樣地方來到集合點，全部人到齊後進行下一項活動，在一些程序，好比並行迭代計算中，每一個線程負責一部分計算，而後在集合點等待其餘線程完成，全部線程到齊後，交換數據和計算結果，再進行下一次迭代。

咱們會探討如何實現這些協做場景，在此以前，咱們先來了解協做的基本方法wait/notify。

wait/notify

咱們知道，Java的根父類是Object，Java在Object類而非Thread類中，定義了一些線程協做的基本方法，使得每一個對象均可以調用這些方法，這些方法有兩類，一類是wait，另外一類是notify。

主要有兩個wait方法：

public final void wait() throws InterruptedException public final native void wait(long timeout) throws InterruptedException;
複製代碼

一個帶時間參數，單位是毫秒，表示最多等待這麼長時間，參數爲0表示無限期等待。一個不帶時間參數，表示無限期等待，實際就是調用wait(0)。在等待期間均可以被中斷，若是被中斷，會拋出InterruptedException，關於中斷及中斷處理，咱們在下節介紹，本節暫時忽略該異常。

wait實際上作了什麼呢？它在等待什麼？上節咱們說過，每一個對象都有一把鎖和等待隊列，一個線程在進入synchronized代碼塊時，會嘗試獲取鎖，獲取不到的話會把當前線程加入等待隊列中，其實，除了用於鎖的等待隊列，每一個對象還有另外一個等待隊列，表示條件隊列，該隊列用於線程間的協做。調用wait就會把當前線程放到條件隊列上並阻塞，表示當前線程執行不下去了，它須要等待一個條件，這個條件它本身改變不了，須要其餘線程改變。當其餘線程改變了條件後，應該調用Object的notify方法：

public final native void notify();
public final native void notifyAll();
複製代碼

notify作的事情就是從條件隊列中選一個線程，將其從隊列中移除並喚醒，notifyAll和notify的區別是，它會移除條件隊列中全部的線程並所有喚醒。

咱們來看個簡單的例子，一個線程啓動後，在執行一項操做前，它須要等待主線程給它指令，收到指令後才執行，代碼以下：

public class WaitThread extends Thread {
    private volatile boolean fire = false;

    @Override
    public void run() {
        try {
            synchronized (this) {
                while (!fire) {
                    wait();
                }
            }
            System.out.println("fired");
        } catch (InterruptedException e) {
        }
    }

    public synchronized void fire() {
        this.fire = true;
        notify();
    }

    public static void main(String[] args) throws InterruptedException {
        WaitThread waitThread = new WaitThread();
        waitThread.start();
        Thread.sleep(1000);
        System.out.println("fire");
        waitThread.fire();
    }
}
複製代碼

示例代碼中有兩個線程，一個是主線程，一個是WaitThread，協做的條件變量是fire，WaitThread等待該變量變爲true，在不爲true的時候調用wait，主線程設置該變量並調用notify。

兩個線程都要訪問協做的變量fire，容易出現競態條件，因此相關代碼都須要被synchronized保護。實際上，wait/notify方法只能在synchronized代碼塊內被調用，若是調用wait/notify方法時，當前線程沒有持有對象鎖，會拋出異常java.lang.IllegalMonitorStateException。

你可能會有疑問，若是wait必須被synchronzied保護，那一個線程在wait時，另外一個線程怎麼可能調用一樣被synchronzied保護的notify方法呢？它不須要等待鎖嗎？咱們須要進一步理解wait的內部過程，雖然是在synchronzied方法內，但調用wait時，線程會釋放對象鎖，wait的具體過程是：

1. 把當前線程放入條件等待隊列，釋放對象鎖，阻塞等待，線程狀態變爲WAITING或TIMED_WAITING
2. 等待時間到或被其餘線程調用notify/notifyAll從條件隊列中移除，這時，要從新競爭對象鎖

• 若是可以得到鎖，線程狀態變爲RUNNABLE，並從wait調用中返回
  複製代碼

• 不然，該線程加入對象鎖等待隊列，線程狀態變爲BLOCKED，只有在得到鎖後纔會從wait調用中返回

線程從wait調用中返回後，不表明其等待的條件就必定成立了，它須要從新檢查其等待的條件，通常的調用模式是：

synchronized (obj) {
    while (條件不成立)
        obj.wait();
    ... // 執行條件知足後的操做
}
複製代碼

好比，上例中的代碼是：

synchronized (this) {
    while (!fire) {
        wait();
    }
}
複製代碼

調用notify會把在條件隊列中等待的線程喚醒並從隊列中移除，但它不會釋放對象鎖，也就是說，只有在包含notify的synchronzied代碼塊執行完後，等待的線程纔會從wait調用中返回。

簡單總結一下，wait/notify方法看上去很簡單，但每每難以理解wait等的究竟是什麼，而notify通知的又是什麼，咱們須要知道，它們與一個共享的條件變量有關，這個條件變量是程序本身維護的，當條件不成立時，線程調用wait進入條件等待隊列，另外一個線程修改了條件變量後調用notify，調用wait的線程喚醒後須要從新檢查條件變量。從多線程的角度看，它們圍繞共享變量進行協做，從調用wait的線程角度看，它阻塞等待一個條件的成立。咱們在設計多線程協做時，須要想清楚協做的共享變量和條件是什麼，這是協做的核心。接下來，咱們經過一些場景來進一步理解wait/notify的應用，本節只介紹生產者/消費者模式，下節介紹更多模式。

生產者/消費者模式

在生產者/消費者模式中，協做的共享變量是隊列，生產者往隊列上放數據，若是滿了就wait，而消費者從隊列上取數據，若是隊列爲空也wait。咱們將隊列做爲單獨的類進行設計，代碼以下：

static class MyBlockingQueue<E> {
    private Queue<E> queue = null;
    private int limit;

    public MyBlockingQueue(int limit) {
        this.limit = limit;
        queue = new ArrayDeque<>(limit);
    }

    public synchronized void put(E e) throws InterruptedException {
        while (queue.size() == limit) {
            wait();
        }
        queue.add(e);
        notifyAll();
    }

    public synchronized E take() throws InterruptedException {
        while (queue.isEmpty()) {
            wait();
        }
        E e = queue.poll();
        notifyAll();
        return e;
    }
}
複製代碼

MyBlockingQueue是一個長度有限的隊列，長度經過構造方法的參數進行傳遞，有兩個方法put和take。put是給生產者使用的，往隊列上放數據，滿了就wait，放完以後調用notifyAll，通知可能的消費者。take是給消費者使用的，從隊列中取數據，若是爲空就wait，取完以後調用notifyAll，通知可能的生產者。

咱們看到，put和take都調用了wait，但它們的目的是不一樣的，或者說，它們等待的條件是不同的，put等待的是隊列不爲滿，而take等待的是隊列不爲空，但它們都會加入相同的條件等待隊列。因爲條件不一樣但又使用相同的等待隊列，因此要調用notifyAll而不能調用notify，由於notify只能喚醒一個線程，若是喚醒的是同類線程就起不到協調的做用。

只能有一個條件等待隊列，這是Java wait/notify機制的侷限性，這使得對於等待條件的分析變得複雜，後續章節咱們會介紹顯式的鎖和條件，它能夠解決該問題。

一個簡單的生產者代碼以下所示：

static class Producer extends Thread {
    MyBlockingQueue<String> queue;

    public Producer(MyBlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        int num = 0;
        try {
            while (true) {
                String task = String.valueOf(num);
                queue.put(task);
                System.out.println("produce task " + task);
                num++;
                Thread.sleep((int) (Math.random() * 100));
            }
        } catch (InterruptedException e) {
        }
    }
}
複製代碼

Producer向共享隊列中插入模擬的任務數據。一個簡單的示例消費者代碼以下所示：

static class Consumer extends Thread {
    MyBlockingQueue<String> queue;

    public Consumer(MyBlockingQueue<String> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                String task = queue.take();
                System.out.println("handle task " + task);
                Thread.sleep((int)(Math.random()*100));
            }
        } catch (InterruptedException e) {
        }
    }
}
複製代碼

主程序的示例代碼以下所示：

public static void main(String[] args) {
    MyBlockingQueue<String> queue = new MyBlockingQueue<>(10);
    new Producer(queue).start();
    new Consumer(queue).start();
}
複製代碼

運行該程序，會看到生產者和消費者線程的輸出交替出現。

咱們實現的MyBlockingQueue主要用於演示，Java提供了專門的阻塞隊列實現，包括：

• 接口BlockingQueue和BlockingDeque
• 基於數組的實現類ArrayBlockingQueue
• 基於鏈表的實現類LinkedBlockingQueue和LinkedBlockingDeque
• 基於堆的實現類PriorityBlockingQueue

咱們會在後續章節介紹這些類，在實際系統中，應該考慮使用這些類。

小結

本節介紹了Java中線程間協做的基本機制wait/notify，協做關鍵要想清楚協做的共享變量和條件是什麼，爲進一步理解，本節針對生產者/消費者模式演示了wait/notify的用法。

下一節，咱們來繼續探討其餘協做模式。

(與其餘章節同樣，本節全部代碼位於 github.com/swiftma/pro…)

---

未完待續，查看最新文章，敬請關注微信公衆號「老馬說編程」(掃描下方二維碼)，深刻淺出，老馬和你一塊兒探索Java編程及計算機技術的本質。用心原創，保留全部版權。