Java多執行緒程式設計(6)--執行緒間通訊(下)
阿新 • • 發佈：2020-08-11
因為本文的內容大部分是以生產者/消費者模式來進行講解和舉例的，所以在開始學習本文介紹的幾種執行緒間的通訊方式之前，我們先來熟悉一下生產者/消費者模式。
在實際的軟體開發過程中，經常會碰到如下場景：某個模組負責產生資料（可能是訊息、檔案、任務等），這些資料由另一個模組來負責處理。產生資料的模組，就形象地被稱為生產者；而處理資料的模組，就被稱為消費者。
單單抽象出生產者和消費者，還稱不上是生產者/消費者模式。該模式還需要有一個緩衝區處於生產者和消費者之間來作為一箇中介。生產者把資料放入緩衝區，而消費者從緩衝區取出資料。因此，生產者/消費者模式大概的結構如下圖：
我們可以使用不同的執行緒來模擬生產者和消費者。對應的，生產資料的執行緒就被稱為是生產者執行緒，而消費資料的執行緒就被稱為是消費者。而緩衝區則可以選用那些執行緒安全的資料結構來模擬，因此，緩衝區在這裡就起到了執行緒間通訊工具的作用。本文將會通過生產者/消費者模式作為例子來介紹幾種執行緒間的通訊方式。
## 一.阻塞佇列
### 1.BlockingQueue介面
如何選擇合適的資料結構來作為緩衝區呢？首先我們來分析一下我們的需求。首先，消費者應該是要按照生產者生產的順序來消費資料的，那麼我們腦海中浮現的一定是具有先進先出特性的隊列了。其次，既然是在多執行緒之間進行傳遞，那麼這個類一定是執行緒安全的。因此，緩衝區應該使用執行緒安全的佇列。我們首先想到的應該是ConcurrentLinkedQueue，它是使用連結串列實現的執行緒安全的佇列。但是，這個類是非阻塞的，這意味著當生產者向緩衝區中放入資料時，緩衝區是否已滿時需要生產者自己去判斷的；同理，當消費者去消費緩衝區中的資料時，緩衝區是否為空也是需要自己去判斷的。由於這個類是非阻塞的，因此我們只能在執行緒中不斷的去輪詢緩衝區，這顯然不是多執行緒程式設計該有的實現方式。
那麼，有沒有可以阻塞執行緒的佇列呢？答案是肯定的。java.util.concurrent包中的BlockingQueue介面定義了阻塞佇列的行為。當阻塞佇列中沒有資料的時候，消費者端的執行緒會被掛起直到有資料被放入佇列；當阻塞佇列中填滿資料的時候，生產者端的執行緒會被掛起直到佇列中有空的位置。
下面來介紹BlockingQueue介面中定義的方法。BlockingQueue介面時Queue介面的子介面，因此它繼承了Queue介面中所有的方法，由於這些方法都比較簡單，因此不再贅述。這裡只介紹BlockingQueue介面中新增的方法。
#### (1)放入資料
- **void put(E e) throws InterruptedException**
將指定元素e放入佇列。如果佇列沒有空間，則當前執行緒會阻塞直到佇列有空間。
- **boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException**
將指定元素e放入佇列。如果佇列沒有空間，則當前執行緒會阻塞直到佇列有空間或等待超時。
#### (2)取出資料
- **E take() throws InterruptedException**
從佇列中取出元素。如果佇列中沒有元素，則當前執行緒會阻塞直到佇列中有元素。
- **E poll(long timeout, TimeUnit unit) throws InterruptedException**
從佇列中取出元素。如果佇列中沒有元素，則當前執行緒會阻塞直到佇列中有元素或等待超時。
- **int drainTo(Collection c)**
一次性從佇列中取出所有可用的資料物件放在指定的集合中。
- **int drainTo(Collection c, int maxElements)**
同上，maxElements可以限制最多獲取的元素個數。
### 2.BlockingQueue介面的實現類
java.util.concurrent包為BlockingQueue介面提供了7個實現類。
#### (1)ArrayBlockingQueue
ArrayBlockingQueue是基於陣列實現的有界的阻塞佇列，其內部維護了一個定長陣列來存放佇列中的資料物件。由於其是有界的，因此在構造ArrayBlockingQueue例項時，必須提供佇列的容量。這是一個非常常用的阻塞佇列，除了一個定長陣列外，ArrayBlockingQueue內部還維護了兩個整形變數，分別標識著佇列的頭部和尾部在陣列中的位置。
> 阻塞佇列按照其儲存空間的容量是否受限制來劃分，可以分為有界佇列和無界佇列。有界佇列的儲存容量限制是在構造例項的時候指定的，而無界佇列實際上也有儲存容量限制，其預設的最大儲存容量為Integer.MAX_VALUE（即231 exchanger = new Exchanger<>();
new Buyer(exchanger).start();
new Seller(exchanger).start();
}
private static class Buyer extends Thread {
private Exchanger exchanger;
Buyer(Exchanger exchanger) {
this.exchanger = exchanger;
}
@Override
public void run() {
try {
String money = "10元";
Thread.sleep(2000);
System.out.println("買家：拿出" + money);
String good = exchanger.exchange(money);
System.out.println("買家：得到" + good);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
private static class Seller extends Thread {
private Exchanger exchanger;
Seller(Exchanger exchanger) {
this.exchanger = exchanger;
}
@Override
public void run() {
try {
String good = "10斤大白菜";
System.out.println("賣家：拿出" + good);
System.out.println("賣家：等待買家...");
String money = exchanger.exchange(good);
System.out.println("賣家：得到" + money);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
```
該程式的輸出如下：
```
賣家：拿出10斤大白菜
賣家：等待買家...
買家：拿出10元
買家：得到10斤大白菜
賣家：得到10元
```
## 五.執行緒中斷機制
有時候我們需要停止一個執行緒，例如一個下載執行緒，該執行緒在沒有下載成功之前不會退出，若此時使用者覺得下載速度慢，不想下載而點選了取消按鈕，此時我們應該停止這個下載執行緒並釋放資源。但是，由於Thread.stop、Thread.suspend和Thread.resume過於暴力而被廢棄，那麼我們應該如何優雅地停止一個執行緒呢？
Java為我們提供了執行緒中斷機制。中斷機制的思想是：一個執行緒不應該由其他執行緒來強制中斷，而是應該由執行緒自己自行判斷。中斷可以看作是由一個執行緒傳送給另外一個執行緒的一種指示，該指示用於表示發起執行緒希望目標執行緒停止其正在執行的操作。但是，中斷一個執行緒並不代表馬上停止該執行緒，而只是通知該執行緒應該中斷了，是否應該中斷以及如何響應中斷則應該交給該執行緒來自行處理。
> 在Java的API或語言規範中，並沒有將中斷與任何取消語義關聯起來。但實際上，如果在取消之外的其他操作中使用中斷，那麼都是不合適的，並且很難支援起更大的應用。
### 1.API
實際上，每個執行緒內部都有一個boolean型別的中斷標記，當中斷一個執行緒時，該執行緒內部的中斷標記將會被設定為true。以下是Thread類中與中斷有關的三個方法：
```java
void interrupt()
boolean isInterrupted()
static boolean interrupted()
```
下面將分別對這三個方法進行介紹。
#### (1)interrupt
呼叫一個執行緒的interrupt方法會將該執行緒的中斷標記設定為true。
1. 如果該執行緒由於呼叫Object.wait()、Object.wait(long)、Object.wait(long, int)、Thread.join()、Thread.join(long)、Thread.join(long, int)、Thread.sleep()或Thread.sleep(long, int)而進入等待狀態，該執行緒的中斷標記將會被清除並收到一個InterruptedException。
2. 如果該執行緒阻塞在一個基於InterruptibleChannel的I/O操作上，這個channel將會被關閉並收到一個ClosedByInterruptException。
3. 如果該執行緒被阻塞在一個Selector裡，則該執行緒會馬上從選擇操作中返回，返回值可能是非0值，就好像Selector的wakeup方法被呼叫過一樣。
4. 如果以上條件均不成立，那麼該執行緒僅僅只是中斷標記被設定為true，並不會表現出其他行為。
上面的2、3兩個條件與NIO有關，這裡只是順便提到而已。後續會推出關於NIO的系列教程。
#### (2)isInterrupted
該方法較為簡單，只是返回該執行緒的中斷標記，並不會影響該中斷標記和產生其他行為。
#### (3)interrupted
該方法是一個靜態方法，也會返回該執行緒的中斷標記。但是該方法與isInterrupted最大的區別在於該方法會清除執行緒的中斷狀態。例如，如果當前執行緒已經被中斷，那麼呼叫interrupted方法將會返回true並同時清除中斷狀態；如果當前執行緒未被中斷，則會返回false。這個方法的好處在於返回了執行緒中斷標記的同時還清除了中斷標記。
### 2.執行緒在不同狀態下對中斷的反應
執行緒一共有6種狀態，分別是NEW、RUNNABLE、WAITING、TIMED_WAITING、BLOCKED和TERMINATED。在不同狀態下，執行緒可能會對中斷產生不同的反應。
#### (1)NEW/TERMINATED
由於處於NEW狀態的執行緒還沒有啟動，而處於TERMINATED狀態的執行緒已經終止，Java認為對處於這兩種狀態下的執行緒進行中斷毫無意義，所以並不會將執行緒的中斷標識設定為true，也不會產生其他的行為。
```java
public class NewAndTerminatedDemo {
public static void main(String[] args) {
Thread thread = new Thread();
System.out.println(thread.getState());
thread.interrupt();
System.out.println(thread.isInterrupted());
thread.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(thread.getState());
thread.interrupt();
System.out.println(thread.isInterrupted());
}
}
```
上面的例子輸出如下：
```
NEW
false
TERMINATED
false
```
#### (2)RUNNABLE
處於RUNNABLE狀態下的執行緒被中斷後，除了中斷標記被設定為true外不會產生其他行為，下面我們來做個實驗：
```java
public class RunnableDemo {
public static void main(String[] args) {
TimeWasteThread timeWasteThread = new TimeWasteThread();
timeWasteThread.start();
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
timeWasteThread.interrupt();
System.out.println("The interrupt flag of timeWasteThread is " + timeWasteThread.isInterrupted());
}
private static class TimeWasteThread extends Thread {
@Override
public void run() {
while (true) {
System.out.println("The 40th fibonacci number is " + fibonacci(40));
}
}
private int fibonacci(int n) {
if (n == 1 || n == 2) {
return 1;
}
return fibonacci(n - 1) + fibonacci(n - 2);
}
}
}
```
在上面的例子中，TimeWasteThread內部執行了一個非常耗時的操作——計算第40個斐波那契數（這種寫法在每次計算時都需要重新遞迴，因此非常耗時），這樣做的目的是使它一直處於RUNNABLE狀態，我們既不希望它太快結束，也不希望使用sleep方法，因為這是下一小節要討論的內容。主執行緒在啟動TimeWasteThread兩秒後中斷了該執行緒。該程式的輸出如下：
```
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The interrupt flag of timeWasteThread is true
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
...
```
可以看到，在主執行緒對TimeWasteThread發出了中斷訊號後，TimeWasteThread的中斷標記確實變成了true，可以程式仍然在繼續執行，沒有受到任何影響。雖然主執行緒已經呼叫了TimeWasteThread的interrupt方法，可該執行緒並沒有中斷執行。既然如此，那我要這中斷機制有何用？
我們在上面提到過，中斷機制的思想是一個執行緒是否中斷應該由該執行緒來判斷，而不應該由其他執行緒來控制。因此，我們可以線上程內部來判斷當前執行緒是否需要被中斷，然後做出相應的決策。
基於這種思想，我們將TimeWasteThread的run方法修改如下：
```java
@Override
public void run() {
while (!Thread.interrupted()) {
System.out.println("The 40th fibonacci number is " + fibonacci(40));
}
}
```
重新執行該程式，輸出如下：
```
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The 40th fibonacci number is 102334155
The interrupt flag of timeWasteThread is true
The 40th fibonacci number is 102334155
```
可以看到，TimeWasteThread在收到中斷訊號後很快就停了下來，達到了中斷的目的。
#### (3)WAITING/TIMED_WAITING
這兩種狀態本質上可以看作是等待狀態，只不過一個是無限期等待，而另一個是有時間限制的等待，因此放在一起討論。
下面的例子中，我們讓SleepingThread進入TIMED_WAITING狀態後將其中斷：
```java
public class WaitingDemo {
public static void main(String[] args) {
Thread sleepingThread = new SleepingThread();
sleepingThread.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
sleepingThread.interrupt();
}
private static class SleepingThread extends Thread {
@Override
public void run() {
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
System.out.println("Thread has been interrupted.");
System.out.println("The interrupt flag of sleepingThread is " + Thread.currentThread().isInterrupted());
}
}
}
}
```
該程式的輸出如下：
```
Thread has been interrupted.
The interrupt flag of sleepingThread is false
```
在上面的例子中，主執行緒在SleepingThread處於TIMED_WAITING狀態時將其中斷，此時SleepingThread被喚醒並丟擲了一個InterruptedException異常。也就是說，處於WAITING或者TIMED_WAITING狀態的執行緒被中斷時往往是通過InterruptedException異常（有時也會通過其他異常，例如ClosedByInterruptException異常）來進行通知的。
不過，當SleepingThread由於中斷而被喚醒時，它的中斷標記卻是false，而我們確確實實在主執行緒中已經呼叫了它的interrupt方法，這是為什麼呢？實際上，按照慣例，丟擲InterruptedException異常的方法，通常會在丟擲該異常時將當前執行緒的中斷標記重置為false。這是因為，當捕獲到InterruptedException異常時，我們已經知道執行緒被中斷了，那麼此時的中斷標記對於我們來說已經沒用了，但是我們還需要手動將它再設定為false，方便下次使用。因此，為了使用方便，方法在丟擲InterruptedException異常之前應該將當前執行緒的終端標記重置為false。
#### (4)BLOCKED
只有在等待一個物件的監視器的執行緒才會處於BLOCKED狀態。下面我們通過一個例子來演示對處於BLOCKED狀態的執行緒呼叫interrupt方法會發生什麼。
```java
public class BlockedDemo {
private static final Integer foo = 1;
public static void main(String[] args) {
Thread thread1 = new Thread1();
thread1.start();
Thread thread2 = new Thread2();
thread2.start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("The state of thread2 is " + thread2.getState());
thread2.interrupt();
System.out.println("The interrupt flag of thread2 is " + thread2.isInterrupted());
System.out.println("The state of thread2 is " + thread2.getState());
}
private static class Thread1 extends Thread {
@Override
public void run() {
synchronized (foo) {
for (int i = 0; i < 5; i++) {
fibonacci(40);
}
}
}
private int fibonacci(int n) {
if (n == 1 || n == 2) {
return 1;
}
return fibonacci(n - 1) + fibonacci(n - 2);
}
}
private static class Thread2 extends Thread {
@Override
public void run() {
System.out.println("Thread2 tries to get the monitor of object foo.");
synchronized (foo) {
System.out.println("Thread2 got the monitor of object foo.");
System.out.println("The interrupt flag of thread2 is " + Thread.currentThread().isInterrupted());
}
}
}
}
```
該程式的輸出如下：
```
Thread2 tries to get the monitor of object foo (1)
The state of thread2 is BLOCKED (2)
The interrupt flag of thread2 is true (3)
The state of thread2 is BLOCKED (4)
Thread2 got the monitor of object foo (5)
The interrupt flag of thread2 is true (6)
```
下面依次分析每一條輸出：
(1)Thread2啟動，嘗試獲取foo物件的監視器；
(2)由於Thread1先獲取到了foo物件的監視器且持有較長時間，Thread2需要等待Thread1釋放foo物件的監視器，因此Thread2進入BLOCKED狀態；
(3)對處於BLOCKED狀態的Thread2執行interrupt方法，該執行緒的中斷標記變成true；
(4)對Thread2執行interrupt方法後，該執行緒仍然處於BLOCKED狀態；
(5)因為Thread1釋放了foo物件的監視器，所以Thread2獲取到了該監視器，繼續執行下面的程式碼；
(6)Thread2重新進入RUNNABLE狀態後，中斷標記仍然為true，此時可以根據中斷標記來決定之後的邏輯。
綜上，對處於BLOCKED狀態的執行緒呼叫interrupt方法，僅僅只是將該執行緒的中斷標記設定為true，除此之外沒有任何