《Java編程思想》讀書筆記（五）
- 2022 年 9 月 6 日
- 筆記
- JAVA, 《Java編程思想》讀書筆記
前言：本文是《Java編程思想》讀書筆記系列的最後一章，本章的內容很多，需要細讀慢慢去理解，文中的示例最好在自己電腦上多運行幾次，相關示例完整程式碼放在碼雲上了，碼雲地址：//gitee.com/reminis_com/thinking-in-java
第二十一章：並發
基本的執行緒機制
並發編程使我們可以將程式劃分為多個分離的、獨立運行的任務。通過使用多執行緒機制，這些獨立任務（也被稱為子任務）中的每一個都將由執行執行緒來驅動。一個執行緒就是在進程中的一個單一的順序控制流，因此，單個進程可以擁有多個並發執行的任務，但是你的程式使得每個任務都好像有其自己的CPU一樣。其底層機制是切分CPU時間，但通常你不需要考慮它。
執行緒模型為編程帶來了便利，它簡化了在單一程式中同時交織在一起的多個操作的處理。在使用執行緒時，CPU將輪流給每個任務分配其佔用時間「。每個任務都覺得自己在一直佔用CPU，但事實上CPU時間是劃分成片段分配給了所有的任務（例外情況是程式確實運行在多個CPU之上）。執行緒的一大好處是可以使你從這個層次抽身出來，即程式碼不必知道它是運行在具有一個還是多個CPU的機器上。所以，使用執行緒機制，是一種建立透明的、可擴展的程式的方法，如果程式運行的太慢，為機器增添一個CPU就很容易的加快程式的運行速度。多任務和多執行緒往往是使用多處理器系統的最合理方式。
定義任務
執行緒可以驅動任務，因此你需要一種描述任務的方式。
- 使用Runnable介面：要想定義任務，只需實現Runable介面並編寫run()方法，使得該任務可以執行你的命令，例如，下面的LiftOff任務將顯示發射之前的倒計時：
package concurrency;
/**
* @author Mr.Sun
* @date 2022年09月03日 10:14
*
* 演示通過實現Runnable介面定義任務
*
* <p>
* 顯示發射之前的倒計時
* </p>
*/
public class LiftOff implements Runnable {
protected int countDown = 10;
private static int taskCount = 0;
private final int id = taskCount++;
public LiftOff() {
}
public LiftOff(int countDown) {
this.countDown = countDown;
}
public String status() {
return "#" + id + "(" + (countDown > 0 ? countDown : "LiftOff!") + "). ";
}
@Override
public void run() {
while (countDown-- > 0) {
System.out.print(status());
Thread.yield();
}
}
} ///:~
標識符id可以用來區分任務的多個實例，它是final的，因為它一旦被初始化之後就不希望被修改。任務的run()方法通常總會有某種形式的循環，使得任務一直運行下去直到不再需要，所以要設定跳出循環的條件（有一種選擇是直接從run）返回）。通常，run()被寫成無限循環的形式，這就意味著，除非有某個條件使得run()終止，否則它將永遠運行下去（在本章後面將會看到如何安全地終止執行緒）。
在run()中對靜態方法Thread.yield()的調用是對執行緒調度器（Java執行緒機制的一部分，可以將CPU從一個執行緒轉移給另一個執行緒）的一種建議，它在聲明:"我已經執行完生命周期中最重要的部分了，此刻正是切換給其他任務執行一段時間的大好時機。」這完全是選擇性的，但是這裡使用它是因為它會在這些示例中產生更加有趣的輸出∶你更有可能會看到任務換進換出的證據。
在下面的實例中，這個任務的run()不是由單獨的執行緒驅動的，它是在main()中直接調用的（實際上，這裡仍舊使用了執行緒，即總是分配給main()的那個執行緒）∶
package concurrency;
/**
* @author Mr.Sun
* @date 2022年09月03日 10:23
*/
public class MainThread {
public static void main(String[] args) {
LiftOff launch = new LiftOff();
launch.run();
}
} /* Output:
#0(9). #0(8). #0(7). #0(6). #0(5). #0(4). #0(3). #0(2). #0(1). #0(LiftOff!).
*///:~
- Thread類：將Runnable對象轉變為工作任務的傳統方式是把它交給一個Thread構造器，下面的示例展示了如何使用Thread來驅動LiftOff對象：
package concurrency;
/**
* @author Mr.Sun
* @date 2022年09月03日 10:29
*
* 將Runnable對象轉變為工作任務的傳統方式是把它交給一個Thread構造器
*/
public class BasicThreads {
public static void main(String[] args) {
Thread t = new Thread(new LiftOff());
t.start();
System.out.println("Waiting for LiftOff");
}
} /* Output:
Waiting for LiftOff
#0(9). #0(8). #0(7). #0(6). #0(5). #0(4). #0(3). #0(2). #0(1). #0(LiftOff!).
*///:~
- 使用Executor
Java SE5的java.util.concurrent包中的執行器（Executor）將為你管理Thread對象，從而簡化了並發編程。Executor在客戶端和任務執行之間提供了一個間接層；與客戶端直接執行任務不同，這個中介對象將執行任務。Executor允許你管理非同步任務的執行，而無須顯式地管理執行緒的生命周期。Executor在Java SE5/6中是啟動任務的優選方法。
我們可以使用Executor來代替在BasicThreads.java中顯示地創建Thread對象。LiftOff 對象知道如何運行具體的任務，與命令設計模式一樣，它暴露了要執行的單一方法。ExecutorService（具有服務生命周期的Executor，例如關閉）知道如何構建恰當的上下文來執行Runnable對象。在下面的示例中，CachedThreadPool將為每個任務都創建一個執行緒。注意，ExecutorService對象是使用靜態的Executor方法創建的，這個方法可以確定其Executor類型∶
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author Mr.Sun
* @date 2022年09月03日 16:04
*
* 使用Executor
*/
public class CachedThreadPool {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
exec.execute(new LiftOff());
}
exec.shutdown();
}
}/* Output:
#0(9). #1(9). #2(9). #3(9). #0(8). #4(9). #1(8). #2(8). #3(8). #0(7). #4(8). #1(7). #2(7). #3(7). #0(6). #4(7). #1(6). #2(6). #3(6). #0(5). #1(5). #4(6). #3(5). #1(4). #2(5). #4(5). #0(4). #1(3). #2(4). #3(4). #0(3). #4(4). #1(2). #2(3). #3(3). #0(2). #4(3). #1(1). #2(2). #3(2). #0(1). #4(2). #1(LiftOff!). #2(1). #3(1). #0(LiftOff!). #4(1). #2(LiftOff!). #3(LiftOff!). #4(LiftOff!).
*///:~
你可以很容易地將上面示例中的CachedThreadPool替換為不同類型的Executor。
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author Mr.Sun
* @date 2022年09月03日 16:10
*/
public class FixedThreadPool {
public static void main(String[] args) {
ExecutorService exec = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
exec.execute(new LiftOff());
}
exec.shutdown();
}
}
有了FixedThreadPool，你就可以一次性預先執行代價高昂的執行緒分配，因而也就可以限制執行緒的數量了。這可以節省時間，因為你不用為每個任務都固定地付出創建執行緒的開銷。在事件驅動的系統中，需要執行緒的事件處理器，通過直接從池中獲取執行緒，也可以如你所願地儘快得到服務。你不會濫用可獲得的資源，因為FixedThreadPool使用的Thread對象的數量是有界的。
注意，在任何執行緒池中，現有執行緒在可能的情況下，都會被自動復用。
儘管本書將使用CachedThreadPool，但是也應該考慮在產生執行緒的程式碼中使用FixedThreadPool。CachedThreadPool在程式執行過程中通常會創建與所需數量相同的執行緒，然後在它回收舊執行緒時停止創建新執行緒，因此它是合理的Executor的首選。只有當這種方式會引發問題時，你才需要切換到FixedThreadPool。
SingleThreadExecutor就像是執行緒數量為1的FixedThreadPool。這對於你希望在另一個執行緒中連續運行的任何事物（長期存活的任務）來說，都是很有用的，例如監聽進入的套接字連接的任務。它對於希望在執行緒中運行的短任務也同樣很方便，例如，更新本地或遠程日誌的小任務，或者是事件分發執行緒。
如果向SingleThreadExecutor提交了多個任務，那麼這些任務將排隊，每個任務都會在下一個任務開始之前運行結束，所有的任務將使用相同的執行緒。在下面的示例中，你可以看到每個任務都是按照它們被提交的順序，並且是在下一個任務開始之前完成的。因此，SingleThread-Executor會序列化所有提交給它的任務，並會維護它自己（隱藏）的懸掛任務隊列。
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author Mr.Sun
* @date 2022年09月03日 16:17
*/
public class SingleThreadExecutor {
public static void main(String[] args) {
ExecutorService exec = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++) {
exec.execute(new LiftOff());
}
exec.shutdown();
}
} /* Output:
#0(9). #0(8). #0(7). #0(6). #0(5). #0(4). #0(3). #0(2). #0(1). #0(LiftOff!). #1(9). #1(8). #1(7). #1(6). #1(5). #1(4). #1(3). #1(2). #1(1). #1(LiftOff!). #2(9). #2(8). #2(7). #2(6). #2(5). #2(4). #2(3). #2(2). #2(1). #2(LiftOff!). #3(9). #3(8). #3(7). #3(6). #3(5). #3(4). #3(3). #3(2). #3(1). #3(LiftOff!). #4(9). #4(8). #4(7). #4(6). #4(5). #4(4). #4(3). #4(2). #4(1). #4(LiftOff!).
*///:~
作為另一個示例，假設你有大量的執行緒，那它們運行的任務將使用文件系統。你可以用SingleThreadExecutor來運行這些執行緒，以確保任意時刻在任何執行緒中都只有唯一的任務在運行。在這種方式中，你不需要在共享資源上處理同步（同時不會過度使用文件系統）。有時更好的解決方案是在資源上同步（你將在本章稍後學習），但是SingleThreadExecutor可以讓你省去只是為了維持某些事物的原型而進行的各種協調努力。通過序列化任務，你可以消除對序列化對象的需求。
- 從任務種產生返回值：Runnable是執行工作的獨立任務，但是它不返回任何值。如果你希望在任務完成時能夠返回一個值，那麼可以實現Callable介面而不是Runnable介面。在Java SE5中引入的Callable是一種具有類型參數的泛型，它的類型參數表示的是從方法call()中返回的值，並且必須使用ExecutorService.submit()方法調用它，下面是一個簡單示例：
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
/**
* @author Mr.Sun
* @date 2022年09月03日 16:23
*
* 從任務中產生返回值
*/
public class CallableDemo {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
List<Future<String>> futureList = new ArrayList<>();
for (int i = 0; i < 10; i++) {
futureList.add(exec.submit(new TaskWithResult(i)));
}
for (Future<String> result : futureList) {
try {
// get() 會阻塞知道任務完成
System.out.println(result.get());
} catch (InterruptedException e) {
System.out.println(e);
return;
} catch (ExecutionException e) {
System.out.println(e);
} finally {
exec.shutdown();
}
}
}
}
class TaskWithResult implements Callable<String> {
private int id;
public TaskWithResult(int id) {
this.id = id;
}
@Override
public String call() throws Exception {
return "result of TaskWithResult " + id;
}
} /* Output:
result of TaskWithResult 0
result of TaskWithResult 1
result of TaskWithResult 2
result of TaskWithResult 3
result of TaskWithResult 4
result of TaskWithResult 5
result of TaskWithResult 6
result of TaskWithResult 7
result of TaskWithResult 8
result of TaskWithResult 9
*///:~
submit()方法會產生Future對象，它用Callable返回結果的特定類型進行了參數化。你可以用isDone()方法來查詢Future是否已經完成。當任務完成時，它具有一個結果，你可以調用get()方法來獲取該結果。你也可以不用isDone()進行檢查就直接調用get()，在這種情況下，get()將阻塞，直至結果準備就緒。你還可以在試圖調用get()來獲取結果之前，先調用具有超時的get()，或者調用isDone()來查看任務是否完成。
捕獲異常
由於執行緒的本質特性，使得你不能捕獲從執行緒中逃逸的異常。下面的任務總是拋出一個異常，該異常會傳播到run()方法的外部，並且main()展示了當你運行它時所發生的事情：
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author Mr.Sun
* @date 2022年09月03日 17:00
*
* 執行緒拋出異常
*/
public class ExceptionThread implements Runnable {
@Override
public void run() {
throw new RuntimeException();
}
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new ExceptionThread());
exec.shutdown();
}
}
運行結果如下圖：
將main()的主體放到try-catch子句中是沒有作用的，如下：
我們發現使用try-catch後，異常仍未被捕獲。
為了解決這個問題，我們要修改Executor產生執行緒的方式。Thread.UncaughtExceptionHandler是Java SE5中的新介面，它允許你在每個Thread對象上都附著一個異常處理器。Thread.UncaughtExceptionHandler.uncaughtException()會在執行緒因未捕獲的異常而臨近死亡時被調用。為了使用它，我們創建了一個新類型的ThreadFactory，它將在每個新創建的Thread對象上附著一個Thread.UncaughtExceptionHandler。我們將這個工廠傳遞給Executors創建新的ExecutorService的方法:
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
/**
* @author Mr.Sun
* @date 2022年09月03日 17:10
*
* 為每個Thread對象附著一個異常處理器
*/
public class CaptureUncaughtException {
public static void main(String[] args) {
ExecutorService exec = Executors.newCachedThreadPool(new HandlerThreadFactory());
exec.execute(new ExceptionThread2());
}
}
class ExceptionThread2 implements Runnable {
@Override
public void run() {
Thread t = Thread.currentThread();
System.out.println("run() by " + t);
System.out.println("eh = " + t.getUncaughtExceptionHandler());
throw new RuntimeException();
}
}
class MyUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
@Override
public void uncaughtException(Thread t, Throwable e) {
System.out.println("caught " + e);
}
}
class HandlerThreadFactory implements ThreadFactory {
@Override
public Thread newThread(Runnable r) {
System.out.println(this + " creating new Thread");
Thread t = new Thread(r);
System.out.println("created " + t);
t.setUncaughtExceptionHandler(new MyUncaughtExceptionHandler());
System.out.println("eh = " + t.getUncaughtExceptionHandler());
return t;
}
} /* Output:
concurrency.HandlerThreadFactory@1540e19d creating new Thread
created Thread[Thread-0,5,main]
eh = concurrency.MyUncaughtExceptionHandler@677327b6
run() by Thread[Thread-0,5,main]
eh = concurrency.MyUncaughtExceptionHandler@677327b6
caught java.lang.RuntimeException
*///:~
在程式中添加了額外的跟蹤機制，用來驗證工廠創建的執行緒會傳遞給UncaughtExceptionHandler。你可以看到，未捕獲的異常是通過uncaughtException來捕獲的。
共享受限資源
1. 不正確訪問資源
考慮下面的例子，其中一個任務產生產生偶數，而其他任務消費這些數字。而這些消費者任務的唯一工作就是校驗偶數的有效性
package concurrency;
/**
* @author Mr.Sun
* @date 2022年09月03日 21:49
*
* <p>
* 創建一個名為IntGenerator的抽象類，它包含EventChecker必須了解的必不可少的方法，即一個next()方法，和一個可以執行撤銷的方法
* </p>
*/
public abstract class IntGenerator {
private volatile boolean canceled = false;
public abstract int next();
/**
* 修改canceled標誌的狀態
*/
public void cancel() {
canceled = true;
}
/**
* 查看該對象是否已被撤銷
*/
public boolean isCanceled() {
return canceled;
}
}
任何IntGenerator 都可以使用下面的EventChecker類來測試：
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @author Mr.Sun
* @date 2022年09月03日 21:52
*
* 校驗偶數的有效性
*/
public class EventChecker implements Runnable {
private IntGenerator generator;
private final int id;
public EventChecker(IntGenerator generator, int id) {
this.generator = generator;
this.id = id;
}
@Override
public void run() {
while (!generator.isCanceled()) {
int val = generator.next();
if (val % 2 != 0) {
System.out.println(val + " not event");
// 取消 事件檢查
generator.cancel();
}
}
}
public static void test(IntGenerator gp, int count) {
System.out.println("按 Ctrl + C 退出");
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < count; i++) {
exec.execute(new EventChecker(gp, i));
}
exec.shutdown();
}
public static void test(IntGenerator gp) {
// count 默認值
test(gp, 10);
}
}
注意，在本例中可以被撤銷的類不是Runnable，而所有依賴於IntGenerator對象的EvenChecker任務將測試它，以查看它是否已經被撤銷，正如你在run()中所見。通過這種方式，共享公共資源（IntGenerator）的任務可以觀察該資源的終止訊號。這可以消除所謂競爭條件，即兩個或更多的任務競爭響應某個條件，因此產生衝突或不一致結果的情況。你必須仔細考慮並防範並發系統失敗的所有可能途徑，例如，一個任務不能依賴於另一個任務，因為任務關閉的順序無法得到保證。這裡，通過使任務依賴於非任務對象，我們可以消除潛在的競爭條件。
test()方法通過啟動大量使用相同的IntGenerator的EvenChecker，設置並執行對任何類型的IntGenerator的測試。如果IntGenerator引發失敗，那麼test()將報告它並返回，否則，你必須按下Ctrl-C來終止它。
我們看到的第一個IntGenerator有一個可以產生一系列數值的next():
package concurrency;
/**
* @author Mr.Sun
* @date 2022年09月03日 22:02
*/
public class EventGenerator extends IntGenerator {
private int currentEventVal = 0;
@Override
public int next() {
++currentEventVal;
// 可能出現資源競爭問題的地方
++currentEventVal;
return currentEventVal;
}
public static void main(String[] args) {
EventChecker.test(new EventGenerator());
}
}
一個任務有可能在另一個任務執行第一個對currentEvenValue的遞增操作之後，但是沒有執行第二個操作之前，調用next()方法。這將使這個值處於「不恰當」的狀態。為了證明這是可能發生的，EvenChecker.test()創建了一組EvenChecker對象，以連續地讀取並輸出同一個EvenGenerator，並測試檢查每個數值是否都是偶數。如果不是，就會報告錯誤，而程式也將關閉。
有一點很重要，那就是要注意到遞增程式自身也需要多個步驟，並且在遞增過程中任務可能會被執行緒機制掛起——也就是說，在Java中，遞增不是原子性的操作。因此，如果不保護任務，即使單一的遞增也不是安全的。
2. 解決共享資源競爭
前面的示例展示了使用執行緒時的一個基本問題∶你永遠都不知道一個執行緒何時在運行。想像一下，你坐在桌邊手拿叉子，正要去叉盤子中的最後一片食物，當你的叉子就要夠著它時，這片食物突然消失了，因為你的執行緒被掛起了，而另一個餐者進入並吃掉了它。這正是在你編寫並發程式時需要處理的問題。對於並發工作，你需
某種方式來防止兩個任務訪問相同的資源，至少在關鍵階段不能出現這種情況。
防止這種衝突的方法就是當資源被一個任務使用時，在其上加鎖。第一個訪問某項資源的任務必須鎖定這項資源，使其他任務在其被解鎖之前，就無法訪問它了，而在其被解鎖之時，另一個任務就可以鎖定並使用它，以此類推。
基本上所有的併發模式在解決執行緒衝突問題的時候，都是採用序列化訪問共享資源的方案。這意味著在給定時刻只允許一個任務訪問共享資源。通常這是通過在程式碼前面加上一條鎖語句來實現的，這就使得在一段時間內只有一個任務可以運行這段程式碼。因為鎖語句產生了一種互相排斥的效果，所以這種機制常常稱為互斥量(mutex)。
考慮一下屋子裡的浴室多個人（即多個由執行緒驅動的任務）都希望能單獨使用浴室（即共享資源）。為了使用浴室，一個人先敲門，看看是否能使用。如果沒人的話，他就進入浴室並鎖上門。這時其他人要使用浴室的話，就會被「阻擋」，所以他們要在浴室門口等待，直到浴室可以使用。當浴室使用完畢，就該把浴室給其他人使用了（別的任務就可以訪問資源了），這個比喻就有點不太準確了。事實上，人們並沒有排隊，我們也不能確定誰將是下一個使用浴室的人，因為執行緒調度機制並不是確定性的。實際情況是∶等待使用浴室的人們簇擁在浴室門口，當鎖住浴室門的那個人打開鎖準備離開的時候，離門最近的那個人可能進入浴室。如前所述，可以通過yield()和setPriority()來給執行緒調度器提供建議，但這些建議未必會有多大效果，這取決於你的具體平台和JVM實現。
1. 使用synchronized關鍵字
Java以提供關鍵字synchronized的形式，為防止資源衝突提供了內置支援。當任務要執行被synchronized關鍵字保護的程式碼片段的時候，它將檢查鎖是否可用，然後獲取鎖，執行程式碼，釋放鎖。共享資源一般是以對象形式存在的記憶體片段，但也可以是文件、輸入/輸出埠，或者是印表機。要控制對共享資源的訪問，得先把它包裝進一個對象。然後把所有要訪問這個資源的方法標記為synchronized。如果某個任務處於一個對標記為synchronized的方法的調用中，那麼在這個執行緒從該方法返回之前，其他所有要調用類中任何標記為synchronized方法的執行緒都會被阻塞。
在生成偶數的程式碼中，你已經看到了，你應該將類的數據成員都聲明為private的，而且只能通過方法來訪問這些數據；所以可以把方法標記為synchronized來防止資源衝突。下面是聲明synchronized方法的方式∶
synchronized void f() { /* ... */ }
synchronized void g() { /* ... */ }
所有對象都自動含有單一的鎖（也稱為監視器）。當在對象上調用其任意synchronized方法的時候，此對象都被加鎖，這時該對象上的其他synchronized方法只有等到前一個方法調用完畢並釋放了鎖之後才能被調用。對於前面的方法，如果某個任務對對象調用了f()，對於同一個對象而言，就只能等到f()調用結束並釋放了鎖之後，其他任務才能調用f()和g()。所以，對於某個特定對象來說，其所有synchronized方法共享同一個鎖，這可以被用來防止多個任務同時訪問被編碼為對象記憶體。
注意，在使用並發時，將域設置為private是非常重要的，否則，synchronized關鍵字就不能防止其他任務直接訪問域，這樣就會產生衝突。
一個任務可以多次獲得對象的鎖。如果一個方法在同一個對象上調用了第二個方法，後者又調用了同一對象上的另一個方法，就會發生這種情況。JVM負責跟蹤對象被加鎖的次數。如果一個對象被解鎖（即鎖被完全釋放），其計數變為0。在任務第一次給對象加鎖的時候，計數變為1。每當這個相同的任務在這個對象上獲得鎖時，計數都會遞增。顯然，只有首先獲得了鎖的任務才能允許繼續獲取多個鎖。每當任務離開一個synchronized方法，計數遞減，當計數為零的時候，鎖被完全釋放，此時別的任務就可以使用此資源。
針對每個類，也有一個鎖（作為類的Class對象的一部分）；所以synchronized static方法可以在類的範圍內防止對static數據的並發訪問。
同步控制EventGenerator，通過在EventGenerator.java中加入synchronized 關鍵字，可以防止不希望的執行緒訪問：
package concurrency;
/**
* @author Mr.Sun
* @date 2022年09月04日 10:49
*
* 同步控制EventGenerator
*/
public class SyncEventGenerator extends IntGenerator {
private int currentCountVal = 0;
@Override
public synchronized int next() {
++currentCountVal;
Thread.yield();
++currentCountVal;
return currentCountVal;
}
public static void main(String[] args) {
EventChecker.test(new SyncEventGenerator());
}
}
對Thread.yield()的調用被插入到了兩個遞增操作之間，以提高在currentEvenValue是奇數狀態時上下文切換的可能性。因為互斥可以防止多個任務同時進入臨界區，所以這不會產生任何失敗。但是如果失敗將會發生，調用yield()是一種促使其發生的有效方式。
第一個進入next()的任務將獲得鎖，任何其他試圖獲取鎖的任務都將從其開始嘗試之時被阻塞，直至第一個任務釋放鎖。通過這種方式，任何時刻只有一個任務可以通過由互斥量看護的程式碼。
2. 使用顯示的Lock對象
Java SE5的java.util.concurrent類庫還包含有定義在java.util.concurrent.locks中的顯式的互斥機制。Lock對象必須被顯式地創建、鎖定和釋放。因此，它與內建的鎖形式相比，程式碼缺乏優雅性。但是，對於解決某些類型的問題來說，它更加靈活。下面用顯式的Lock重寫的是SyncEventGenerator.java:
package concurrency;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author Mr.Sun
* @date 2022年09月04日 11:13
*
* 使用顯示的Lock對象重寫SyncEventGenerator
*/
public class MutexEventGenerator extends IntGenerator{
private int currentCountVal = 0;
private Lock lock = new ReentrantLock();
@Override
public int next() {
lock.lock();
try {
++currentCountVal;
Thread.yield();
++currentCountVal;
return currentCountVal;
} finally {
lock.unlock();
}
}
public static void main(String[] args) {
EventChecker.test(new MutexEventGenerator());
}
}
MutexEvenGenerator添加了一個被互斥調用的鎖，並使用lock()和unlock()方法在next()內部創建了臨界資源。當你在使用Lock對象時，將這裡所示的慣用法內部化是很重要的：緊接著的對lock()的調用，你必須放置在finally子句中帶有unlock()的try-finally語句中。注意，return 語句必須在try子句中出現，以確保unlock()不會過早發生，從而將數據暴露給了第二個任務。
儘管try-finally所需的程式碼比synchronized關鍵字要多，但是這也代表了顯式的Lock對象的優點之一。如果在使用synchronized關鍵字時，某些事物失敗了，那麼就會拋出一個異常。但是你沒有機會去做任何清理工作，以維護系統使其處於良好狀態。有了顯式的Lock對象，你就可以使用finally子句將系統維護在正確的狀態了。
大體上，當你使用synchronized關鍵字時，需要寫的程式碼量更少，並且用戶錯誤出現的可能性也會降低，因此通常只有在解決特殊問題時，才使用顯式的Lock對象。例如，用synchronized 關鍵字不能嘗試著獲取鎖且最終獲取鎖會失敗，或者嘗試著獲取鎖一段時間，然後放棄它，要實現這些，你必須使用concurrent類庫∶
package concurrency;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author Mr.Sun
* @date 2022年09月04日 11:20
*
* 演示使用Lock對象嘗試獲取鎖
*/
public class AttemptLocking {
private ReentrantLock lock = new ReentrantLock();
public void unTimed () {
boolean captured = lock.tryLock();
try {
System.out.println("tryLock(): " + captured);
} finally {
// 獲取到鎖才能執行釋放鎖操作，否則會出現IllegalMonitorStateException異常
if (captured) {
lock.unlock();
}
}
}
public void timed () {
boolean captured = false;
try {
captured = lock.tryLock(2, TimeUnit.SECONDS);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
try {
System.out.println("lock.tryLock(2, TimeUnit.MINUTES): " + captured);
} finally {
if (captured) {
lock.unlock();
}
}
}
public static void main(String[] args) throws InterruptedException {
final AttemptLocking al = new AttemptLocking();
al.unTimed(); // True -- 鎖是可用的
al.timed(); // True -- 鎖是可用的
// 現在創建一個單獨的任務來獲取鎖
new Thread() {
{setDaemon(true);}
@Override
public void run() {
al.lock.lock();
System.out.println("acquired");
}
}.start();
// 給第二個任務一個機會
Thread.sleep(100);
al.unTimed(); // false -- lock grabbed by task
al.timed(); // false -- lock grabbed by task
}
} /* Output:
tryLock(): true
lock.tryLock(2, TimeUnit.MINUTES): true
acquired
tryLock(): false
lock.tryLock(2, TimeUnit.MINUTES): false
*///:~
ReentrantLock允許你嘗試著獲取但最終未獲取鎖，這樣如果其他人已經獲取了這個鎖，那你就可以決定離開去執行其他一些事情，而不是等待直至這個鎖被釋放，就像在untimed()方法中所看到的。在timed()中，做出了嘗試去獲取鎖，該嘗試可以在2秒之後失敗（注意，使用了Java SE5的TimeUnit類來指定時間單位）。在main()中，作為匿名類而創建了一個單獨的Thread，它將獲取鎖，這使得untimed()和timed()方法對某些事物將產生競爭。
顯式的Lock對象在加鎖和釋放鎖方面，相對於內建的synchronized鎖來說，還賦予了你更細粒度的控制力。這對於實現專有同步結構是很有用的，例如用於遍歷鏈接列表中的節點的節節傳遞的加鎖機制（也稱為鎖耦合），這種遍歷程式碼必須在釋放當前節點的鎖之前捕獲下一個節點的鎖。
3. 關於volatile
volatile關鍵字確保了應用中的可視性。如果你將一個域聲明為volatile的，那麼只要對這個域產生了寫操作，那麼所有的讀操作就都可以看到這個修改。即便使用了本地快取，情況也確實如此，volatile域會立即被寫入到主存中，而讀取操作就發生在主存中。
理解原子性和易變性是不同的概念這一點很重要。在非volatile域上的原子操作不必刷新到主存中去，因此其他讀取該域的任務也不必看到這個新值。如果多個任務在同時訪問某個域，那麼這個域就應該是volatile的，否則，這個域就應該只能經由同步來訪問。同步也會導致向主存中刷新，因此如果一個域完全由synchronized方法或語句塊來防護，那就不必將其設置為是volatile的。
使用volatile而不是synchronized的唯一安全的情況是類中只有一個可變的域。再次提醒，你的第一選擇應該是使用synchronized關鍵字，這是最安全的方式，而嘗試其他任何方式都是有風險的。
原子類
Java SE5引入了諸如AtomicInteger、AtomicLong、AtomicReference等特殊的原子性變數類，它們提供下面形式的原子性條件更新操作∶
boolean compareAndSet(expectedValue, updateValue);
這些類被調整為可以使用在某些現代處理器上的可獲得的，並且是在機器級別上的原子性，因此在使用它們時，通常不需要擔心。對於常規編程來說，它們很少會派上用場，但是在涉及性能調優時，它們就大有用武之地了。例如，我們可以使用AtomicInteger來重寫MutexEventGenerator.java∶
package concurrency;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* @author Mr.Sun
* @date 2022年09月04日 15:17
*
* 使用AtomicInteger來重寫MutexEventGenerator
*/
public class AtomicEventGenerator extends IntGenerator {
private AtomicInteger currentEventVal = new AtomicInteger(0);
@Override
public int next() {
return currentEventVal.addAndGet(2);
}
public static void main(String[] args) {
EventChecker.test(new AtomicEventGenerator());
}
}
所有其他形式的同步通過使用AtomicInteger得到了根除。
應該強調的是，Atomic類被設計用來構建javautil.concurrent中的類，因此只有在特殊情況下才在自己的程式碼中使用它們，即便使用了也需要確保不存在其他可能出現的問題。通常依賴於鎖要更安全一些（要麼是synchronized關鍵字，要麼是顯式的Lock對象）。
臨界區
有時，你只是希望防止多個執行緒同時訪問方法內部的部分程式碼而不是防止訪問整個方法。通過這種方式分離出來的程式碼段被稱為臨界區（critical section），它也使用synchronized關鍵字建立。這裡，synchronized被用來指定某個對象，此對象的鎖被用來對花括弧內的程式碼進行同步控制∶
synchronized(syncObject) {
// This code can be accessed
// by only one task at a time 1
}
這也被稱為同步拉制塊；在進入此段程式碼前，必須得到syncObject對象的鎖。如果其他執行緒已經得到這個鎖，那麼就得等到鎖被釋放以後，才能進入臨界區。
通過使用同步控制塊，而不是對整個方法進行同步控制，可以使多個任務訪問對象的時間性能得到顯著提高，下面的例子比較了這兩種同步控制方法。此外，它也演示了如何把一個非保護類型的類，在其他類的保護和控制之下，應用於多執行緒的環境∶
package concurrency;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author Mr.Sun
* @date 2022年09月04日 15:27
*
* 臨界區 測試
* <p>
* 演示了如何把一個非執行緒安全的類，在其他類的保護和控制之下，應用於多執行緒的環境
* </p>
*/
public class CriticalSelection {
public static void main(String[] args) {
PairManager pman1 = new PairManager1(),
pman2 = new PairManager2();
testApproaches(pman1, pman2);
}
// 測試兩種不同的方法
static void testApproaches(PairManager pman1, PairManager pman2) {
ExecutorService exec = Executors.newCachedThreadPool();
PairManipulator
pm1 = new PairManipulator(pman1),
pm2 = new PairManipulator(pman2);
PairChecker
pcheck1 = new PairChecker(pman1),
pcheck2 = new PairChecker(pman2);
exec.execute(pm1);
exec.execute(pm2);
exec.execute(pcheck1);
exec.execute(pcheck2);
try {
TimeUnit.MILLISECONDS.sleep(200);
} catch(InterruptedException e) {
System.out.println("Sleep interrupted");
} finally {
exec.shutdown();
}
System.out.println("pm1: " + pm1 + "\npm2: " + pm2);
System.exit(0);
}
}
class PairChecker implements Runnable {
private PairManager pm;
public PairChecker(PairManager pm) {
this.pm = pm;
}
@Override
public void run() {
while(true) {
pm.checkCounter.incrementAndGet();
pm.getPair().checkState();
}
}
}
class PairManipulator implements Runnable {
private PairManager pm;
public PairManipulator(PairManager pm) {
this.pm = pm;
}
@Override
public void run() {
while(true) {
pm.increment();
}
}
public String toString() {
return "Pair: " + pm.getPair() + " checkCounter = " + pm.checkCounter.get();
}
}
// 執行緒安全的Pair
abstract class PairManager {
AtomicInteger checkCounter = new AtomicInteger(0);
protected Pair p = new Pair();
private List<Pair> storage = Collections.synchronizedList(new ArrayList<>());
public synchronized Pair getPair() {
// 複製一份以確保原件的安全:
return new Pair(p.getX(), p.getY());
}
// 假設這是一個耗時的操作
protected void store(Pair p) {
storage.add(p);
try {
TimeUnit.MILLISECONDS.sleep(50);
} catch (InterruptedException ignore) {}
}
public abstract void increment();
}
// 同步整個方法
class PairManager1 extends PairManager {
@Override
public synchronized void increment() {
p.incrementX();
p.incrementY();
store(getPair());
}
}
// 同步臨界區程式碼塊
class PairManager2 extends PairManager {
@Override
public void increment() {
Pair temp;
synchronized(this) {
p.incrementX();
p.incrementY();
temp = getPair();
}
store(temp);
}
}
// 非執行緒安全
class Pair {
private int x, y;
public Pair(int x, int y) {
this.x = x;
this.y = y;
}
public Pair() {
this(0, 0);
}
public int getX() {
return x;
}
public int getY() {
return y;
}
public void incrementX () {
// 自增操作不是執行緒安全的
x++;
}
public void incrementY () {
// 自增操作不是執行緒安全的
y++;
}
@Override
public String toString() {
return "x: " + x + ", y: " + y ;
}
public class PairValueNotEqualException extends RuntimeException {
public PairValueNotEqualException() {
super("一對不相等的值：" + Pair.this);
}
}
// 任意不變數 —— 兩個變數必須相等
public void checkState() {
if(x != y) {
throw new PairValueNotEqualException();
}
}
} /* Output: (Sample)
pm1: Pair: x: 15, y: 15 checkCounter = 272565
pm2: Pair: x: 16, y: 16 checkCounter = 3956974
*///:~
正如注釋中註明的，Pair不是執行緒安全的，因為它的約束條件（雖然是任意的）需要兩個變數要維護成相同的值。此外，如本章前面所述，自增加操作不是執行緒安全的，並且因為Pair類沒有任何方法被標記為synchronized，所以不能保證一個Pair對象在多執行緒程式中不會被破壞。
你可以想像一下這種情況∶某人交給你一個非執行緒安全的Pair類，而你需要在一個執行緒環境中使用它。通過創建PairManager類就可以實現這一點，PairManager類持有一個Pair對象並控制對它的一切訪問。注意唯一的public方法是getPair()，它是synchronized的。對於抽象方法increment()，對increment()的同步控制將在實現的時候進行處理。
至於PairManager類的結構，它的一些功能在基類中實現，並且其一個或多個抽象方法在派生類中定義，這種結構在設計模式中稱為模板方法。設計模式使你得以把變化封裝在程式碼里；在此，發生變化的部分是模板方法increment()。在PairManager1中，整個increment()方法是被同步控制的；但在PairManager2中，increment()方法使用同步控制塊進行同步。注意，synchronized關鍵字不屬於方法特徵簽名的組成部分，所以可以在覆蓋方法的時候加上去。
store()方法將一個Pair對象添加到了synchronized ArrayList中，所以這個操作是執行緒安全的。因此，該方法不必進行防護，可以放置在PairManager2的synchronized語句塊的外部。
PairManipulator被創建用來測試兩種不同類型的PairManager，其方法是在某個任務中調用increment()，而PairChecker則在另一個任務中執行。為了跟蹤可以運行測試的頻度，PairChecker在每次成功時都遞增checkCounter。在main()中創建了兩個PairManipulator對象，並允許它們運行一段時間，之後每個PairManipulator的結果會得到展示。
儘管每次運行的結果可能會非常不同，但一般來說，對於PairChecker的檢查頻率，PairManagerl.increment()不允許有PairManager2.increment（）那樣多。後者採用同步控制塊進行同步，所以對象不加鎖的時間更長。這也是寧願使用同步控制塊而不是對整個方法進行同步控制的典型原因∶使得其他執行緒能更多地訪問（在安全的情況下儘可能多）。
在其它對象上同步
synchronized塊必須給定一個在其上進行同步的對象，並且最合理的方式是，使用其方法正在被調用的當前對象∶synchronized(this)，這正是PairManager2所使用的方式。在這種方式中，如果獲得了synchronized塊上的鎖，那麼該對象其他的synchronized方法和臨界區就不能被調用了。因此，如果在this上同步，臨界區的效果就會直接縮小在同步的範圍內。
有時必須在另一個對象上同步，但是如果你要這麼做，就必須確保所有相關的任務都是在同一個對象上同步的。下面的示例演示了兩個任務可以同時進入同一個對象，只要這個對象上的方法是在不同的鎖上同步的即可∶
package concurrency;
import java.util.Objects;
/**
* @author Mr.Sun
* @date 2022年09月04日 16:19
*
* 在其它對象上同步
*/
public class SyncObject {
public static void main(String[] args) {
final DualSync ds = new DualSync();
new Thread() {
@Override
public void run() {
ds.f();
}
}.start();
ds.g();
}
}
class DualSync {
private Object obj = new Object();
public synchronized void f() {
for (int i = 0; i < 5; i++) {
System.out.println("f()");
Thread.yield();
}
}
public void g() {
synchronized (obj) {
for (int i = 0; i < 5; i++) {
System.out.println("g()");
Thread.yield();
}
}
}
} /* Output: (Sample)
g()
g()
f()
g()
f()
g()
f()
g()
f()
f()
*///:~
DualSync.f()（通過同步整個方法）在this同步，而g()有一個在syncObject上同步的synchronized塊。因此，這兩個同步是互相獨立的。通過在main()中創建調用f()的Thread對這一點進行了演示，因為main()執行緒是被用來調用g()的。從輸出中可以看到，這兩個方式在同時運行，因此任何一個方法都沒有因為對另一個方法的同步而被阻塞。
執行緒本地存儲
防止任務在共享資源上產生衝突的第二種方式是根除對變數的共享。執行緒本地存儲是一種自動化機制，可以為使用相同變數的每個不同的執行緒都創建不同的存儲。因此，如果你有5個執行緒都要使用變數x所表示的對象，那執行緒本地存儲就會生成5個用於x的不同的存儲塊。主要是，它們使得你可以將狀態與執行緒關聯起來。
創建和管理執行緒本地存儲可以由java.lang.ThreadLocal類來實現，如下所示∶
package concurrency;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月04日 16:29
*
* 創建和管理執行緒本地存儲可以使用ThreadLocal來實現
*/
public class ThreadLocalVariableHolder {
private static ThreadLocal<Integer> value = new ThreadLocal<Integer>() {
private Random random = new Random(47);
@Override
protected Integer initialValue() {
return random.nextInt(10000);
}
};
public static void increment () {
value.set(value.get() + 1);
}
public static int get() {
return value.get();
}
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
exec.execute(new Accessor(i));
}
TimeUnit.MILLISECONDS.sleep(1000);
exec.shutdown();
System.exit(0);
}
}
class Accessor implements Runnable {
private final int id;
public Accessor(int id) {
this.id = id;
}
@Override
public void run() {
while (!Thread.currentThread().isInterrupted()) {
ThreadLocalVariableHolder.increment();
System.out.println(this);
Thread.yield();
}
}
@Override
public String toString() {
return "#" + id + ": " + ThreadLocalVariableHolder.get();
}
} /* Output: (Sample)
#0: 9259
#1: 556
#2: 6694
#3: 1862
#4: 962
#0: 9260
#1: 557
#2: 6695
#3: 1863
#4: 963
...
*///:~
ThreadLocal對象通常當作靜態域存儲。在創建ThreadLocal時，你只能通過get()和set()方法來訪問該對象的內容，其中，get()方法將返回與其執行緒相關聯的對象的副本，而set()會將參數插入到為其執行緒存儲的對象中，並返回存儲中原有的對象。increment()和get()方法在ThreadLocalVariableHolder中演示了這一點。注意，increment()和get()方法都不是synchronized 的，因為ThreadLocal保證不會出現競爭條件。
當運行這個程式時，你可以看到每個單獨的執行緒都被分配了自己的存儲，因為它們每個都需要跟蹤自己的計數值，即便只有一個ThreadLocalVariableHolder對象。
終結任務
在某些情況下，任務必須突然地終止，首先，讓我們觀察一個示例，它不僅演示了終止問題，而且還是一個資源共享的示例：
1. 裝飾性花園
在這個模擬程式中，花園委員會希望了解每天通過多個大門進入公園的總人數。每個大門都有一個十字轉門或者某種其他形式的計數器，並且任何一個十字轉門的計數值遞增時，就表示公園中的總人數的共享計數值也會遞增。
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月04日 20:11
*
* 裝飾花園：演示終結任務
* <p>
* 在這個模擬程式中，花園委員會希望了解每天通過多個大門進入公園的總人數。
* 每個大門都有一個十字轉門或者某種其他形式的計數器，並且任何一個十字轉門的計數值遞增時，就表示公園中的總人數的共享計數值也會遞增。
* </p>
*/
public class OrnamentalGarden {
public static void main(String[] args) throws InterruptedException {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
exec.execute(new Entrance(i));
}
TimeUnit.SECONDS.sleep(3);
Entrance.cancel();
exec.shutdown();
if (!exec.awaitTermination(250, TimeUnit.MILLISECONDS)) {
System.out.println("某些任務未被終止！");
}
System.out.println("Total: " + Entrance.getTotalCount());
System.out.println("Sum of Entrances: " + Entrance.sumEntrances());
}
}
class Count {
private int count = 0;
private Random ran = new Random(47);
// 刪除synchronized關鍵字以查看計數失敗
public synchronized int increment() {
int temp = count;
if (ran.nextBoolean()) {
Thread.yield();
}
return count = ++temp;
}
public synchronized int value () {
return count;
}
}
class Entrance implements Runnable {
private static Count count = new Count();
private static List<Entrance> entranceList = new ArrayList<>();
private int number = 0;
private final int id;
private static volatile boolean canceled = false;
/**
* 原子操作
*/
public static void cancel() {
canceled = true;
}
public Entrance(int id) {
this.id = id;
// 將此任務保留在列表中, 也防止無用任務被垃圾回收
entranceList.add(this);
}
@Override
public void run() {
while (! canceled) {
synchronized (this) {
++ number;
}
System.out.println(this + " Total: " + count.increment());
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
System.out.println("sleep interrupted");
}
}
System.out.println("Stopping " + this);
}
public synchronized int getValue() { return number; }
@Override
public String toString() {
return "Entrance " + id + ": " + getValue();
}
public static int getTotalCount() {
return count.value();
}
public static int sumEntrances() {
int sum = 0;
for(Entrance entrance : entranceList)
sum += entrance.getValue();
return sum;
}
} /* Output: (Sample)
Entrance 0: 1 Total: 1
Entrance 2: 1 Total: 3
Entrance 1: 1 Total: 2
Entrance 4: 1 Total: 5
Entrance 3: 1 Total: 4
Entrance 2: 2 Total: 6
Entrance 4: 2 Total: 7
Entrance 0: 2 Total: 8
...
Entrance 3: 29 Total: 143
Entrance 0: 29 Total: 144
Entrance 4: 29 Total: 145
Entrance 2: 30 Total: 147
Entrance 1: 30 Total: 146
Entrance 0: 30 Total: 149
Entrance 3: 30 Total: 148
Entrance 4: 30 Total: 150
Stopping Entrance 2: 30
Stopping Entrance 1: 30
Stopping Entrance 0: 30
Stopping Entrance 3: 30
Stopping Entrance 4: 30
Total: 150
Sum of Entrances: 150
*///:~
這裡使用單個的Count對象來跟蹤花園參觀者的主計數值，並且將其當作Entrance類中的一個靜態域進行存儲。Count.increment()和Count.value()都是synchronized的，用來控制對count域的訪問。increment()方法使用了Random對象，目的是在從把count讀取到temp中，到遞增temp並將其存儲回count的這段時間裡，有大約一半的時間產生讓步。如果你將increment()上的synchronized關鍵字注釋掉，那麼這個程式就會崩潰，因為多個任務將同時訪問並修改count（yield()會使問題更快地發生）。
每個Entrance任務都維護著一個本地值number,它包含通過某個特定入口進入的參觀者的數量。這提供了對count對象的雙重檢査，以確保其記錄的參觀者數量是正確的。Entrance.run()只是遞增number和count對象，然後休眠100毫秒。
因為Entrance.canceled是一個volatile布爾標誌，而它只會被讀取和賦值（不會與其他域組合在一起被讀取），所以不需要同步對其的訪問，就可以安全地操作它。如果你對諸如此類的情況有任何疑慮，那麼最好總是使用synchronized。這個程式在以穩定的方式關閉所有事物方面還有一些小麻煩，其部分原因是為了說明在終止多執行緒程式時你必須相當小心，而另一部分原因是為了演示interrupt()的值，稍後你將學習有關這個值的知識。
在3秒鐘之後，main()向Entrance發送static cancel()消息，然後調用exec對象的shutdown。方法，之後調用exec上的awaitTermination()方法。ExecutorService.awaitTermination()等待每個任務結束，如果所有的任務在超時時間達到之前全部結束，則返回true，否則返回false，表示不是所有的任務都已經結束了。儘管這會導致每個任務都退出其run()方法，並因此作為任務而終止，但是Entrance對象仍舊是有效的，因為在構造器中，每個Entrance對象都存儲在稱為entranceList的靜態List
當這個程式運行時，你將看到，在人們通過十字轉門時，將顯示總人數和通過每個入口的人數。如果移除Count.increment()上面的synchronized聲明，你將會注意到總人數與你的期望有差異，每個十字轉門統計的人數將與count中的值不同。只要用互斥來同步對Count的訪問，問題就可以解決了。請記住，Count.increment()通過使用temp和yield()，增加了失敗的可能性。在真正的執行緒問題中，失敗的可能性從統計學角度看可能非常小，因此你可能很容易就掉進了輕信所有事物都將正確工作的陷阱里。就像在上面的示例中，有些還未發生的問題就有可能會隱藏起來，因此在複審並發程式碼時，要格外地仔細。
2.在阻塞時終結
前面示例中的Entrance.run()在其循環中包含對sleep()的調用。我們知道，sleep()最終將喚醒，而任務也將返回循環的開始部分，去檢查canceled標誌，從而決定是否跳出循環。但是，sleep()一種情況，它使任務從執行狀態變為被阻塞狀態，而有時你必須終止被阻塞的任務。
執行緒狀態
一個執行緒可以處於以下四種狀態之一：
1）新建（new）∶當執行緒被創建時，它只會短暫地處於這種狀態。此時它已經分配了必需的系統資源，並執行了初始化。此刻執行緒已經有資格獲得CPU時間了，之後調度器將把這個執行緒轉變為可運行狀態或阻塞狀態。
2）就緒（Runnable）∶在這種狀態下，只要調度器把時間片分配給執行緒，執行緒就可以運行。也就是說，在任意時刻，執行緒可以運行也可以不運行。只要調度器能分配時間片給執行緒，它就可以運行；這不同於死亡和阻塞狀態。
3）阻塞（Blocked）∶執行緒能夠運行，但有某個條件阻止它的運行。當執行緒處於阻塞狀態時，調度器將忽略執行緒，不會分配給執行緒任何CPU時間。直到執行緒重新進入了就緒狀態，它才有可能執行操作。
4）死亡（Dead）∶處於死亡或終止狀態的執行緒將不再是可調度的，並且再也不會得到CPU時間，它的任務已結束，或不再是可運行的。任務死亡的通常方式是從run()方法返回，但是任務的執行緒還可以被中斷，你將要看到這一點。
進入阻塞狀態
一個任務進入阻塞狀態，可能有如下原因∶
1）通過調用sleep(milliseconds)使任務進入休眠狀態，在這種情況下，任務在指定的時間內不會運行。
2）你通過調用wait()使執行緒掛起。直到執行緒得到了notify()或notifyAll()消息（或者在Java SE5的java.util.concurrent類庫中等價的signal()或signalAll()消息），執行緒才會進入就緒狀態。我們將在稍後的小節中驗證這一點。
3）任務在等待某個輸入/輸出完成。
4）任務試圖在某個對象上調用其同步控制方法，但是對象鎖不可用，因為另一個任務已經獲取了這個鎖。
中斷
Thread類包含interrupt()方法，因此你可以終止被阻塞的任務，這個方法將設置執行緒的中斷狀態。如果一個執行緒已經被阻塞，或者試圖執行一個阻塞操作，那麼設置這個執行緒的中斷狀態將拋出InterruptedException。當拋出該異常或者該任務調用Thread.interrupted()時，中斷狀態將被複位。正如你將看到的，Thread.interrupted()提供了離開run()循環而不拋出異常的第二種方式。
為了調用interrupt()，你必須持有Thread對象。你可能已經注意到了，新的concurrent類庫似乎在避免對Thread對象的直接操作，轉而盡量通過Executor來執行所有操作。如果你在Executor上調用shutdownNow()，那麼它將發送一個interrupt()調用給它啟動的所有執行緒。這麼做是有意義的，因為當你完成工程中的某個部分或者整個程式時，通常會希望同時關閉某個特定Executor的所有任務。然而，你有時也會希望只中斷某個單一任務。如果使用Executor，那麼通過調用submit()而不是executor()來啟動任務，就可以持有該任務的上下文。submit()將返回一個泛型Future<?>，其中有一個未修飾的參數，因為你永遠都不會在其上調用get()——持有這種Future的關鍵在於你可以在其上調用cancel()，並因此可以使用它來中斷某個特定任務。如果你將true傳遞給cancel()，那麼它就會擁有在該執行緒上調用interrupt()以停止這個執行緒的許可權。因此，cancel()是一種中斷由Executor啟動的單個執行緒的方式。
package concurrency;
import java.io.IOException;
import java.io.InputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月05日 10:33
*
* 演示interrupt()的使用
*/
public class Interrupting {
private static ExecutorService exex = Executors.newCachedThreadPool();
static void test(Runnable r) throws InterruptedException {
Future<?> future = exex.submit(r);
TimeUnit.MILLISECONDS.sleep(100);
System.out.println("Interrupting " + r.getClass().getName());
future.cancel(true);
System.out.println("Interrupting sent to " + r.getClass().getName());
}
public static void main(String[] args) throws Exception {
test(new SleepBlocked());
test(new IOBlocked(System.in));
test(new SynchronizedBlocked());
TimeUnit.SECONDS.sleep(3);
System.out.println("Aborting with System.exit(0)");
System.exit(0);
}
}
class SleepBlocked implements Runnable {
@Override
public void run() {
try {
TimeUnit.SECONDS.sleep(100);
} catch (InterruptedException e) {
System.out.println("InterruptedException");
}
System.out.println("Exiting SleepBlocked.run()");
}
}
class IOBlocked implements Runnable {
private InputStream in;
public IOBlocked(InputStream in) {
this.in = in;
}
@Override
public void run() {
System.out.println("Waiting for read(): ");
try {
in.read();
} catch (IOException e) {
System.out.println("Interrupted from blocked I/O");
}
System.out.println("Exiting IOBlocked.run()");
}
}
class SynchronizedBlocked implements Runnable {
public synchronized void f() {
// 永不釋放鎖，一直佔有
while (true) {
// Thread.yield();
}
}
public SynchronizedBlocked() {
new Thread() {
@Override
public void run() {
f(); // 鎖會一直被當前執行緒佔用
}
}.start();
}
@Override
public void run() {
System.out.println("Trying to call f()");
f();
System.out.println("Exiting SynchronizedBlocked.run()");
}
} /* Output: (95% match)
Interrupting concurrency.SleepBlocked
InterruptedException
Exiting SleepBlocked.run()
Interrupting sent to concurrency.SleepBlocked
Waiting for read():
Interrupting concurrency.IOBlocked
Interrupting sent to concurrency.IOBlocked
Trying to call f()
Interrupting concurrency.SynchronizedBlocked
Interrupting sent to concurrency.SynchronizedBlocked
Aborting with System.exit(0)
*///:~
上面的每個任務都表示了一種不同類型的阻塞。SleepBlock是可中斷的阻塞示例，而IOBlocked和SynchronizedBlocked是不可中斷的阻塞示例。這個程式證明I/O和在synchronized 塊上的等待是不可中斷的，但是通過瀏覽程式碼，你也可以預見到這一點——無論是I/O還是嘗試調用synchronized方法，都不需要任何InterruptedException處理器。
前兩個類很簡單直觀∶在第一個類中run()方法調用了sleep()，而在第二個類中調用了read()。但是，為了演示SynchronizedBlock，我們必須首先獲取鎖。這是通過在構造器中創建匿名的Thread類的實例來實現的，這個匿名Thread類的對象通過調用f()獲取了對象鎖（這個執行緒必須有別於為SynchronizedBlock驅動run()的執行緒，因為一個執行緒可以多次獲得某個對象鎖）。由於f()永遠都不返回，因此這個鎖永遠不會釋放，而SynchronizedBlock.run()在試圖調用f()，並阻塞以等待這個鎖被釋放。
從輸出中可以看到，你能夠中斷對sleep()的調用（或者任何要求拋出InterruptedException 的調用）。但是，你不能中斷正在試圖獲取synchronized鎖或者試圖執行I/O操作的執行緒。這有點令人煩惱，特別是在創建執行I/O的任務時，因為這意味著I/O具有鎖住你的多執行緒程式的潛在可能。特別是對於基於Web的程式，這更是關乎利害。
對於這類問題，有一個略顯笨拙但是有時確實行之有效的解決方案，即關閉任務在其上發生阻塞的底層資源∶
package concurrency;
import java.io.InputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月05日 11:01
*
* 中斷I/O操作的執行緒：關閉任務在其上發生阻塞的底層資源
*/
public class CloseResource {
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
InputStream in = new Socket("localhost", 8080).getInputStream();
exec.execute(new IOBlocked(in));
exec.execute(new IOBlocked(System.in));
TimeUnit.MILLISECONDS.sleep(100);
System.out.println("Shutting down all threads");
exec.shutdownNow();
TimeUnit.SECONDS.sleep(1);
System.out.println("Closing " + in.getClass().getName());
in.close();
TimeUnit.SECONDS.sleep(1);
System.out.println("Closing " + System.in.getClass().getName());
System.in.close();
}
} /* Output: (85% match)
Waiting for read():
Waiting for read():
Shutting down all threads
Closing java.net.SocketInputStream
Interrupted from blocked I/O
Exiting IOBlocked.run()
Closing java.io.BufferedInputStream
Exiting IOBlocked.run()
*///:~
執行緒之間的協作
wait()和notifyAll()
有兩種形式的wait()。第一種版本接受毫秒數作為參數，含義與sleep()方法里參數的意思相同，都是指「在此期間暫停」。但是與sleep()不同的是，對於wait()而言∶
1）在wait()期間對象鎖是釋放的。而調用sleep()的時候鎖並沒有被釋放，調用yield()也屬於這種情況。
2）可以通過notify()、notifyAll()，或者令時間到期，從wait()中恢復執行。
第二種，也是更常用形式的wait()不接受任何參數。這種wait()將無限等待下去，直到執行緒接收到notify()或者notifyAll()消息。
wait()、notify()以及notifyAll()有一個比較特殊的方面，那就是這些方法是基類Object的一部分，而不是屬於Thread的一部分。儘管開始看起來有點奇怪——僅僅針對執行緒的功能卻作為通用基類的一部分而實現，不過這是有道理的，因為這些方法操作的鎖也是所有對象的一部分。所以，你可以把wait()放進任何同步控制方法里，而不用考慮這個類是繼承自Thread還是實現了Runnable介面。實際上，只能在同步控制方法或同步控制塊里調用wait()、notify()和notifyAll() （因為不用操作鎖，所以sleep()可以在非同步控制方法里調用）。如果在非同步控制方法里調用這些方法，程式能通過編譯，但運行的時候，將得到IlegalMonitorStateException異常，並伴隨著一些含糊的消息，比如「當前執行緒不是擁有者」。消息的意思是，調用wait()、notify()和notifyAll()的任務在調用這些方法前必須"擁有"（獲取）對象的鎖。
可以讓另一個對象執行某種操作以維護其自己的鎖。要這麼做的話，必須首先得到對象的鎖。比如，如果要向對象x發送notifyAll()，那麼就必須在能夠取得x的鎖的同步控制塊中這麼做∶
synchronized(x) {
x.notifyAll();
}
讓我們看一個簡單的示例，WaxOMatic.java有兩個過程∶一個是將蠟塗到Car上，一個是拋光它。拋光任務在塗蠟任務完成之前，是不能執行其工作的，而塗蠟任務在塗另一層蠟之前，必須等待拋光任務完成。WaxOn和WaxOff都使用了Car對象，該對象在這些任務等待條件變化的時候，使用wait()和notifyAll()來掛起和重新啟動這些任務∶
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月05日 15:25
*/
public class WaxOMatic {
public static void main(String[] args) throws Exception {
Car car = new Car();
ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new WaxOff(car));
exec.execute(new WaxOn(car));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow(); // 中斷所有任務
}
}
class Car {
private boolean waxOn = false;
// 塗蠟
public synchronized void waxed() {
waxOn = true; // 準備拋光
notifyAll();
}
// 拋光
public synchronized void buffed() {
waxOn = false; // 為塗另外一層蠟做準備
notifyAll();
}
// 等待塗蠟
public synchronized void waitForWaxing() throws InterruptedException {
while (!waxOn) {
wait();
}
}
// 等待拋光
public synchronized void waitForBuffing() throws InterruptedException {
while (waxOn) {
wait();
}
}
}
class WaxOn implements Runnable {
private Car car;
public WaxOn(Car car) {
this.car = car;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
System.out.print("Wax on! ");
TimeUnit.MILLISECONDS.sleep(200);
// 給車塗蠟
car.waxed();
// 塗蠟完成，準備拋光
car.waitForBuffing();
}
} catch (InterruptedException e) {
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax on task");
}
}
class WaxOff implements Runnable {
private Car car;
public WaxOff(Car car) {
this.car = car;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
car.waitForWaxing();
System.out.print("Wax off! ");
TimeUnit.MILLISECONDS.sleep(200);
car.buffed();
}
} catch (InterruptedException e) {
System.out.println("Exiting via interrupt");
}
System.out.println("Ending Wax off task");
}
} /* Output:
Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Wax off! Wax on! Exiting via interrupt
Ending Wax off task
Exiting via interrupt
Ending Wax on task
*///:~
這裡，Car有一個單一的布爾屬性waxOn，表示塗蠟-拋光處理的狀態。
在waitForWaxing()中將檢查waxOn標誌，如果它為false，那麼這個調用任務將通過調用wait()而被掛起。這個行為發生在synchronized方法中這一點很重要，因為在這樣的方法中，任務已經獲得了鎖。當你調用wait()時，執行緒被掛起，而鎖被釋放。鎖被釋放這一點是本質所在，因為為了安全地改變對象的狀態（例如，將waxOn改變為true，如果被掛起的任務要繼續執行，就必須執行該動作），其他某個任務就必須能夠獲得這個鎖。在本例中，如果另一個任務調用waxed()來表示「是時候該干點什麼了」，那麼就必須獲得這個鎖，從而將waxOn改變為true。之後，waxed()調用notifyAll()，這將喚醒在對wait()的調用中被掛起的任務。為了使該任務從wait()中喚醒，它必須首先重新獲得當它進入wait()時釋放的鎖。在這個鎖變得可用之前，這個任務是不會被喚醒的。
WaxOn.run()表示給汽車打蠟過程的第一個步驟，因此它將執行它的操作∶調用sleep()以模擬需要塗蠟的時間，然後告知汽車塗蠟結束，並調用waitForBuffing()，這個方法會用一個wait()調用來掛起這個任務，直至WaxOff任務調用這輛車的buffed()，.從而改變狀態並調用notifyAll() 為止。另一方面，WaxOff.run()立即進入waitForWaxing()，並因此而被掛起，直至WaxOn塗完蠟並且waxed()被調用。在運行這個程式時，你可以看到當控制權在兩個任務之間來回互相傳遞時，這個兩步驟過程在不斷地重複。在5秒鐘之後，interrupt()會中止這兩個執行緒；當你調用某個ExecutorService的shutdownNow()時，它會調用所有由它控制的執行緒的interrupt()。
notify()和notifyAll()
使用notify()而不是notifyAll()是一種優化。使用notify()時，在眾多等待同一個鎖的任務中只有一個會被喚醒，因此如果你希望使用notify()，就必須保證被喚醒的是恰當的任務。另外，為了使用notify()，所有任務必須等待相同的條件，因為如果你有多個任務在等待不同的條件，那麼你就不會知道是否喚醒了恰當的任務。如果使用notify()，當條件發生變化時，必須只有一個任務能夠從中受益。最後，這些限制對所有可能存在的子類都必須總是起作用的。如果這些規則中有任何一條不滿足，那麼你就必須使用notifyAll()而不是notify()。
在有關Java的執行緒機制的討論中，有一個令人困惑的描述∶notifyAll()將喚醒「所有正在等待的任務」。這是否意味著在程式中任何地方，任何處於wait()狀態中的任務都將被任何對notifyAll()的調用喚醒呢？在下面的示例中，與Task2相關的程式碼說明了情況並非如此——事實上，當notifyAll()因某個特定鎖而被調用時，只有等待這個鎖的任務才會被喚醒∶
package concurrency;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月05日 16:00
*
* notifyAll()並非喚醒所有「正在等待」的任務，而是只會喚醒等待 調用notifyAll()某個特定鎖的任務
*/
public class NotifyVsNotifyAll {
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
exec.execute(new Task());
}
exec.execute(new Task2());
Timer timer = new Timer();
timer.scheduleAtFixedRate(new TimerTask() {
boolean prod = true;
@Override
public void run() {
if (prod) {
System.out.print("\nnotify() ");
Task.blocker.prod();
prod = false;
} else {
System.out.print("\nnotifyAll() ");
Task.blocker.prodAll();
prod = true;
}
}
}, 400, 400);
TimeUnit.SECONDS.sleep(5);
timer.cancel();
System.out.println("\nTimer canceled");
TimeUnit.MILLISECONDS.sleep(500);
System.out.print("Task2.blocker.prodAll() ");
Task2.blocker.prodAll();
TimeUnit.MILLISECONDS.sleep(500);
System.out.println("\nShutting down");
exec.shutdownNow(); // 中斷所有任務
}
}
class Blocker {
synchronized void waitingCall() {
try {
while (!Thread.interrupted()) {
wait();
System.out.println(Thread.currentThread() + " ");
}
} catch (InterruptedException e) {
}
}
synchronized void prod() {
notify();
}
synchronized void prodAll() {
notifyAll();
}
}
class Task implements Runnable {
static Blocker blocker = new Blocker();
@Override
public void run() {
blocker.waitingCall();
}
}
class Task2 implements Runnable {
// 一個單獨的blocker對象
static Blocker blocker = new Blocker();
@Override
public void run() {
blocker.waitingCall();
}
} /* Output:
notify() Thread[pool-1-thread-1,5,main]
notifyAll() Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-2,5,main]
notify() Thread[pool-1-thread-1,5,main]
notifyAll() Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
notify() Thread[pool-1-thread-1,5,main]
notifyAll() Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-2,5,main]
notify() Thread[pool-1-thread-1,5,main]
notifyAll() Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
notify() Thread[pool-1-thread-1,5,main]
notifyAll() Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-5,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-2,5,main]
notify() Thread[pool-1-thread-1,5,main]
notifyAll() Thread[pool-1-thread-1,5,main]
Thread[pool-1-thread-2,5,main]
Thread[pool-1-thread-3,5,main]
Thread[pool-1-thread-4,5,main]
Thread[pool-1-thread-5,5,main]
Timer canceled
Task2.blocker.prodAll() Thread[pool-1-thread-6,5,main]
Shutting down
*///:~
Task和Task2每個都有其自己的Blocker對象，因此每個Task對象都會在Task.blocker上阻塞，而每個Task2都會在Task2.blocker上阻塞。在main()中，java.util.Timer對象被設置為每4/10秒執行一次run()方法，而這個run()方法將經由「激勵」方法交替地在Task.blocker上調用notify()和notifyAll()。
從輸出中你可以看到，即使存在Task2.blocker上阻塞的Task2對象，也沒有任何在Task.blocker上的notify()或notifyAll()調用會導致Task2對象被喚醒。與此類似，在main()的結尾，調用了timer的cancel()，即使計時器被撤銷了，前5個任務也依然在運行，並仍舊在它們對Task.blocker.waitingCall()的調用中被阻塞。對Task2.blocker.prodAll()的調用所產生的輸出不包括任何在Task.blocker中的鎖上等待的任務。
生產者和消費者
請考慮這樣一個飯店，它有一個廚師和一個服務員。這個服務員必須等待廚師準備好膳食。當廚師準備好時，他會通知服務員，之後服務員上菜，然後返回繼續等待。這是一個任務協作的示例∶廚師代表生產者，而服務員代表消費者。兩個任務必須在膳食被生產和消費時進行握手，而系統必須以有序的方式關閉。下面是對這個敘述建模的程式碼∶
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月05日 16:34
*
* 生產者消費者示例
*/
public class Restaurant {
Meal meal;
ExecutorService exec = Executors.newCachedThreadPool();
final WaitPerson waitPerson = new WaitPerson(this);
final Chef chef = new Chef(this);
public Restaurant() {
exec.execute(chef);
exec.execute(waitPerson);
}
public static void main(String[] args) {
new Restaurant();
}
}
class Meal {
private final int orderNum;
public Meal(int orderNum) {
this.orderNum = orderNum;
}
@Override
public String toString() {
return "Meal " + orderNum;
}
}
// 服務員 - 消費者
class WaitPerson implements Runnable {
private Restaurant restaurant;
public WaitPerson(Restaurant restaurant) {
this.restaurant = restaurant;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (this) {
while (restaurant.meal == null) {
wait(); // 等待廚師製作食物
}
}
System.out.println("Waitperson got " + restaurant.meal);
synchronized(restaurant.chef) {
restaurant.meal = null;
restaurant.chef.notifyAll(); // Ready for another
}
}
} catch (InterruptedException e) {
System.out.println("WaitPerson interrupted");
}
}
}
// 廚師 - 生產者
class Chef implements Runnable {
private final Restaurant restaurant;
private int count = 0;
public Chef(Restaurant restaurant) {
this.restaurant = restaurant;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (this) {
if (restaurant.meal != null) {
wait(); // 等待被消費
}
}
if (++count == 10) {
System.out.println("Out of food, closing");
restaurant.exec.shutdownNow();
}
System.out.print("order up! ");
synchronized (restaurant.waitPerson) {
restaurant.meal = new Meal(count);
restaurant.waitPerson.notifyAll();
}
TimeUnit.MILLISECONDS.sleep(100);
}
} catch (InterruptedException e) {
System.out.println("Chef interrupted");
}
}
} /* Output:
order up! Waitperson got Meal 1
order up! Waitperson got Meal 2
order up! Waitperson got Meal 3
order up! Waitperson got Meal 4
order up! Waitperson got Meal 5
order up! Waitperson got Meal 6
order up! Waitperson got Meal 7
order up! Waitperson got Meal 8
order up! Waitperson got Meal 9
Out of food, closing
WaitPerson interrupted
order up! Chef interrupted
*///:~
Restaurant是WaitPerson和Chef的焦點，他們都必須知道在為哪個Restaurant工作，因為他們必須和這家飯店的「餐窗」打交道，以便放置或拿取膳食restaurant.meal。在run()中，WaitPerson進入wait()模式，停止其任務，直至被Chef的notifyAll()喚醒。由於這是一個非常簡單的程式，因此我們知道只有一個任務將在WaitPerson的鎖上等待∶即WaitPerson任務自身。出於這個原因，理論上可以調用notify()而不是notifyAll()。但是，在更複雜的情況下，可能會有多個任務在某個特定對象鎖上等待，因此你不知道哪個任務應該被喚醒。因此，調用notifyAll() 要更安全一些，這樣可以喚醒等待這個鎖的所有任務，而每個任務都必須決定這個通知是否與自己相關。
一旦Chef送上Meal並通知WaitPerson，這個Chef就將等待，直至WaitPerson收集到訂單並通知Chef，之後Chef就可以燒下一份Meal了。
生產者-消費者與隊列
wait()和notifyAll()方法以一種非常低級的方式解決了任務互操作問題，即每次交互時都握手。在許多情況下，你可以瞄向更高的抽象級別，使用同步隊列來解決任務協作問題，同步隊列在任何時刻都只允許一個任務插入或移除元素。在java.util.concurrent.BlockingQueue介面中提供了這個隊列，這個介面有大量的標準實現。你通常可以使用LinkedBlockingQueue，它是一個無界隊列，還可以使用ArrayBlockingQueue，它具有固定的尺寸，因此你可以在它被阻塞之前，向其中放置有限數量的元素。
如果消費者任務試圖從隊列中獲取對象，而該隊列此時為空，那麼這些隊列還可以掛起消費者任務，並且當有更多的元素可用時恢復消費者任務。阻塞隊列可以解決非常大量的問題，而其方式與wait()和notifyAll()相比，則簡單並可靠得多。
考慮下面這個使用BlockingQueue的示例，有一台機器具有三個任務∶一個製作吐司、一個給吐司抹黃油，另一個在抹過黃油的吐司上塗果醬。我們可以通過各個處理過程之間的BlockingQueue來運行這個吐司製作程式∶
package concurrency.juc;
import container.Stack;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月06日 9:53
*
* 製作吐司 -> 塗黃油 -> 塗果醬 -> 咀嚼消費
*/
public class ToastOMatic {
public static void main(String[] args) throws Exception {
ToastQueue dryQueue = new ToastQueue(),
butterQueue = new ToastQueue(),
finishedQueue = new ToastQueue();
ExecutorService exec = Executors.newCachedThreadPool();
// 製作吐司
exec.execute(new Toaster(dryQueue));
// 為製作好的吐司塗上黃油
exec.execute(new Butterer(dryQueue, butterQueue));
// 為塗上黃油的吐司塗抹果醬
exec.execute(new Jammer(butterQueue, finishedQueue));
// 將塗抹好果醬的吐司給用戶咀嚼消費
exec.execute(new Eater(finishedQueue));
TimeUnit.SECONDS.sleep(5);
exec.shutdownNow();
}
}
class Toast {
public enum Status {DRY, BUFFERED, JAMMED}
private Status status = Status.DRY;
private final int id;
public Toast(int id) {
this.id = id;
}
// 塗抹黃油
public void butter() {
status = Status.BUFFERED;
}
// 塗果醬
public void jam() {
status = Status.JAMMED;
}
public Status getStatus() {
return status;
}
public int getId() {
return id;
}
@Override
public String toString() {
return "Toast " + id + ": " + status;
}
}
class ToastQueue extends LinkedBlockingQueue<Toast> { }
class Toaster implements Runnable {
private final ToastQueue toastQueue;
private int count = 0;
private final Random random = new Random(47);
public Toaster(ToastQueue toastQueue) {
this.toastQueue = toastQueue;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
TimeUnit.MILLISECONDS.sleep(100 + random.nextInt(500));
// 製作吐司
Toast t = new Toast(count++);
System.out.println(t);
// 插入吐司隊列
toastQueue.put(t);
}
} catch (InterruptedException e) {
System.out.println("Toaster Interrupted");
}
System.out.println("Toaster off");
}
}
// 給吐司塗黃油
class Butterer implements Runnable {
private final ToastQueue dryQueue, butteredQueue;
public Butterer(ToastQueue dry, ToastQueue buttered) {
dryQueue = dry;
butteredQueue = buttered;
}
@Override
public void run() {
try {
while(!Thread.interrupted()) {
// 若剛製作的吐司隊列為空，則阻塞等待
Toast t = dryQueue.take();
// 不為空，則為該吐司塗上黃油
t.butter();
System.out.println(t);
// 將塗好黃油的吐司放進黃油吐司隊列中
butteredQueue.put(t);
}
} catch(InterruptedException e) {
System.out.println("Butterer interrupted");
}
System.out.println("Butterer off");
}
}
// 給塗黃油過後的吐司塗上果醬
class Jammer implements Runnable {
private final ToastQueue butteredQueue, finishedQueue;
public Jammer(ToastQueue buttered, ToastQueue finished) {
butteredQueue = buttered;
finishedQueue = finished;
}
public void run() {
try {
while(!Thread.interrupted()) {
// 若已經塗黃油的吐司隊列為空，則阻塞等待
Toast t = butteredQueue.take();
// 獲取到塗抹黃油的吐司後，在塗上果醬
t.jam();
System.out.println(t);
// 放入已製作完成的土司隊列 中
finishedQueue.put(t);
}
} catch(InterruptedException e) {
System.out.println("Jammer interrupted");
}
System.out.println("Jammer off");
}
}
// 消費吐司
class Eater implements Runnable {
private final ToastQueue finishedQueue;
private int counter = 0;
public Eater(ToastQueue finished) {
finishedQueue = finished;
}
@Override
public void run() {
try {
while(!Thread.interrupted()) {
// 若已完成的吐司隊列為空，則阻塞等待
Toast t = finishedQueue.take();
// 驗證吐司是否按順序進行，所有吐司是否已經加了果醬
if(t.getId() != counter++ || t.getStatus() != Toast.Status.JAMMED) {
System.out.println(">>>> Error: " + t);
System.exit(1);
} else
System.out.println("咀嚼! " + t);
}
} catch(InterruptedException e) {
System.out.println("Eater interrupted");
}
System.out.println("Eater off");
}
}
Toast是一個使用enum值的優秀示例。注意，這個示例中沒有任何顯式的同步（即使用Lock 對象或synchronized關鍵字的同步），因為同步由隊列（其內部是同步的）和系統的設計隱式地管理了——每片Toast在任何時刻都只由一個任務在操作。因為隊列的阻塞，使得處理過程將被自動地掛起和恢復。你可以看到由BlockingQueue產生的簡化十分明顯。在使用顯式的wait()和notifyAll()時存在的類和類之間的耦合被消除了，因為每個類都只和它的BlockingQuene通訊。
死鎖
我們已經知道，任務可以變成阻塞狀態，所以就可能出現這種情況：某個任務在等待另一個任務，而後者又在等待別的任務，這樣一直下去，直到這個鏈條上的任務又在等待第一個任務釋放鎖。這得到了一個任務之間相互等待的連續循環，沒有哪個執行緒能繼續，這被稱之為死鎖。
一個經典的死鎖例證就是哲學將就餐問題：給問題的基本描述是指定五個哲學家，這些哲學將花部分時間思考，花部分時間就餐。當他們思考的時候，不需要任何共享資源；當他們就餐的時候，將使用有限數量的餐具。問題中引入的難點是：作為哲學家，他們很窮，他們每人只買得起一根筷子。他們圍坐在桌子周圍，每人之間放一根筷子。當一個哲學家要就餐的時候，這個哲學家就必須同時得到左邊和右邊的筷子，如果一個哲學家左邊或右邊已經有人在使用筷子了，那麼這個哲學家就必須等待，直至可得到必需的筷子。
public class Chopstick {
// 筷子是否被拿走
private boolean taken = false;
// 拿筷子
public synchronized void take() throws InterruptedException {
while (taken) {
wait();
}
taken = true;
}
// 歸還筷子
public synchronized void drop() {
taken = false;
notifyAll();
}
}
任何兩個Philosopher(哲學家)都不能成功take()同一根筷子。另外，如果一根Chopstick(筷子)已經被某個Philosopher獲得，那麼另一個Philosopher可以wait()，直至這根Chopstick的當前持有者調用drop()使其可用為止。
當一個Philosopher任務調用take()時，這個Philosopher將等待，直至taken標誌變為false（直至當前持有Chopstick的Philosopher釋放它）。然後這個任務會將taken標誌設置為true，以表示現在由新的Philosopher持有這根Chopstick。當這個Philosopher使用完這根Chopstick時，它會調用drop()來修改標誌的狀態，並notifyAll()所有其他的Philosopher，這些Philosopher中有些可能就在wait()這根Chopstick。
public class Philosopher implements Runnable {
// 左邊的筷子
private final Chopstick left;
// 右邊的筷子
private final Chopstick right;
private final int id;
// 權重因子
private final int ponderFactor;
private Random random = new Random(47);
public Philosopher(Chopstick left, Chopstick right, int id, int ponderFactor) {
this.left = left;
this.right = right;
this.id = id;
this.ponderFactor = ponderFactor;
}
private void pause() throws InterruptedException {
if (ponderFactor == 0) {
return;
}
TimeUnit.MILLISECONDS.sleep(random.nextInt(ponderFactor * 250));
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
System.out.println(this + " thinking");
pause();
// 哲學家餓了
System.out.println(this + " grabbing right");
right.take();
System.out.println(this + " grabbing left");
left.take();
System.out.println(this + " eating");
pause();
right.drop();
left.drop();
}
} catch (InterruptedException e) {
System.out.println(this + " exiting via interrupt");
}
}
@Override
public String toString() {
return "Philosopher " + id;
}
}
在Philosopher.run()中，每個Philosopher只是不斷地思考和吃飯。如果PonderFactor不為0，則pause()方法會休眠（sleeps）一段隨機的時間。通過使用這種方式，你將看到Philosopher會在思考上花掉一段隨機化的時間，然後嘗試著獲取（take())右邊和左邊的Chopstick，隨後在吃飯上再花掉一段隨機化的時間，之後重複此過程。
現在我們可以建立這個程式的將會產生死鎖的版本了∶
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月06日 11:28
*
* 死鎖示例: 哲學家進餐問題
*
* {args: 0 5 timeout}
*/
public class DeadLockDiningPhilosopher {
public static void main(String[] args) throws Exception {
int ponder = 5;
if (args.length > 0 ) {
ponder = Integer.parseInt(args[0]);
}
int size = 5;
if (args.length > 1) {
size = Integer.parseInt(args[1]);
}
ExecutorService exec = Executors.newCachedThreadPool();
Chopstick[] sticks = new Chopstick[size];
for (int i = 0; i < size; i++) {
sticks[i] = new Chopstick();
}
for (int i = 0; i < size; i++) {
exec.execute(new Philosopher(sticks[i], sticks[(i+1) % size], i, ponder));
}
if (args.length == 3 && args[2].equals("timeout")) {
TimeUnit.SECONDS.sleep(10);
} else {
System.out.println("Press 'Enter' to quit");
System.in.read();
}
exec.shutdownNow();
}
}
你會發現，如果Philosopher花在思考上的時間非常少，那麼當他們想要進餐時，全都會在Chopstick上產生競爭，而死鎖也就會更快地發生。
第一個命令行參數可以調整ponder因子，從而影響每個Philosopher花費在思考上的時間長度。如果有許多Philosopher，或者他們花費很多時間去思考，那麼儘管存在死鎖的可能，但你可能永遠也看不到死鎖。值為0的命令行參數傾向於使死鎖儘快發生。
注意，Chopstick對象不需要內部標識符，它們是由在數組sticks中的位置來標識的。每個Philosopher構造器都會得到一個對左邊和右邊Chopstick對象的引用。除了最後一個Philosopher，其他所有的Philosopher都是通過將這個Philosopher定位於下一對Chopstick對象之間而被初始化的，而最後一個Philosopher右邊的Chopstick是第0個Chopstick，這樣這個循環表也就結束了。因為最後一個Philosopher坐在第一個Philosopher的右邊，所以他們會共享第0個Chopstick。現在，所有的Philosopher都有可能希望進餐，從而等待其臨近的Philosopher放下它們的Chopstick。這將使程式死鎖。
如果Philosopher花費更多的時間去思考而不是進餐（使用非0的ponder值，或者大量的Philosopher），那麼他們請求共享資源（Chopstick）的可能性就會小許多，這樣你就會確信該程式不會死鎖，儘管它們並非如此。這個示例相當有趣，因為它演示了看起來可以正確運行，但實際上會死鎖的程式。
要修正死鎖問題，你必須明白，當以下四個條件同時滿足時，就會發生死鎖∶
1）互斥條件。任務使用的資源中至少有一個是不能共享的。這裡，一根Chopstick一次就只能被一個Philosopher使用。
2）至少有一個任務它必須持有一個資源且正在等待獲取一個當前被別的任務持有的資源。也就是說，要發生死鎖，Philosopher必須拿著一根Chopstick並且等待另一根。
3）資源不能被任務搶佔，任務必須把資源釋放當作普通事件。Philosopher很有禮貌，他們不會從其他Philosopher那裡搶Chopstick。
4）必須有循環等待，這時，一個任務等待其他任務所持有的資源，後者又在等待另一個任務所持有的資源，這樣一直下去，直到有一個任務在等待第一個任務所持有的資源，使得大家都被鎖住。在DeadlockDiningPhilosophers.java中，因為每個Philosopher都試圖先得到右邊的Chopstick，然後得到左邊的Chopstick，所以發生了循環等待。
因為要發生死鎖的話，所有這些條件必須全部滿足；所以要防止死鎖的話，只需破壞其中一個即可。在程式中，防止死鎖最容易的方法是破壞第4個條件。有這個條件的原因是每個Philosopher都試圖用特定的順序拿Chopstick∶先右後左。正因為如此，就可能會發生「每個人都拿著右邊的Chopstick，並等待左邊的Chopstick」的情況，這就是循環等待條件。然而，如果最後一個Philosopher被初始化成先拿左邊的Chopstick，後拿右邊的Chopstick，那麼這個Philosopher將永遠不會阻止其右邊的Philosopher拿起他們的Chopstick。在本例中，這就可以防止循環等待。這只是問題的解決方法之一，也可以通過破壞其他條件來防止死鎖（具體細節請參考更高級的討論執行緒的書籍）∶
package concurrency;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月06日 11:52
*
* 解決哲學家就餐的死鎖問題：通過修改循環等待條件，即修改最後一個哲學家先拿走左手邊的筷子，再拿走右手邊的筷子
* {args: 5 5 timeout}
*/
public class FixedDiningPhilosopher {
public static void main(String[] args) throws Exception {
int ponder = 5;
if (args.length > 0 ) {
ponder = Integer.parseInt(args[0]);
}
int size = 5;
if (args.length > 1) {
size = Integer.parseInt(args[1]);
}
ExecutorService exec = Executors.newCachedThreadPool();
Chopstick[] sticks = new Chopstick[size];
for (int i = 0; i < size; i++) {
sticks[i] = new Chopstick();
}
for (int i = 0; i < size; i++) {
// 若不是最後一個哲學家，則都先拿走右手邊的筷子
if (i < (size - 1)) {
exec.execute(new Philosopher(sticks[i], sticks[(i+1)], i, ponder));
} else {
// 最後一個哲學家先拿起左手邊的筷子
exec.execute(new Philosopher(sticks[0], sticks[(i)], i, ponder));
}
}
if (args.length == 3 && args[2].equals("timeout")) {
TimeUnit.SECONDS.sleep(10);
} else {
System.out.println("Press 'Enter' to quit");
System.in.read();
}
exec.shutdownNow();
}
}
通過確保最後一個Philosopher先拿起和放下左邊的Chopstick，我們可以移除死鎖，從而使這個程式平滑地運行。
新類庫中的構件
Java SE5的java.util.concurrent引入了大量設計用來解決並發問題的新類，學習使用它們將有助於你編寫出更加簡單而健壯的並發程式。
CountDownLatch
它被用來同步一個或多個任務，強制它們等待由其他任務執行的一組操作完成。
你可以向CountDownLatch對象設置一個初始計數值，任何在這個對象上調用wait()的方法都將阻塞，直至這個計數值到達0。其他任務在結束其工作時，可以在該對象上調用countDown()來減小這個計數值。CountDownLatch被設計為只觸發一次，計數值不能被重置。如果你需要能夠重置計數值的版本，則可以使用CyclicBarrier。
調用countDown()的任務在產生這個調用時並沒有被阻塞，只有對await()的調用會被阻塞，直至計數值到達0。
CountDownLatch的典型用法是將一個程式分為n個互相獨立的可解決任務，並創建值為0的CountDownLatch。當每個任務完成時，都會在這個鎖存器上調用countDown()。等待問題被解決的任務在這個鎖存器上調用await()，將它們自己攔住，直至鎖存器計數結束。下面是演示這種技術的一個框架示例∶
package concurrency;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月06日 15:02
*
* 倒計時鎖存器 使用示例
*/
public class CountDownLatchDemo {
static final int SIZE = 100;
public static void main(String[] args) throws Exception {
ExecutorService exec = Executors.newCachedThreadPool();
// 所有任務都必須共享一個CountDownLatch對象
CountDownLatch latch = new CountDownLatch(SIZE);
for (int i = 0; i < 10; i++) {
exec.execute(new WaitingTask(latch));
}
for (int i = 0; i < SIZE; i++) {
exec.execute(new TaskPortion(latch));
}
System.out.println("latch all tasks");
// 當所有任務都完成的時候退出程式
exec.shutdown();
}
}
class TaskPortion implements Runnable {
private static int counter = 0;
private final int id = counter++;
private static Random random = new Random(47);
private final CountDownLatch latch;
public TaskPortion(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void run() {
try {
doWork();
latch.countDown();
} catch (InterruptedException e) {
}
}
public void doWork() throws InterruptedException {
TimeUnit.MILLISECONDS.sleep(random.nextInt(2000));
System.out.println(this + "completed");
}
@Override
public String toString() {
return String.format("%1$-3d", id);
}
}
class WaitingTask implements Runnable {
private static int counter = 0;
private final int id = counter++;
private final CountDownLatch latch;
public WaitingTask(CountDownLatch latch) {
this.latch = latch;
}
@Override
public void run() {
try {
latch.await();
System.out.println("Latch barrier passed for " + this);
} catch (InterruptedException e) {
System.out.println(this + " interrupted");
}
}
@Override
public String toString() {
return String.format("WaitingTask %1$-3d", id);
}
}
TaskPortion將隨機地休眠一段時間，以模擬這部分工作的完成，而WaitingTask表示系統中必須等待的部分，它要等待問題的初始化部分完成為止。所有任務都是用在main()中定義的同一個單一的CountDownLatch。
CyclicBarrier
CyclicBarrier適用於這樣的情況∶你希望創建一組任務，它們並行地執行工作，然後在進行下一個步驟之前等待，直至所有任務都完成（看起來有些像join0）。它使得所有的並行任務都將在柵欄處列隊，因此可以一致地向前移動。這非常像CountDownLatch，只是CountDownLatch是只觸發一次的事件，而CyclicBarrier可以多次重用。
下面以賽馬遊戲為示例，演示CyclicBarrier的使用：
package concurrency;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
/**
* @author Mr.Sun
* @date 2022年09月06日 15:22
*
* 演示CyclicBarrier的使用
*/
public class HorseRace {
static final int FINISH_LINE = 75;
private List<Horse> horses = new ArrayList<>();
private ExecutorService exec = Executors.newCachedThreadPool();
private CyclicBarrier barrier;
public HorseRace(int nhorses, final int pause) {
barrier = new CyclicBarrier(nhorses, new Runnable() {
@Override
public void run() {
StringBuilder s = new StringBuilder();
for (int i = 0; i < FINISH_LINE; i++) {
s.append("="); // 跑道上的柵欄
}
System.out.println(s);
for (Horse horse : horses) {
System.out.println(horse.tracks());
}
for (Horse horse : horses) {
if (horse.getStrides() >= FINISH_LINE) {
System.out.println(horse + "won!");
exec.shutdownNow();
return;
}
try {
TimeUnit.MILLISECONDS.sleep(pause);
} catch (InterruptedException e) {
System.out.println("barrier-action sleep interrupted");
}
}
}
});
for (int i = 0; i < nhorses; i++) {
Horse horse = new Horse(barrier);
horses.add(horse);
exec.execute(horse);
}
}
public static void main(String[] args) {
int nHorses = 7;
int pause = 200;
if (args.length > 0) {
int n = new Integer(args[0]);
nHorses = n > 0 ? n : nHorses;
}
if (args.length > 1) {
int p = new Integer(args[1]);
pause = p > -1 ? p : pause;
}
new HorseRace(nHorses, pause);
}
}
class Horse implements Runnable {
private static int counter = 0;
private final int id = counter++;
private int strides = 0;
private Random random = new Random(47);
private CyclicBarrier barrier;
public Horse(CyclicBarrier barrier) {
this.barrier = barrier;
}
@Override
public void run() {
try {
while (!Thread.interrupted()) {
synchronized (this) {
strides += random.nextInt(3);
}
barrier.await();
}
} catch (InterruptedException e) {
} catch (BrokenBarrierException e) {
throw new RuntimeException(e);
}
}
public synchronized int getStrides() {
return strides;
}
@Override
public String toString() {
return "Horse " + id + " ";
}
public String tracks() {
StringBuilder s = new StringBuilder();
for (int i = 0; i < getStrides(); i++) {
s.append("*");
}
s.append(id);
return s.toString();
}
}
可以向CyclicBarrier提供一個"柵欄動作"，它是一個Runnable，當計數值到達0時自動執行————這是CyclicBarrier和CountDownLatch之間的另一個區別。這裡，柵欄動作是作為匿名內部類創建的，它被提交給了CyclicBarrier的構造器。
我試圖讓每匹馬都列印自己，但是之後的顯示順序取決於任務管理器。CyclicBarrier使得每匹馬都要執行為了向前移動所必需執行的所有工作，然後必須在柵欄處等待其他所有的馬都準備完畢。當所有的馬都向前移動時，CyclicBarrier將自動調用Runnable柵欄動作任務，按順序顯示馬和終點線的位置。
一旦所有的任務都越過了柵欄，它就會自動地為下一回合比賽做好準備。
為了展示這個非常簡單的動畫效果，你需要將控制台視窗的尺寸調整為小到只有馬時，才會展示出來。
DelayQueue
這是一個無界的BlockingQueue，用於放置實現了Delayed介面的對象，其中的對象只能在其到期時才能從隊中取走。這種隊列是有序的，即隊頭對象的延遲到期的時間最長。如果沒有任何延遲到期，那麼就不會有任何頭元素，並且poll（）將返回mull（正因為這樣，你不能將mull放置到這種隊列中）。
下面是一個示例，其中的Delayed對象自身就是任務，而DelayedTaskConsumer將最「緊急」的任務（到期時間最長的任務）從隊列中取出，然後運行它。注意，這樣DelayQueue就成為了優先順序隊列的一種變體∶
package concurrency.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
/**
* @author Mr.Sun
* @date 2022年09月06日 16:08
*
* 延遲隊列演示
*/
public class DelayQueueDemo {
public static void main(String[] args) {
Random rand = new Random(47);
ExecutorService exec = Executors.newCachedThreadPool();
DelayQueue<DelayedTask> queue = new DelayQueue<>();
// Fill with tasks that have random delays:
for(int i = 0; i < 20; i++) {
queue.put(new DelayedTask(rand.nextInt(5000)));
}
// Set the stopping point
queue.add(new DelayedTask.EndSentinel(5000, exec));
exec.execute(new DelayedTaskConsumer(queue));
}
}
class DelayedTask implements Runnable, Delayed {
private static int counter = 0;
private final int id = counter++;
private final int delta;
private final long trigger;
protected static List<DelayedTask> sequence = new ArrayList<>();
public DelayedTask(int delayMillisecond) {
this.delta = delayMillisecond;
this.trigger = System.nanoTime() + NANOSECONDS.convert(delta, MILLISECONDS);
sequence.add(this);
}
@Override
public int compareTo(Delayed arg) {
DelayedTask that = (DelayedTask) arg;
if (trigger < that.trigger) {
return -1;
}
if (trigger > that.trigger) {
return 1;
}
return 0;
}
@Override
public void run() {
System.out.print(this + " ");
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(trigger - System.nanoTime(), NANOSECONDS);
}
@Override
public String toString() {
return String.format("[%1$-4d]", delta) + " Task " + id;
}
public String summary() {
return "(" + id + ":" + delta + ")";
}
public static class EndSentinel extends DelayedTask {
private ExecutorService exec;
public EndSentinel(int delay, ExecutorService e) {
super(delay);
exec = e;
}
@Override
public void run() {
for(DelayedTask pt : sequence) {
System.out.print(pt.summary() + " ");
}
System.out.println();
System.out.println(this + " Calling shutdownNow()");
exec.shutdownNow();
}
}
}
class DelayedTaskConsumer implements Runnable {
private DelayQueue<DelayedTask> q;
public DelayedTaskConsumer(DelayQueue<DelayedTask> q) {
this.q = q;
}
@Override
public void run() {
try {
while(!Thread.interrupted())
q.take().run(); // Run task with the current thread
} catch(InterruptedException e) {
// Acceptable way to exit
}
System.out.println("Finished DelayedTaskConsumer");
}
} /* Output:
[128 ] Task 11 [200 ] Task 7 [429 ] Task 5 [520 ] Task 18 [555 ] Task 1 [961 ] Task 4 [998 ] Task 16 [1207] Task 9 [1693] Task 2 [1809] Task 14 [1861] Task 3 [2278] Task 15 [3288] Task 10 [3551] Task 12 [4258] Task 0 [4258] Task 19 [4522] Task 8 [4589] Task 13 [4861] Task 17 [4868] Task 6 (0:4258) (1:555) (2:1693) (3:1861) (4:961) (5:429) (6:4868) (7:200) (8:4522) (9:1207) (10:3288) (11:128) (12:3551) (13:4589) (14:1809) (15:2278) (16:998) (17:4861) (18:520) (19:4258) (20:5000)
[5000] Task 20 Calling shutdownNow()
Finished DelayedTaskConsumer
*///:~
DelayedTask包含一個稱為sequence的List
Delayed介面有一個方法名為getDelay()，它可以用來告知延遲到期有多長時間，或者延遲在多長時間之前已經到期。這個方法將強制我們去使用TimeUnit類，因為這就是參數類型。這會產生一個非常方便的類，因為你可以很容易地轉換單位而無需作任何聲明。例如，delta的值是以毫秒為單位存儲的，但是Java SE5的方法System.nanoTime()產生的時間則是以納秒為單位的。你可以轉換delta的值，方法是聲明它的單位以及你希望以什麼單位來表示，就像下面這樣∶
NANOSECONDS.convert(delta, MILLISECONDS);
在getDelay()中，希望使用的單位是作為unit參數傳遞進來的，你使用它將當前時間與觸發時間之間的差轉換為調用者要求的單位，而無需知道這些單位是什麼（這是策略設計模式的一個簡單示例，在這種模式中，演算法的一部分是作為參數傳遞進來的）。
為了排序，Delayed介面還繼承了Comparable介面，因此必須實現compareTo()，使其可以產生合理的比較。toString()和summary()提供了輸出格式化，而嵌套的EndSentinel類提供了一種關閉所有事物的途徑，具體做法是將其放置為隊列的最後一個元素。
注意，因為DelayedTaskConsumer自身是一個任務，所以它有自己的Thread，它可以使用這個執行緒來運行從隊中獲取的所有任務。由於任務是按照隊列優先順序的順序執行的，因此在本例中不需要啟動任何單獨的執行緒來運行DelayedTask。
從輸出中可以看到，任務創建的順序對執行順序沒有任何影響，任務是按照所期望的延遲順序執行的。
PriorityBlockingQueue
這是一個很基礎的優先順序隊列，它具有可阻塞的讀取操作。下面是一個示例，其中在優先順序隊列中的對象是按照優先順序順序從隊列中出現的任務。PrioritizedTask被賦予了一個優先順序數字，以此來提供這種順序∶
package concurrency.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月06日 16:35
*
* 演示優先順序阻塞隊列
*/
public class PriorityBlockingQueueDemo {
public static void main(String[] args) {
Random rand = new Random(47);
ExecutorService exec = Executors.newCachedThreadPool();
PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<>();
exec.execute(new PrioritizedTaskProducer(queue, exec));
exec.execute(new PrioritizedTaskConsumer(queue));
}
}
class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
private Random random = new Random(47);
private static int counter = 0;
private final int id = counter++;
private final int priority;
protected static List<PrioritizedTask> sequence = new ArrayList<>();
public PrioritizedTask(int priority) {
this.priority = priority;
sequence.add(this);
}
@Override
public int compareTo(PrioritizedTask arg) {
return priority < arg.priority ? 1 : (priority > arg.priority ? -1 :0);
}
@Override
public void run() {
try {
TimeUnit.MILLISECONDS.sleep(random.nextInt(250));
} catch(InterruptedException e) {
// Acceptable way to exit
}
System.out.println(this);
}
@Override
public String toString() {
return String.format("[%1$-3d]", priority) + " Task " + id;
}
public String summary() {
return "(" + id + ":" + priority + ")";
}
public static class EndSentinel extends PrioritizedTask {
private ExecutorService exec;
public EndSentinel(ExecutorService e) {
super(-1); // 此程式中的最低優先順序
exec = e;
}
@Override
public void run() {
int count = 0;
for(PrioritizedTask pt : sequence) {
System.out.print(pt.summary());
if(++count % 5 == 0) {
System.out.println();
}
}
System.out.println();
System.out.println(this + " Calling shutdownNow()");
exec.shutdownNow();
}
}
}
class PrioritizedTaskProducer implements Runnable {
private Random rand = new Random(47);
private Queue<Runnable> queue;
private ExecutorService exec;
public PrioritizedTaskProducer(Queue<Runnable> q, ExecutorService e) {
queue = q;
exec = e; // Used for EndSentinel
}
@Override
public void run() {
// 無界隊列，不會被阻塞
// 用隨機優先順序快速填充
for(int i = 0; i < 20; i++) {
queue.add(new PrioritizedTask(rand.nextInt(10)));
Thread.yield();
}
// 最高優先順序工作的任務
try {
for(int i = 0; i < 10; i++) {
TimeUnit.MILLISECONDS.sleep(250);
queue.add(new PrioritizedTask(10));
}
// 添加作業時，優先順序最低的優先：
for(int i = 0; i < 10; i++) {
queue.add(new PrioritizedTask(i));
}
// 停止所有任務的哨兵:
queue.add(new PrioritizedTask.EndSentinel(exec));
} catch(InterruptedException e) {
// Acceptable way to exit
}
System.out.println("Finished PrioritizedTaskProducer");
}
}
class PrioritizedTaskConsumer implements Runnable {
private PriorityBlockingQueue<Runnable> q;
public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> q) {
this.q = q;
}
@Override
public void run() {
try {
while(!Thread.interrupted())
// 使用當前執行緒運行任務
q.take().run();
} catch(InterruptedException e) {
// Acceptable way to exit
}
System.out.println("Finished PrioritizedTaskConsumer");
}
}
與前一個示例相同，PrioritizedTask對象的創建序列被記錄在sequence List中，用於和實際的執行順序比較。run()方法將休眠一小段隨機的時間，然後列印對象資訊，而EndSentinel提供了和前面相同的功能，要確保它是隊列中最後一個對象。
PrioritizedTaskProducer和PrioritizedTaskComsumer通過PriorityBlockinQueue彼此連接。因為這種隊列的阻塞特性提供了所有必需的同步，所以你應該注意到了，這裡不需要任何顯式的同步——不必考慮當你從這種隊列中讀取時，其中是否有元素，因為這個隊列在沒有元素時，將直接阻塞讀取者。
Semaphore
正常的鎖（來自concurrent.locks或內建的synchronized鎖）在任何時刻都只允許一個任務訪問一項資源，而計數訊號量允許n個任務同時訪問這個資源。你還可以將訊號量看作是在向外分發使用資源的「許可證」，儘管實際上沒有使用任何許可證對象。
作為一個示例，請考慮對象池的概念，它管理著數量有限的對象，當要使用對象時可以簽出它們，而在用戶使用完畢時，可以將它們簽回。這種功能可以被封裝到一個泛型類中∶
package concurrency.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;
/**
* @author Mr.Sun
* @date 2022年09月06日 17:13
*
* 演示 Semaphore
*/
public class Pool<T> {
private int size;
private List<T> items = new ArrayList<>();
private volatile boolean[] checkedOut;
private Semaphore available;
public Pool(Class<T> classObject, int size) {
this.size = size;
checkedOut = new boolean[size];
// 載入具有可檢出對象的池
available = new Semaphore(size, true);
for (int i = 0; i < size; i++) {
try {
// Assumes a default constructor:
items.add(classObject.newInstance());
} catch(Exception e) {
throw new RuntimeException(e);
}
}
}
public T checkOut() throws InterruptedException {
available.acquire();
return getItem();
}
public void checkIn(T x) {
if(releaseItem(x)) {
available.release();
}
}
private synchronized T getItem() {
for(int i = 0; i < size; ++i)
if(!checkedOut[i]) {
checkedOut[i] = true;
return items.get(i);
}
return null; // 訊號量阻止到達這裡
}
private synchronized boolean releaseItem(T item) {
int index = items.indexOf(item);
if(index == -1) {
return false; // Not in the list
}
if(checkedOut[index]) {
checkedOut[index] = false;
return true;
}
return false; // Wasn't checked out
}
}
在這個簡化的形式中，構造器使用newInstance0來把對象載入到池中。如果你需要一個新對象，那麼可以調用checkOut()，並且在使用完之後，將其遞交給checkIn()。
boolean類型的數組checkedOut可以跟蹤被簽出的對象，並且可以通過getItem()和releaseItem()方法來管理。而這些都將由Semaphore類型的available來加以確保，因此，在checkOut()中，如果沒有任何訊號量許可證可用（這意味著在池中沒有更多的對象了），available 將阻塞調用過程。在checkIn()中，如果被簽入的對象有效，則會向訊號量返回一個許可證。
為了創建一個示例，我們可以使用Fat，這是一種創建代價高昂的對象類型，因為它的構造器運行起來很耗時∶
package concurrency.juc;
/**
* @author Mr.Sun
* @date 2022年09月06日 17:22
*/
public class Fat {
private volatile double d; // 防止優化
private static int counter = 0;
private final int id = counter++;
public Fat() {
// 耗時，可中斷的操作
for(int i = 1; i < 10000; i++) {
d += (Math.PI + Math.E) / (double)i;
}
}
public void operation() {
System.out.println(this);
}
@Override
public String toString() {
return "Fat id: " + id;
}
}
我們在池中管理這些對象，以限制這個構造器所造成的影響。我們可以創建一個任務，它將簽出Fat對象，持有一段時間之後再將它們簽入，以此來測試Pool這個類∶
package concurrency.juc;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
/**
* @author Mr.Sun
* @date 2022年09月06日 17:25
*/
public class SemaphoreDemo {
final static int SIZE = 25;
public static void main(String[] args) throws Exception {
final Pool<Fat> pool = new Pool<>(Fat.class, SIZE);
ExecutorService exec = Executors.newCachedThreadPool();
for(int i = 0; i < SIZE; i++) {
exec.execute(new CheckoutTask<>(pool));
}
System.out.println("All CheckoutTasks created");
List<Fat> list = new ArrayList<>();
for(int i = 0; i < SIZE; i++) {
Fat f = pool.checkOut();
System.out.print(i + ": main() thread checked out ");
f.operation();
list.add(f);
}
Future<?> blocked = exec.submit(new Runnable() {
@Override
public void run() {
try {
// Semaphore prevents additional checkout,
// so call is blocked:
pool.checkOut();
} catch(InterruptedException e) {
System.out.println("checkOut() Interrupted");
}
}
});
TimeUnit.SECONDS.sleep(2);
blocked.cancel(true); // Break out of blocked call
System.out.println("Checking in objects in " + list);
for(Fat f : list) {
pool.checkIn(f);
}
for(Fat f : list) {
pool.checkIn(f); // Second checkIn ignored
}
exec.shutdown();
}
}
// 從池中檢出資源的任務:
class CheckoutTask<T> implements Runnable {
private static int counter = 0;
private final int id = counter++;
private Pool<T> pool;
public CheckoutTask(Pool<T> pool) {
this.pool = pool;
}
@Override
public void run() {
try {
T item = pool.checkOut();
System.out.println(this + "checked out " + item);
TimeUnit.SECONDS.sleep(1);
System.out.println(this +"checking in " + item);
pool.checkIn(item);
} catch(InterruptedException e) {
// Acceptable way to terminate
}
}
@Override
public String toString() {
return "CheckoutTask " + id + " ";
}
}
在main()中，創建了一個持有Fat對象的Pool，而一組CheckoutTask則開始操練這個Pool。然後，main()執行緒簽出池中的Fat對象，但是並不簽入它們。一旦池中所有的對象都被簽出，Semaphore將不再允許執行任何簽出操作。blocked的run()方法因此會被阻塞，2秒鐘之後，cancel()方法被調用，以此來掙脫Future的束縛。注意，冗餘的簽入將被Pool忽略。
Exchanger
Exchanger是在兩個任務之間交換對象的柵欄。當這些任務進入柵欄時，它們各自擁有一個對象，當它們離開時，它們都擁有之前由對象持有的對象。Exchanger的典型應用場景是∶一個任務在創建對象，這些對象的生產代價很高昂，而另一個任務在消費這些對象。通過這種方式，可以有更多的對象在被創建的同時被消費。
