Java并發程式設計的藝術(六)——線程間的通信

版權聲明：本文為部落客原創文章，未經部落客允許不得轉載。 https://blog.csdn.net/qq_34173549/article/details/79612409

多條線程之間有時需要資料互動，下面介紹五種線程間資料互動的方式，他們的使用場景各有不同。

1. volatile、synchronized關鍵字

PS：關于volatile的詳細介紹請移步至：

Java并發程式設計的藝術(三)——volatile

1.1 如何實作通信？

這兩種方式都采用了同步機制實作多條線程間的資料通信。與其說是“通信”，倒不如說是“共享變量”來的恰當。當一個共享變量被volatile修飾 或 被同步塊包裹後，他們的讀寫操作都會直接操作共享記憶體，進而各個線程都能看到共享變量最新的值，也就是實作了記憶體的可見性。

1.2 特點

• 這種方式本質上是“共享資料”，而非“傳遞資料”；隻是從結果來看，資料好像是從寫線程傳遞到了讀線程；
• 這種通信方式無法指定特定的接收線程。當資料被修改後究竟哪條線程最先通路到，這由作業系統随機決定。
• 總的來說，這種方式并不是真正意義上的“通信”，而是“共享”。

1.3 使用場景

這種方式能“傳遞”變量。當需要傳遞一些公用的變量時就可以使用這種方式。如：傳遞boolean flag，用于表示狀态、傳遞一個存儲所有任務的隊列等。

1.4 例子

用這種方式實作線程的開關控制。

// 用于控制線程目前的執行狀态
private volatile boolean running = false;

// 開啟一條線程
Thread thread = new Thread(new Runnable(){
    void run(){
        // 開關
        while(!running){
            Thread.sleep(1000);
        }
        // 執行線程任務
        doSometing();
    }
}).start();

// 開始執行
public void start(){
    running = true;
}           

• 1
• 2
• 3
• 4
• 5
• 6
• 7
• 8
• 9
• 10
• 11
• 12
• 13
• 14
• 15
• 16
• 17
• 18
• 19

2. 等待/通知機制

2.1 如何實作？

等待/通知機制的實作由Java完成，我們隻需調用Object類的幾個方法即可。

• wait()：将目前線程的狀态改為“等待态”，加入等待隊列，釋放鎖；直到目前線程發生中斷或調用了notify方法，這條線程才會被從等待隊列轉移到同步隊列，此時可以開始競争鎖。
• wait(long)：和wait()功能一樣，隻不過多了個逾時動作。一旦逾時，就會繼續執行wait之後的代碼，它不會抛逾時異常！
• notify()：将等待隊列中的一條線程轉移到同步隊列中去。
• notifyAll()：将等待隊列中的所有線程都轉移到同步隊列中去。

2.2 注意點

• 以上方法都必須放在同步塊中；
• 并且以上方法都隻能由所處同步塊的鎖對象調用；
• 鎖對象A.notify()/notifyAll()隻能喚醒由鎖對象A wait的線程；
• 調用notify/notifyAll函數後僅僅是将線程從等待隊列轉移到阻塞隊列，隻有當該線程競争到鎖後，才能從wait方法中傳回，繼續執行接下來的代碼；

2.3 QA

• 為什麼wait必須放在同步塊中調用？ 

  因為等待/通知機制需要和共享狀态變量配合使用，一般是先檢查狀态，若狀态為true則執行wait，即包含“先檢查後執行”，是以需要把這一過程加鎖，確定其原子執行。 

  舉個例子：

// 共享的狀态變量
boolean flag = false;

// 線程1
Thread t1 = new Thread(new Runnable(){
    public void run(){
        while(!flag){
            wait();
        }
    }
}).start();

// 線程2
Thread t2 = new Thread(new Runnable(){
    public void run(){
        flag = true;
        notifyAll();
    }
}).start();           

上述例子thread1未加同步。當thread1執行到while那行後，判斷其狀态為true，此時若發生上下文切換，線程2開始執行，并一口氣執行完了；此時flag已經是true，然而thread1繼續執行，遇到wait後便進入等待态；但此時已經沒有線程能喚醒它了，是以就一直等待下去。

