Java并發(fā)編程多線程間的同步控制和通信詳解

更新時(shí)間：2022年11月29日 16:52:45 作者：kevinyan

這篇文章主要為大家介紹了Java并發(fā)編程多線程間的同步控制和通信詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

正文

使用多線程并發(fā)處理，目的是為了讓程序更充分地利用CPU ，好能加快程序的處理速度和用戶體驗(yàn)。如果每個(gè)線程各自處理的部分互不相干，那真是極好的，我們?cè)诔绦蛑骶€程要做的同步控制最多也就是等待幾個(gè)工作線程的執(zhí)行完畢，如果不 Care 結(jié)果的話，連同步等待都能省去，主線程撒開手讓這些線程干就行了。

不過(guò)，現(xiàn)實(shí)還是很殘酷的，大部分情況下，多個(gè)線程是會(huì)有競(jìng)爭(zhēng)操作同一個(gè)對(duì)象的情況的，這個(gè)時(shí)候就會(huì)導(dǎo)致并發(fā)常見的一個(gè)問(wèn)題--數(shù)據(jù)競(jìng)爭(zhēng)（Data Racing）。

這篇文章我們就來(lái)討論一下這個(gè)并發(fā)導(dǎo)致的問(wèn)題，以及多線程間進(jìn)行同步控制和通信的知識(shí)，本文大綱如下：

并發(fā)導(dǎo)致的Data Racing問(wèn)題

怎么理解這個(gè)問(wèn)題呢，拿一個(gè)多個(gè)線程同時(shí)對(duì)累加器對(duì)象進(jìn)行累加的例子來(lái)解釋吧。

package com.learnthread;
public class DataRacingTest {
    public static void main(String[] args) throws InterruptedException {
        final DataRacingTest test = new DataRacingTest();
        // 創(chuàng)建兩個(gè)線程，執(zhí)行 add100000() 操作
        // 創(chuàng)建Thread 實(shí)例時(shí)的 Runnable 接口實(shí)現(xiàn)，這里直接使用了 Lambda
        Thread th1 = new Thread(()-> test.add100000());
        Thread th2 = new Thread(()-> test.add100000());
        // 啟動(dòng)兩個(gè)線程
        th1.start();
        th2.start();
        // 等待兩個(gè)線程執(zhí)行結(jié)束
        th1.join();
        th2.join();
        System.out.println(test.count);
    }
    private long count = 0;
    // 想復(fù)現(xiàn) Data Racing，去掉這里的 synchronized
    private  void add100000() {
        int idx = 0;
        while(idx++ < 100000) {
            count += 1;
        }
    }
}

上面這個(gè)例程，如果我們不啟動(dòng) th2 線程，只用 th1 一個(gè)線程進(jìn)行累加操作的話結(jié)果是 100000。按照這個(gè)思維，如果我們啟動(dòng)兩個(gè)線程那么最后累加的結(jié)果就應(yīng)該是 200000。但實(shí)際上并不是，我們運(yùn)行一下上面的例程，得到的結(jié)果是：

168404
Process finished with exit code 0

當(dāng)然這個(gè)在每個(gè)人的機(jī)器上的結(jié)果是不一樣的，而且也是有可能恰好等于 200000，需要多運(yùn)行幾次，或者是多開幾個(gè)線程執(zhí)行累加，出現(xiàn) Data Racing 的幾率才高。

程序出現(xiàn) Data Racing 的現(xiàn)象，就意味著最終拿到的數(shù)據(jù)是不正確的。那么為了避免這個(gè)問(wèn)題就需要通過(guò)加鎖來(lái)解決了，讓同一時(shí)間只有持有鎖的線程才能對(duì)數(shù)據(jù)對(duì)象進(jìn)行操作。當(dāng)然針對(duì)簡(jiǎn)單的運(yùn)算、賦值等操作我們也能直接使用原子操作實(shí)現(xiàn)無(wú)鎖解決 Data Racing，我們?yōu)榱耸纠銐蚝?jiǎn)單易懂才舉了一個(gè)累加的例子，實(shí)際上如果是一段業(yè)務(wù)邏輯操作的話，就只能使用加鎖來(lái)保證不會(huì)出現(xiàn) Data Racing了。

