Java編程的邏輯 (67) - 線程的基本協作機制 (上)


本系列文章經補充和完善,已修訂整理成書《Java編程的邏輯》,由機械工業出版社華章分社出版,於2018年1月上市熱銷,讀者好評如潮!各大網店和書店有售,歡迎購買,京東自營鏈接http://item.jd.com/12299018.html


 

上節介紹了多線程之間競爭訪問同一個資源的問題及解決方案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的用法。

下一節,我們來繼續探討其他協作模式。

(與其他章節一樣,本節所有代碼位於 https://github.com/swiftma/program-logic)

----------------

未完待續,查看最新文章,敬請關注微信公眾號“老馬說編程”(掃描下方二維碼),從入門到高級,深入淺出,老馬和你一起探索Java編程及計算機技術的本質。用心原創,保留所有版權。


免責聲明!

本站轉載的文章為個人學習借鑒使用,本站對版權不負任何法律責任。如果侵犯了您的隱私權益,請聯系本站郵箱yoyou2525@163.com刪除。



 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM