JAVA並發 線程間的消息傳遞

概要

線程間的通信是用volatile和synchronized兩個關鍵字實現同步完成的線程間的通信；但是在JAVA中的線程之間的通信其實就是共享內存，當一個變量被volatile修飾或者被同步塊包括時，那么線程的操作會實時更新到共享內存，然后各個線程都會知道最新變量的值，也就是內存的可見性；看起來實現了線程間的通信，但是實際是共享內存。關於Volatile的詳解到JAVA並發Volatile。

特點

1. 這種方式的本質是共享數據，而不是傳遞數據；只是從結果上看。數據好像從寫線程傳遞到了讀線程。
2. 這通通信機制無法指定特定的線程接受消息，具體要哪一個接受消息，由操作系統決定。
3. 總的來說不是真正意義上的通信，是共享數據。

例子

private volatile static boolean runing=false;
    public static void main(String[] args) {
        Thread t1=new Thread(new Runnable() {
            
            @Override
            public void run() {
                while(!runing) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
                
            }
        });
        
        t1.start(); 
    }
public void start() {
        runing=true;
    }

等待通知機制

實現方式

1. wait()：將當前線程狀態改為等待狀態，加入等待隊列，釋放占用鎖；直到當線程發生中斷或者調用notify方法，這條線程才會從等待隊列轉移到同步隊列開始競爭鎖。
2. wait(long)：和wait一樣，只不過多了一個超時動作。一旦超時，就會繼續執行wait后面的代碼，它不會拋出異常。
3. notify()：將等待隊列中的一條線程轉移到同步隊列中去。
4. notifyAll()：將等待隊列中的所有的線程都轉移到同步隊列中去。

注意

1. 以上方法必須放在一個同步塊中。
2. 並且以上方法只能夠方法所處的同步塊的鎖對象調用。
3. 鎖對象A.notify只能夠喚醒A.wait()。
4. 調用notify/notifyAll函數僅僅是將線程從等待隊列轉移到阻塞隊列，只有當線程競爭到資源鎖時，才能夠從wait中返回，繼續執行接下來的代碼。

例子

Thread t2=new Thread(new Runnable() {
            
            @Override
            public void run() {
                while(!runing) {
                    try {
                        wait();
                        System.out.println("wait after");
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                }
            }
        });
        t2.start();
        
        Thread t3=new  Thread(new Runnable() {
            
            @Override
            public void run() {
                runing=true;
                notifyAll();
            }
        });

運行結果

Exception in thread "Thread-0" java.lang.IllegalMonitorStateException
    at java.lang.Object.wait(Native Method)
    at java.lang.Object.wait(Object.java:502)
    at javaTest.ThreadApi$1.run(ThreadApi.java:35)
    at java.lang.Thread.run(Thread.java:748)

具體問題請看注意點

重要問題

①為什么wait必須放在同步塊中調試。

因為同步/等待機制需要和共享變量配合使用，一般是先檢查狀態，再執行。因此會對這個過程加一把鎖，確保其原子性運行。

②為什么notify要加鎖？還必須和wait同一把鎖

首先加鎖是為了內存的可見性，使其發生的修改能夠及時顯示給其他線程；和wait一起使用是保證wait和notify之間的互斥，即：同一時刻，只能有其中一條線程運行。

③為什么必須使用同步塊的鎖對象調用wait函數和notify函數。

調用wait函數：由於wait要釋放鎖，所有通過鎖對象告訴是哪個要釋放鎖，然后告訴線程你是在哪個鎖上等待的，只有當前鎖對象調用notify時才會被喚醒。

管道流

管道流用於兩個線程之間的字符流動或者字節流動；

管道流主要：PipedOutputSTream，PipedInputStream，PipedWriter，PipedReader。

他們和io的區別是：Io流是在硬盤，內存，socket之間流動，管道流是在線程之間流動。

實現

    static PipedWriter out=new PipedWriter();
    static PipedReader in =new PipedReader();
    class WriteThread extends Thread{
        private PipedWriter out;
        
        public WriteThread(PipedWriter out) {
            this.out=out;
        }
        public void run() {
            try {
                out.write("hello world");
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    
    class ReadThread extends Thread{
        private PipedReader in;
        public ReadThread(PipedReader in) {
            this.in=in;
        }
        public void run() {
            try {
                in.read();
            } catch (IOException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args) throws IOException {
        out.connect(in);
    }

Join

join能夠使並發的多線程串行運行

join屬於Thread類，通過一個Thread對象調用。當在線程B中執行ThreadA.join時，線程B會被阻塞，等到線程A運行完成。

被等待的那條線程可能會執行很長時間，因此join函數會拋出InterruptedException。當調用threadA.interrupt()后，join函數就會拋出該異常。

實現

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){
        // 中斷處理……
    }

}