加鎖，只是線程并發(fā)同步控制的一種，還有釋放鎖、喚醒線程、同步等待線程執(zhí)行完畢等操作，下面我們會(huì)逐一進(jìn)行學(xué)習(xí)。

同步控制--synchronized

開頭的那個(gè)例程，如果想避免 Data Racing，那么就需要加上同步鎖，讓同一個(gè)時(shí)間只能有一個(gè)線程操作數(shù)據(jù)對(duì)象。針對(duì)我們的例程，我們只需要在 add100000 方法的聲明中加上 synchronized 即可。

    // 想復(fù)現(xiàn) Data Racing，去掉這里的 synchronized
    private synchronized void add100000() {
        int idx = 0;
        while(idx++ < 100000) {
            count += 1;
        }
    }

是不是很簡(jiǎn)單，當(dāng)然 synchronized 的用法遠(yuǎn)不止這個(gè)，它可以加在實(shí)例方法、靜態(tài)方法、代碼塊上，如果使用的不對(duì)，就不能正確地給需要同步鎖保護(hù)的對(duì)象加上鎖。

synchronized 是 Java 中的關(guān)鍵字，是利用鎖的機(jī)制來(lái)實(shí)現(xiàn)互斥同步的。 synchronized 可以保證在同一個(gè)時(shí)刻，只有一個(gè)線程可以執(zhí)行某個(gè)方法或者某個(gè)代碼塊。如果不需要 Lock 、讀寫鎖ReadWriteLock 所提供的高級(jí)同步特性，應(yīng)該優(yōu)先考慮使用synchronized 這種方式加鎖，主要原因如下：

Java 自 1.6 版本以后，對(duì) synchronized 做了大量的優(yōu)化，其性能已經(jīng)與 JUC 包中的 Lock 、ReadWriteLock 基本上持平。從趨勢(shì)來(lái)看，Java 未來(lái)仍將繼續(xù)優(yōu)化 synchronized ，而不是 ReentrantLock 。
ReentrantLock 是 Oracle JDK 的 API，在其他版本的 JDK 中不一定支持；而 synchronized 是 JVM 的內(nèi)置特性，所有 JDK 版本都提供支持。

synchronized 可以應(yīng)用在實(shí)例方法、靜態(tài)方法和代碼塊上：

用 synchronized 關(guān)鍵字修飾實(shí)例方法，即為同步實(shí)例方法，鎖是當(dāng)前的實(shí)例對(duì)象。
用 synchronized 關(guān)鍵字修飾類的靜態(tài)方法，即為同步靜態(tài)方法，鎖是當(dāng)前的類的 Class 對(duì)象。
如果把 synchronized 應(yīng)用在代碼塊上，鎖是 synchronized 括號(hào)里配置的對(duì)象，synchronized(this) {..} 鎖就是代碼塊所在實(shí)例的對(duì)象，synchronized(類名.class) {...} ，鎖就是類的 Class 對(duì)象。

同步實(shí)例方法和代碼塊

上面我們已經(jīng)看過(guò)怎么給實(shí)例方法加 synchronized 讓它變成同步方法了。下面我們看一下，synchronized 給實(shí)例方法加鎖時(shí)，不能保證資源被同步鎖保護(hù)的例子。

class Account {
  private int balance;
  // 轉(zhuǎn)賬
  synchronized void transfer(Account target, int amt){
    if (this.balance > amt) {
      this.balance -= amt;
      target.balance += amt;
    }
  }
}

在這段代碼中，臨界區(qū)內(nèi)有兩個(gè)資源，分別是轉(zhuǎn)出賬戶的余額 this.balance 和轉(zhuǎn)入賬戶的余額 target.balance，并且用的是一把實(shí)例對(duì)象的鎖。問(wèn)題就出在 this 這把鎖上，this 這把鎖可以保護(hù)自己的余額 this.balance，卻保護(hù)不了別人的余額 target.balance，就像你不能用自家的鎖來(lái)保護(hù)別人家的資產(chǎn)一個(gè)道理。