• 為什麼notify需要加鎖？且必須和wait使用同一把鎖？ 

  首先，加鎖是為了保證共享變量的記憶體可見性，讓它發生修改後能直接寫入共享記憶體，好讓wait所處的線程立即看見。 

  其次，和wait使用同一把鎖是為了確定wait、notify之間的互斥，即：同一時刻，隻能有其中一條線程執行。

• 為什麼必須使用同步塊的鎖對象調用wait函數？ 

  首先，由于wait會釋放鎖，是以通過鎖對象調用wait就是告訴wait釋放哪個鎖。 

  其次，告訴線程，你是在哪個鎖對象上等待的，隻有當該鎖對象調用notify時你才能被喚醒。

• 為什麼必須使用同步塊的鎖對象調用notify函數？ 

  告訴notify，隻喚醒在該鎖對象上等待的線程。

2.4 代碼實作

等待/通知機制用于實作生産者和消費者模式。

• 生産者

synchronized(鎖A){
    flag = true;// 或者：list.add(xx);
    鎖A.notify();
}           

• 消費者

synchronized(鎖A){
    // 不滿足條件
    while(!flag){ // 或者：list.isEmpty()
        鎖A.wait();
    }

    // doSometing……
}           

2.5 逾時等待模式

在之前的生産者-消費者模式中，如果生産者沒有發出通知，那麼消費者将永遠等待下去。為了避免這種情況，我們可以給消費者增加逾時等待功能。該功能依托于wait(long)方法，隻需在wait前的檢查條件中增加逾時辨別位，實作如下：

public void get(long mills){
    synchronized( list ){
        // 不加逾時功能
        if ( mills <= 0 ) {
            while( list.isEmpty() ){
                list.wait();
            }
        }

        // 添加逾時功能
        else {
            boolean isTimeout = false;
            while(list.isEmpty() && isTimeout){
                list.wait(mills);
                isTimeout = true;
            }

            // doSometing……
        }
    }
}           

• 20
• 21

3. 管道流

3.1 作用

管道流用于在兩個線程之間進行位元組流或字元流的傳遞。

3.2 特點

• 管道流的實作依靠PipedOutputStream、PipedInputStream、PipedWriter、PipedReader。分别對應位元組流和字元流。
• 他們與IO流的差別是：IO流是在硬碟、記憶體、Socket之間流動，而管道流僅在記憶體中的兩條線程間流動。

3.3 實作

步驟如下： 

1. 在一條線程中分别建立輸入流和輸出流； 

2. 将輸入流和輸出流連接配接起來； 

3. 将輸入流和輸出流分别傳遞給兩條線程； 

4. 調用read和write方法就可以實作線程間通信。

// 建立輸入流與輸出流對象
PipedWriter out = new PipedWriter();
PipedReader in = new PipedReader();

// 連接配接輸入輸出流
out.connect(in);

// 建立寫線程
class WriteThread extends Thread{
    private PipedWriter out;

    public WriteThread(PipedWriter out){
        this.out = out;
    }

    public void run(){
        out.write("hello concurrent world!");
    }
}

// 建立讀線程
class ReaderThread extends Thread{
    private PipedReader in;

    public ReaderThread(PipedReader in){
        this.in = in;
    }

    public void run(){
        in.read();
    }
}

//            

• 22
• 23
• 24
• 25
• 26
• 27
• 28
• 29
• 30
• 31
• 32
• 33
• 34

4. join

4.1 作用

• join能将并發執行的多條線程串行執行；
• join函數屬于Thread類，通過一個thread對象調用。當線上程B中執行threadA.join()時，線程B将會被阻塞(底層調用wait方法)，等到threadA線程運作結束後才會傳回join方法。
• 被等待的那條線程可能會執行很長時間，是以join函數會抛出InterruptedException。當調用threadA.interrupt()後，join函數就會抛出該異常。

4.2 實作

public static void main(String[] args){

    // 開啟一條線程
    Thread t = new Thread(new Runnable(){
        public void run(){
            // doSometing
        }
    }).start();

    // 調用join，等待t線程執行完畢
    try{
        t.join();
    }catch(InterruptedException e){
        // 中斷處理……
    }

}