快捷導(dǎo)航

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

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

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

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

鎖與同步

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

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

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

我們先來(lái)看看一個(gè)無(wú)鎖的程序：

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í)行這個(gè)程序，你會(huì)在控制臺(tái)看到，線程A和線程B各自獨(dú)立工作，輸出自己的打印值。每一次運(yùn)行結(jié)果都會(huì)不一樣。如下是我的電腦上某一次運(yùn)行的結(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)在有一個(gè)需求，想等A先執(zhí)行完之后，再由B去執(zhí)行，怎么辦呢？最簡(jiǎn)單的方式就是使用一個(gè)“對(duì)象鎖”。

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();
    }
}

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

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

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

等待/通知機(jī)制

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

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

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

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

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

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

用代碼來(lái)實(shí)現(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()方法叫醒另一個(gè)正在等待的線程，然后自己使用wait()方法陷入等待并釋放lock鎖。

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

信號(hào)量--Volatile

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

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

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

比如我現(xiàn)在有一個(gè)需求，我想讓線程A輸出0，然后線程B輸出1，再然后線程A輸出2…以此類推。我應(yīng)該怎樣實(shí)現(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

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

信號(hào)量的應(yīng)用場(chǎng)景

假如在一個(gè)停車場(chǎng)中，車位是我們的公共資源，線程就如同車輛，而看門的管理員就是起的“信號(hào)量”的作用。因?yàn)樵谶@種場(chǎng)景下，多個(gè)線程需要相互合作，我們用簡(jiǎn)單的“鎖”和“等待通知機(jī)制”就不那么方便了。這個(gè)時(shí)候就可以用到信號(hào)量。

管道輸入/輸出流

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

其中，前面兩個(gè)是基于字符的，后面兩個(gè)是基于字節(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對(duì)象?？梢院?jiǎn)單分析一下這個(gè)示例代碼的執(zhí)行流程：

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

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

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

Thread.join()方法

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

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

如果主線程想等待子線程執(zhí)行完畢后，獲得子線程中的處理完的某個(gè)數(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方法，我會(huì)先被打出來(lái)，加了就不一樣了");
    }
}

ThreadLocal類

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

嚴(yán)格來(lái)說(shuō)，ThreadLocal類并不屬于多線程間的通信，而是讓每個(gè)線程有自己”獨(dú)立“的變量，線程之間互不影響。它為每個(gè)線程都創(chuàng)建一個(gè)副本，每個(gè)線程可以訪問自己內(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("耗時(shí)：" + Profiler.end() + "mills");
    }
}

// 輸出：
耗時(shí)：1001mills

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

ThreadLocal的應(yīng)用場(chǎng)景

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