應(yīng)該保證使用的鎖能保護(hù)所有應(yīng)受保護(hù)資源。我們可以使用Account.class 作為加鎖的對(duì)象。Account.class 是所有 Account 類的對(duì)象共享的，而且是 Java 虛擬機(jī)在加載 Account 類的時(shí)候創(chuàng)建的，保證了它的全局唯一性。

class Account {
  private int balance;
  // 轉(zhuǎn)賬
  void transfer(Account target, int amt){
    synchronized(Account.class) {
      if (this.balance > amt) {
        this.balance -= amt;
        target.balance += amt;
      }
    }
  }
}

用 synchronized 給 Account.class 加鎖，這樣就保證出賬、入賬兩個(gè) Account 對(duì)象在同步代碼塊里都能收到保護(hù)。

當(dāng)然我們也可以使用這筆轉(zhuǎn)賬的交易對(duì)象作為加鎖的對(duì)象，保證只有這比交易的兩個(gè) Account 對(duì)象受保護(hù)，這樣就不會(huì)影響到其他轉(zhuǎn)賬交易里的出賬、入賬 Account 對(duì)象了。

class Account {
  private Trans trans；
  private int balance;
  private Account();
  // 創(chuàng)建 Account 時(shí)傳入同一個(gè) 交易對(duì)象作為 lock 對(duì)象
  public Account(Trans trans) {
    this.trans = trans;
  }
  // 轉(zhuǎn)賬
  void transfer(Account target, int amt){
    // 此處檢查所有對(duì)象共享的鎖
    synchronized(trans) {
      if (this.balance > amt) {
        this.balance -= amt;
        target.balance += amt;
      }
    }
  }
}

通過(guò)解決上面這個(gè)問(wèn)題我們順道就把 synchronized 修飾同步代碼塊的知識(shí)點(diǎn)學(xué)了，現(xiàn)在我們來(lái)看 synchronized 的最后一個(gè)用法--修飾同步靜態(tài)方法。

同步靜態(tài)方法

靜態(tài)方法的同步是指，用 synchronized 修飾的靜態(tài)方法，與使用所在類的 Class 對(duì)象實(shí)現(xiàn)的同步代碼塊，效果類似。因?yàn)樵?JVM 中一個(gè)類只能對(duì)應(yīng)一個(gè)類的 Class 對(duì)象，所以同時(shí)只允許一個(gè)線程執(zhí)行同一個(gè)類中的靜態(tài)同步方法。

對(duì)于同一個(gè)類中的多個(gè)靜態(tài)同步方法，持有鎖的線程可以執(zhí)行每個(gè)類中的靜態(tài)同步方法而無(wú)需等待。不管類中的哪個(gè)靜態(tài)同步方法被調(diào)用，一個(gè)類只能由一個(gè)線程同時(shí)執(zhí)行。

package com.learnthread;
public class SynchronizedStatic implements Runnable {
    private static final int MAX = 100000;
    private static int count = 0;
    public static void main(String[] args) throws InterruptedException {
        SynchronizedStatic instance = new SynchronizedStatic();
        Thread t1 = new Thread(instance);
        Thread t2 = new Thread(instance);
        t1.start();
        t2.start();
        // 等待工作線程執(zhí)行結(jié)束
        t1.join();
        t2.join();
        System.out.println(count);
    }
    @Override
    public void run() {
        for (int i = 0; i < MAX; i++) {
            increase();
        }
    }
    /**
     * synchronized 修飾靜態(tài)方法
     */
    public synchronized static void increase() {
        count++;
    }
}

線程掛起和喚醒

上面我們看了使用 synchronized 給對(duì)象加同步鎖，讓同一時(shí)間只有一個(gè)線程能操作臨界區(qū)的控制。接下來(lái)，我們看一下線程的掛起和喚醒，這兩個(gè)操作使用被線程成功加鎖的對(duì)象的 wait 和 notify 方法來(lái)完成，喚醒除了notify 外還有 notifyAll方法用來(lái)喚醒所有線程。下面我們先看一下這幾個(gè)方法的解釋。

