快捷導(dǎo)航

深入理解Java中線程間的通信

更新時間：2022年11月25日 08:17:10 作者：初念初戀

一般來講，線程內(nèi)部有自己私有的線程上下文，互不干擾。但是當(dāng)我們需要多個線程之間相互協(xié)作的時候，就需要我們掌握J(rèn)ava線程的通信方式。本文將介紹Java線程之間的幾種通信原理，需要的可以參考一下

合理的使用Java多線程可以更好地利用服務(wù)器資源。一般來講，線程內(nèi)部有自己私有的線程上下文，互不干擾。但是當(dāng)我們需要多個線程之間相互協(xié)作的時候，就需要我們掌握J(rèn)ava線程的通信方式。本文將介紹Java線程之間的幾種通信原理。

鎖與同步

在Java中，鎖的概念都是基于對象的，所以我們又經(jīng)常稱它為對象鎖。一個鎖同一時間只能被一個線程持有。也就是說，一個鎖如果被一個線程所持有，那其他線程如果需要得到這個鎖，就得等這個線程釋放該鎖。

線程之間，有一個同步的概念。在多線程中，可能有多個線程試圖訪問一個有限的資源，必須預(yù)防這種情況的發(fā)生。

所以引入了同步機制：在線程使用一個資源時為其加鎖，這樣其他的線程便不能訪問那個資源了，直到解鎖后才可以訪問。線程同步是線程之間按照一定的順序執(zhí)行。

我們先來看看一個無鎖的程序：

public class NoLock {

    static class ThreadA implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                System.out.println("Thread A " + i);
            }
        }
    }

    static class ThreadB implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 100; i++) {
                System.out.println("Thread B " + i);
            }
        }
    }

    public static void main(String[] args) {
        new Thread(new ThreadA()).start();
        new Thread(new ThreadB()).start();
    }
}

執(zhí)行這個程序，你會在控制臺看到，線程A和線程B各自獨立工作，輸出自己的打印值。每一次運行結(jié)果都會不一樣。如下是我的電腦上某一次運行的結(jié)果。

...
Thread B 74
Thread B 75
Thread B 76
Thread B 77
Thread B 78
Thread B 79
Thread B 80
Thread A 3
Thread A 4
Thread A 5
Thread A 6
Thread A 7
Thread A 8
Thread A 9
...

現(xiàn)在有一個需求，想等A先執(zhí)行完之后，再由B去執(zhí)行，怎么辦呢？最簡單的方式就是使用一個“對象鎖”。

public class ObjLock {

    private static Object lock = new Object();

    static class ThreadA implements Runnable {
        @Override
        public void run() {
            synchronized (lock) {
                for (int i = 0; i < 100; i++) {
                    System.out.println("Thread A " + i);
                }
            }
        }
    }

    static class ThreadB implements Runnable {
        @Override
        public void run() {
            synchronized (lock) {
                for (int i = 0; i < 100; i++) {
                    System.out.println("Thread B " + i);
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new Thread(new ThreadA()).start();
        Thread.sleep(10);
        new Thread(new ThreadB()).start();
    }
}

這里聲明了一個名字為lock的對象鎖。我們在ThreadA和ThreadB內(nèi)需要同步的代碼塊里，都是用synchronized關(guān)鍵字加上了同一個對象鎖lock。

上文我們說到了，根據(jù)線程和鎖的關(guān)系，同一時間只有一個線程持有一個鎖，那么線程B就會等線程A執(zhí)行完成后釋放lock，線程B才能獲得鎖lock。

這里在主線程里使用sleep方法睡眠了10毫秒，是為了防止線程B先得到鎖。因為如果同時start，線程A和線程B都是出于就緒狀態(tài)，操作系統(tǒng)可能會先讓B運行。這樣就會先輸出B的內(nèi)容，然后B執(zhí)行完成之后自動釋放鎖，線程A再執(zhí)行。

等待/通知機制

上面一種基于“鎖”的方式，線程需要不斷地去嘗試獲得鎖，如果失敗了，再繼續(xù)嘗試。這可能會耗費服務(wù)器資源。而等待/通知機制是另一種方式。

Java多線程的等待/通知機制是基于Object類的wait()方法和notify(), notifyAll()方法來實現(xiàn)的。

notify()方法會隨機叫醒一個正在等待的線程，而notifyAll()會叫醒所有正在等待的線程。

前面我們講到，一個鎖同一時刻只能被一個線程持有。而假如線程A現(xiàn)在持有了一個鎖lock并開始執(zhí)行，它可以使用lock.wait()讓自己進入等待狀態(tài)。這個時候，lock這個鎖是被釋放了的。

這時，線程B獲得了lock這個鎖并開始執(zhí)行，它可以在某一時刻，使用lock.notify()，通知之前持有lock鎖并進入等待狀態(tài)的線程A，說“線程A你不用等了，可以往下執(zhí)行了”。

需要注意的是，這個時候線程B并沒有釋放鎖lock，除非線程B這個時候使用lock.wait()釋放鎖，或者線程B執(zhí)行結(jié)束自行釋放鎖，線程A才能得到lock鎖。

用代碼來實現(xiàn)一下：

public class WaitNotify {

    private static Object lock = new Object();

    static class ThreadA implements Runnable {
        @Override
        public void run() {
            synchronized (lock) {
                for (int i = 0; i < 5; i++) {
                    try {
                        System.out.println("ThreadA: " + i);
                        lock.notify();
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                lock.notify();
            }
        }
    }

    static class ThreadB implements Runnable {
        @Override
        public void run() {
            synchronized (lock) {
                for (int i = 0; i < 5; i++) {
                    try {
                        System.out.println("ThreadB: " + i);
                        lock.notify();
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                lock.notify();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new Thread(new ThreadA()).start();
        Thread.sleep(1000);
        new Thread(new ThreadB()).start();
    }
}

// 輸出：
ThreadA: 0
ThreadB: 0
ThreadA: 1
ThreadB: 1
ThreadA: 2
ThreadB: 2
ThreadA: 3
ThreadB: 3
ThreadA: 4
ThreadB: 4

線程A和線程B首先打印出自己需要的東西，然后使用notify()方法叫醒另一個正在等待的線程，然后自己使用wait()方法陷入等待并釋放lock鎖。

需要注意的是等待/通知機制使用的是使用同一個對象鎖，如果你兩個線程使用的是不同的對象鎖，那它們之間是不能用等待/通知機制通信的。

信號量--Volatile

信號量(Semaphore) ：有時被稱為信號燈，是在多線程環(huán)境下使用的一種設(shè)施，是可以用來保證兩個或多個關(guān)鍵代碼段不被并發(fā)調(diào)用。在進入一個關(guān)鍵代碼段之前，線程必須獲取一個信號量；一旦該關(guān)鍵代碼段完成了，那么該線程必須釋放信號量。其它想進入該關(guān)鍵代碼段的線程必須等待直到第一個線程釋放信號量。

本文不是要介紹這個類，而是介紹一種基于volatile關(guān)鍵字的自己實現(xiàn)的信號量通信。

volatile關(guān)鍵字能夠保證內(nèi)存的可見性，如果用volatile關(guān)鍵字聲明了一個變量，在一個線程里面改變了這個變量的值，那其它線程是立馬可見更改后的值的。

比如我現(xiàn)在有一個需求，我想讓線程A輸出0，然后線程B輸出1，再然后線程A輸出2…以此類推。我應(yīng)該怎樣實現(xiàn)呢？

public class Count {
    private static volatile int count = 0;

    static class ThreadA implements Runnable {
        @Override
        public void run() {
            while (count < 5) {
                if (count % 2 == 0) {
                    System.out.println("threadA: " + count);
                    synchronized (this) {
                        count++;
                    }
                }
            }
        }
    }

    static class ThreadB implements Runnable {
        @Override
        public void run() {
            while (count < 5) {
                if (count % 2 == 1) {
                    System.out.println("threadB: " + signal);
                    synchronized (this) {
                        count++;
                    }
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        new Thread(new ThreadA()).start();
        Thread.sleep(1000);
        new Thread(new ThreadB()).start();
    }
}

// 輸出：
threadA: 0
threadB: 1
threadA: 2
threadB: 3
threadA: 4

我們可以看到，使用了一個volatile變量count來實現(xiàn)了“信號量”的模型。這里需要注意的是，volatile變量需要進行原子操作，而count++并不是一個原子操作，根據(jù)需要使用synchronized給它“上鎖”，或者是使用AtomicInteger等原子類。

信號量的應(yīng)用場景

假如在一個停車場中，車位是我們的公共資源，線程就如同車輛，而看門的管理員就是起的“信號量”的作用。因為在這種場景下，多個線程需要相互合作，我們用簡單的“鎖”和“等待通知機制”就不那么方便了。這個時候就可以用到信號量。

管道輸入/輸出流

管道是基于“管道流”的通信方式。JDK提供了PipedWriter、 PipedReader、 PipedOutputStream、 PipedInputStream。

其中，前面兩個是基于字符的，后面兩個是基于字節(jié)流的。

以下示例代碼使用的是基于字符的：

public class Pipe {
    static class ReaderThread implements Runnable {
        private PipedReader reader;

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

        @Override
        public void run() {
            System.out.println("this is reader");
            int receive = 0;
            try {
                while ((receive = reader.read()) != -1) {
                    System.out.print((char)receive);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    static class WriterThread implements Runnable {

        private PipedWriter writer;

        public WriterThread(PipedWriter writer) {
            this.writer = writer;
        }

        @Override
        public void run() {
            System.out.println("this is writer");
            int receive = 0;
            try {
                writer.write("test");
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    writer.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        PipedWriter writer = new PipedWriter();
        PipedReader reader = new PipedReader();
        writer.connect(reader);

        new Thread(new ReaderThread(reader)).start();
        Thread.sleep(1000);
        new Thread(new WriterThread(writer)).start();
    }
}

// 輸出：
this is reader
this is writer
test

我們通過線程的構(gòu)造函數(shù)，傳入了PipedWrite和PipedReader對象?？梢院唵畏治鲆幌逻@個示例代碼的執(zhí)行流程：

線程ReaderThread開始執(zhí)行，
線程ReaderThread使用管道reader.read()進入”阻塞“，
線程WriterThread開始執(zhí)行，
線程WriterThread用writer.write("test")往管道寫入字符串，
線程WriterThread使用writer.close()結(jié)束管道寫入，并執(zhí)行完畢，
線程ReaderThread接受到管道輸出的字符串并打印，
線程ReaderThread執(zhí)行完畢。

管道通信的應(yīng)用場景

這個很好理解。使用管道多半與I/O流相關(guān)。當(dāng)我們一個線程需要先另一個線程發(fā)送一個信息（比如字符串）或者文件等等時，就需要使用管道通信了。

Thread.join()方法

join()方法是Thread類的一個實例方法。它的作用是讓當(dāng)前線程陷入“等待”狀態(tài)，等join的這個線程執(zhí)行完成后，再繼續(xù)執(zhí)行當(dāng)前線程。

有時候，主線程創(chuàng)建并啟動了子線程，如果子線程中需要進行大量的耗時運算，主線程往往將早于子線程結(jié)束之前結(jié)束。

如果主線程想等待子線程執(zhí)行完畢后，獲得子線程中的處理完的某個數(shù)據(jù)，就要用到j(luò)oin方法了。

示例代碼：

public class Join {
    static class ThreadA implements Runnable {

        @Override
        public void run() {
            try {
                System.out.println("我是子線程，我先睡一秒");
                Thread.sleep(1000);
                System.out.println("我是子線程，我睡完了一秒");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Thread thread = new Thread(new ThreadA());
        thread.start();
        thread.join();
        System.out.println("如果不加join方法，我會先被打出來，加了就不一樣了");
    }
}

ThreadLocal類

ThreadLocal是一個本地線程副本變量工具類。內(nèi)部是一個弱引用的Map來維護。

嚴(yán)格來說，ThreadLocal類并不屬于多線程間的通信，而是讓每個線程有自己”獨立“的變量，線程之間互不影響。它為每個線程都創(chuàng)建一個副本，每個線程可以訪問自己內(nèi)部的副本變量。

ThreadLocal類最常用的就是set方法和get方法。示例代碼：

public class Profiler {

    private static final ThreadLocal<Long> TIME_THREADLOCAL = new ThreadLocal<Long>() {
        protected Long initialValue() {
            return System.currentTimeMillis();
        }
    };

    public static void begin() {
        TIME_THREADLOCAL.set(System.currentTimeMillis());
    }

    public static long end() {
        return System.currentTimeMillis() - TIME_THREADLOCAL.get();
    }

    public static void main(String[] args) throws InterruptedException {
        Profiler.begin();
        TimeUnit.SECONDS.sleep(1);
        System.out.println("耗時：" + Profiler.end() + "mills");
    }
}

// 輸出：
耗時：1001mills

Profiler 可以被復(fù)用在方法的耗時統(tǒng)計的功能上，在方法的入口前執(zhí)行begin()方法，在方法調(diào)用后執(zhí)行end()方法，好處是兩個方法的調(diào)用不用在一個方法和類中，比如在AOP（面向切面編程）中，可以在方法調(diào)用前的切入點執(zhí)行begin()方法，而在方法調(diào)用切入點執(zhí)行end()方法，這樣依舊可以獲得方法的執(zhí)行耗時。

ThreadLocal的應(yīng)用場景

最常見的ThreadLocal使用場景為用來解決數(shù)據(jù)庫連接、Session管理等。數(shù)據(jù)庫連接和Session管理涉及多個復(fù)雜對象的初始化和關(guān)閉。如果在每個線程中聲明一些私有變量來進行操作，那這個線程就變得不那么“輕量”了，需要頻繁的創(chuàng)建和關(guān)閉連接。