wait - wait 會(huì)自動(dòng)釋放當(dāng)前線程占有的對(duì)象鎖，并請(qǐng)求操作系統(tǒng)掛起當(dāng)前線程，讓線程從 Running 狀態(tài)轉(zhuǎn)入 Waiting 狀態(tài)，等待被 notify / notifyAll 來(lái)喚醒。如果沒有釋放鎖，那么其它線程就無(wú)法進(jìn)入對(duì)象的同步方法或者同步代碼塊中，那么就無(wú)法執(zhí)行 notify 或者 notifyAll 來(lái)喚醒掛起的線程，會(huì)造成死鎖。
notify - 喚醒一個(gè)正在 Waiting 狀態(tài)的線程，并讓它拿到對(duì)象鎖，具體喚醒哪一個(gè)線程由 JVM 控制。
notifyAll - 喚醒所有正在 Waiting 狀態(tài)的線程，接下來(lái)它們需要競(jìng)爭(zhēng)對(duì)象鎖。

這里有兩點(diǎn)需要各位注意的地方，第一個(gè)是 wait、notify、notifyAll 都是 Object 類中的方法，而不是 Thread 類的。

因?yàn)?Object 是始祖類，是不是意味著所有類的對(duì)象都能調(diào)用這幾個(gè)方法呢？是，也不是... 因?yàn)?wait、notify、notifyAll 只能用在 synchronized 方法或者 synchronized 代碼塊中使用，否則會(huì)在運(yùn)行時(shí)拋出 IllegalMonitorStateException。換句話說(shuō)，只有被 synchronized 加上鎖的對(duì)象，才能調(diào)用這三個(gè)方法。

為什么 wait、notify、notifyAll 不定義在 Thread 類中？為什么 wait、notify、notifyAll 要配合 synchronized 使用？理解為什么這么設(shè)計(jì)，需要了解幾個(gè)基本知識(shí)點(diǎn)：

每一個(gè) Java 對(duì)象都有一個(gè)與之對(duì)應(yīng)的監(jiān)視器（monitor）
每一個(gè)監(jiān)視器里面都有一個(gè) 對(duì)象鎖、一個(gè) 等待隊(duì)列、一個(gè) 同步隊(duì)列

了解了以上概念，我們回過(guò)頭來(lái)理解前面兩個(gè)問(wèn)題。

為什么這幾個(gè)方法不定義在 Thread 中？

由于每個(gè)對(duì)象都擁有對(duì)象鎖，讓當(dāng)前線程等待某個(gè)對(duì)象鎖，自然應(yīng)該基于這個(gè)對(duì)象（Object）來(lái)操作，而非使用當(dāng)前線程（Thread）來(lái)操作。因?yàn)楫?dāng)前線程可能會(huì)等待多個(gè)線程釋放鎖，如果基于線程（Thread）來(lái)操作，就非常復(fù)雜了。

為什么 wait、notify、notifyAll 要配合 synchronized 使用？

如果調(diào)用某個(gè)對(duì)象的 wait 方法，當(dāng)前線程必須擁有這個(gè)對(duì)象的對(duì)象鎖，因此調(diào)用 wait 方法必須在 synchronized 方法和 synchronized 代碼塊中。

下面看一個(gè) wait、notify、notifyAll 的一個(gè)經(jīng)典使用案例，實(shí)現(xiàn)一個(gè)生產(chǎn)者、消費(fèi)者模式:

package com.learnthread;
import java.util.PriorityQueue;
public class ThreadWaitNotifyDemo {
    private static final int QUEUE_SIZE = 10;
    private static final PriorityQueue&lt;Integer&gt; queue = new PriorityQueue&lt;&gt;(QUEUE_SIZE);
    public static void main(String[] args) {
        new Producer("生產(chǎn)者A").start();
        new Producer("生產(chǎn)者B").start();
        new Consumer("消費(fèi)者A").start();
        new Consumer("消費(fèi)者B").start();
    }
    static class Consumer extends Thread {
        Consumer(String name) {
            super(name);
        }
        @Override
        public void run() {
            while (true) {
                synchronized (queue) {
                    while (queue.size() == 0) {
                        try {
                            System.out.println("隊(duì)列空，等待數(shù)據(jù)");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notifyAll();
                        }
                    }
                    queue.poll(); // 每次移走隊(duì)首元素
                    queue.notifyAll();
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + " 從隊(duì)列取走一個(gè)元素，隊(duì)列當(dāng)前有：" + queue.size() + "個(gè)元素");
                }
            }
        }
    }
    static class Producer extends Thread {
        Producer(String name) {
            super(name);
        }
        @Override
        public void run() {
            while (true) {
                synchronized (queue) {
                    while (queue.size() == QUEUE_SIZE) {
                        try {
                            System.out.println("隊(duì)列滿，等待有空余空間");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notifyAll();
                        }
                    }
                    queue.offer(1); // 每次插入一個(gè)元素
                    queue.notifyAll();
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println(Thread.currentThread().getName() + " 向隊(duì)列取中插入一個(gè)元素，隊(duì)列當(dāng)前有：" + queue.size() + "個(gè)元素");
                }
            }
        }
    }
}

上面的例程有兩個(gè)生產(chǎn)者和兩個(gè)消費(fèi)者。生產(chǎn)者向隊(duì)列中放數(shù)據(jù)，每次向隊(duì)列中放入數(shù)據(jù)后使用 notifyAll 喚醒消費(fèi)者線程，當(dāng)隊(duì)列滿后生產(chǎn)者會(huì) wait 讓出線程，等待消費(fèi)者取走數(shù)據(jù)后再被喚醒（消費(fèi)者取數(shù)據(jù)后也會(huì)調(diào)用 notifyAll ）。同理消費(fèi)者在隊(duì)列空后也會(huì)使用 wait 讓出線程，等待生產(chǎn)者向隊(duì)列中放入數(shù)據(jù)后被喚醒。

線程等待--join

與 wait 和 notify 方法一樣，join 是另一種線程間同步機(jī)制。當(dāng)我們調(diào)用線程對(duì)象 join 方法時(shí)，調(diào)用線程會(huì)進(jìn)入等待狀態(tài)，它會(huì)一直處于等待狀態(tài)，直到被引用的線程執(zhí)行結(jié)束。在上面的幾個(gè)例子中，我們已經(jīng)使用過(guò)了 join 方法

   ...
	public static void main(String[] args) throws InterruptedException {
        SynchronizedStatic instance = new SynchronizedStatic();
        Thread t1 = new Thread(instance);
        Thread t2 = new Thread(instance);
        t1.start();
        t2.start();
        // 等待工作線程執(zhí)行結(jié)束
        t1.join();
        t2.join();
        System.out.println(count);
    }

這個(gè)例子里，主線程調(diào)用 t1 和 t2 的 join 方法后，就會(huì)一直等待，直到他們兩個(gè)執(zhí)行結(jié)束。如果 t1 或者 t2 線程處理時(shí)間過(guò)長(zhǎng)，調(diào)用它們 join 方法的主線程將一直等待，程序阻塞住。為了避免這些情況，可以使用能指定超時(shí)時(shí)間的重載版本的 join 方法。

    t2.join(1000); // 最長(zhǎng)等待1s

如果引用的線程被中斷，join方法也會(huì)返回。在這種情況下，還會(huì)觸發(fā) InterruptedException。所以上面的main方法為了演示方便，直接選擇拋出了 InterruptedException。

總結(jié)

同步控制的一大思路就是加鎖，除了本問(wèn)學(xué)習(xí)到的 sychronized 同步控制，Java 里還有 JUC 的可重入鎖、讀寫鎖這種加鎖方式，這個(gè)我們后續(xù)介紹 JUC 的時(shí)候會(huì)給大家講解。

另外一種思路是不加鎖，讓線程和線程之間盡量不要使用共享數(shù)據(jù)，ThreadLocal 就是這種思路，下篇我們介紹 Java 的線程本地存儲(chǔ) -- ThreadLocal，更多關(guān)于Java 并發(fā)多線程同步控制通信的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: