think-in-java 并發前半部分（并發1）參見：??https://blog.csdn.net/PacosonSWJTU/article/details/104855730
【21.4.3】中斷
1、Thread類包含 interrupt方法，可以終止被阻塞的任務。這個方法將設置線程的中斷狀態。 如果一個線程被阻塞，或者視圖執行一個阻塞操作，那么設置這個線程的中斷狀態將拋出 InterruptedException異常。當拋出該異常或該任務調用了 Thread.interrupted() 方法時， 中斷狀態將被復位，設置為true；?
2、如果調用 Executor上調用 shutdownNow方法，那么它將發送一個 interrupte方法調用給他啟動的所有線程。
通過調用submit()而不是executor() 來啟動任務，就可以持有該任務的上下文。submit()方法將返回一個泛型Future<?>， 持有這個Future的關鍵在于
/*** 中斷由線程池管理的某個線程 */
public class Interrupting {private static ExecutorService exec = Executors.newCachedThreadPool();// 線程池 static void test(Runnable r) throws InterruptedException {// 使用 ExecutorService.submit() 而不是 ExecutorService.execute()方法來啟動任務，就可以持有該任務的上下文 ，submit() 方法返回 Future 對象 // exec.execute(r); 不用 execute() 方法 Future<?> f = exec.submit(r);TimeUnit.MILLISECONDS.sleep(1000); // 睡眠1秒 System.out.println("interrupting " + r.getClass().getName()); // 正在中斷某個線程 // 調用Future.cancel() 方法來中斷某個特定任務 // 把true傳給cancel() 方法，該方法就擁有在該線程上調用interrupt() 方法以停止這個線程的權限 // cancel 是一種中斷由 Executor啟動的單個線程的方式 f.cancel(true);System.out.println("interrupt sent to " + r.getClass().getName()); // 中斷信號發送給線程 System.out.println("====================================== seperate line ==================================== ");}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 {@Overridepublic void run() {try {TimeUnit.SECONDS.sleep(3);// 睡眠3秒} catch (InterruptedException e ) { // 捕獲中斷異常System.out.println("interrupted exception in SleepBlocked ");}System.out.println("exiting SleepBlocked.run()");}
}
// IO式阻塞線程 ， 不可中斷的阻塞
class IOBlocked implements Runnable {private InputStream in;public IOBlocked(InputStream is) {in = is; }@Overridepublic void run() {try {System.out.println("waiting for read();");in.read(); // 等待輸入流輸入數據 } catch (IOException e) { // IO 異常 ， 但執行結果沒有報 IO 異常 if (Thread.currentThread().isInterrupted()) {System.out.println("interrupted from blocked IO");} else {throw new RuntimeException();}}System.out.println("exiting IOBlocked.run()");}
}
// 線程同步式阻塞，不可中斷的阻塞
class SynchronizedBlocked implements Runnable {public synchronized void f() {while(true) {Thread.yield(); // 讓出cpu時間片 }}public SynchronizedBlocked() { // 構造器開啟一個線程 new Thread() { // 匿名線程調用f() 方法，獲取 SynchronizedBlocked 對象鎖，且不釋放；其他線程只能阻塞 public void run() {f();// f() 為同步方法 }}.start(); }@Override public void run() {System.out.println("trying to call 同步f()");f(); // 調用f() 同步方法 ， 讓出cpu時間片 System.out.println("exiting SynchronizedBlocked.run()"); // 這里永遠不會執行 }
}
/*interrupting diy.chapter21.SleepBlocked
interrupt sent to diy.chapter21.SleepBlocked
====================================== seperate line ====================================
interrupted exception in SleepBlocked
exiting SleepBlocked.run()
waiting for read();
interrupting diy.chapter21.IOBlocked
interrupt sent to diy.chapter21.IOBlocked
====================================== seperate line ====================================
trying to call 同步f()
interrupting diy.chapter21.SynchronizedBlocked
interrupt sent to diy.chapter21.SynchronizedBlocked
====================================== seperate line ====================================
aborting with System.exit(0)
*/
小結：
序號 阻塞方式 是否可以中斷 1 sleep 是 2 IO 否 3 synchronized獲取鎖 否
所以，對于IO操作線程或synchronized操作的線程，其具有鎖住多線程程序的潛在危險。
如何解決呢？ 關閉任務在其上發生阻塞的底層資源；
/*** 無法中斷線程，但可以關閉任務阻塞所依賴的資源。* 這里只能夠中斷 基于socket輸入流的io線程，因為socket輸入流可以關閉；* 但無法中斷基于系統輸入流的io線程，因為系統輸入流無法關閉；*/
public class CloseResource {public static void main(String[] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 ServerSocket server = new ServerSocket(8080); // 服務端套接字 InputStream socketInput = new Socket("localhost", 8080).getInputStream();/* 啟動線程 */exec.execute(new IOBlocked(socketInput));exec.execute(new IOBlocked(System.in));TimeUnit.MILLISECONDS.sleep(1000); // 睡眠1秒 System.out.println("shutting down all threads");exec.shutdownNow(); // 發送一個interrupte() 信號給exec啟動的所有線程 TimeUnit.SECONDS.sleep(1); // 睡眠1秒 System.out.println("closing " + socketInput.getClass().getName());socketInput.close(); // 關閉io線程依賴的資源 TimeUnit.SECONDS.sleep(1);System.out.println("closing " + System.in.getClass().getName());System.in.close(); // 關閉io線程依賴的資源 }
}
/**
waiting for read();
waiting for read();
shutting down all threads
closing java.net.SocketInputStream
interrupted from blocked IO
exiting IOBlocked.run()
closing java.io.BufferedInputStream
*/
3、nio類提供了更人性化的IO中斷，被阻塞的nio通道會自動響應中斷；?
/*** page 698* nio中斷 */
public class NIOInterruption {public static void main(String[] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 ServerSocket ss = new ServerSocket(8080); // 服務器套接字 // InetAddress:類的主要作用是封裝IP及DNS, // InetSocketAddress類主要作用是封裝端口 他是在在InetAddress基礎上加端口，但它是有構造器的。InetSocketAddress isa = new InetSocketAddress("localhost", 8080);SocketChannel sc1 = SocketChannel.open(isa); // 套接字通道 SocketChannel sc2 = SocketChannel.open(isa); // 套接字通道 // 使用 ExecutorService.submit() 而不是 ExecutorService.execute()方法來啟動任務，就可以持有該任務的上下文 ，submit() 方法返回 Future 對象 Future<?> f = exec.submit(new NIOBlocked(sc1));// 以submit方式啟動線程 exec.execute(new NIOBlocked(sc2)); // 以 execute方式啟動線程 exec.shutdown(); // 關閉所有線程 TimeUnit.SECONDS.sleep(1); // 睡眠1秒// 調用Future.cancel() 方法來中斷某個特定任務 // 把true傳給cancel() 方法，該方法就擁有在該線程上調用interrupt() 方法以停止這個線程的權限 // cancel 是一種中斷由 Executor啟動的單個線程的方式f.cancel(true); // sc2.close(); // }
}
// NIO 新io式阻塞
class NIOBlocked implements Runnable {private final SocketChannel sc;public NIOBlocked(SocketChannel sc) {this.sc = sc; }@Overridepublic void run() {try {System.out.println("waiting for read() in " + this);sc.read(ByteBuffer.allocate(1));} catch (ClosedByInterruptException e1) {System.out.println("ClosedByInterruptException, this = " + this);} catch (AsynchronousCloseException e2) {System.out.println("AsynchronousCloseException, this = " + this);} catch (IOException e3) {throw new RuntimeException(e3); }System.out.println("exiting NIOBlocked.run() " + this);}
}
/**
waiting for read() in diy.chapter21.NIOBlocked@3856c761
waiting for read() in diy.chapter21.NIOBlocked@55de2e48
ClosedByInterruptException, this = diy.chapter21.NIOBlocked@55de2e48
exiting NIOBlocked.run() diy.chapter21.NIOBlocked@55de2e48
AsynchronousCloseException, this = diy.chapter21.NIOBlocked@3856c761
exiting NIOBlocked.run() diy.chapter21.NIOBlocked@3856c761
*/
4、被互斥所阻塞： 一個任務能夠調用在同一個對象中的其他的 synchronized 方法，而這個任務已經持有鎖了 ；?
/*** 被互斥所阻塞* 同步方法f1 和 f2 相互調用直到 count為0 * 一個任務應該能夠調用在同一個對象中的其他 synchronized 方法，因為這個任務已經獲取這個對象的鎖* 2020/04/16 */
public class MultiLock {public synchronized void f1(int count) { // 同步方法 f1 if(count-- > 0) {System.out.println("f1() calling f2() with count = " + count);f2(count); // 調用 f2 }}public synchronized void f2(int count) { // 同步方法f2 if(count-- > 0) {System.out.println("f2() calling f1() with count = " + count);f1(count); // 調用f1 }}public static void main(String[] args) {final MultiLock multiLock = new MultiLock();new Thread() {public void run() {multiLock.f1(5);}}.start(); }
}
/**
f1() calling f2() with count = 4
f2() calling f1() with count = 3
f1() calling f2() with count = 2
f2() calling f1() with count = 1
f1() calling f2() with count = 0
*/
5、java se5 并發類庫中添加了一個特性，在 ReentrantLock? 可重入鎖上阻塞的任務具備可以被中斷的能力；?
/*** 可重入鎖的可中斷式加鎖 * page 700 */
public class Interrupting2 {public static void main(String[] args) throws Exception {Thread t = new Thread(new Blocked2());t.start();TimeUnit.SECONDS.sleep(1);System.out.println("issuing t.interrupt()"); // 2 t.interrupt(); // 中斷線程 }
}
/*** 阻塞互斥量 */
class BlockedMutex {private Lock lock = new ReentrantLock(); // 可重入鎖 public BlockedMutex() {lock.lock(); // 構造器即加鎖，且從不會釋放鎖 }public void f() {try {lock.lockInterruptibly(); // 可中斷式加鎖 System.out.println("lock acquired in f()");} catch(InterruptedException e) {System.out.println("interrupted from lock acquisition in f()"); // 3 可中斷阻塞，捕獲中斷異常 }}
}
class Blocked2 implements Runnable {BlockedMutex blocked = new BlockedMutex(); @Overridepublic void run() {System.out.println("waiting for f() in Blocked Mutex"); // 1 blocked.f();System.out.println("broken out of blocked call"); // 4 }
}
/*** waiting for f() in Blocked Mutex
issuing t.interrupt()
interrupted from lock acquisition in f()
broken out of blocked call*/
【21.4.4】檢查中斷
1、在線程上調用 interrupt方法去中斷線程執行時，能夠中斷線程的前提是： 任務要進入到阻塞操作中，已經在阻塞操作內部；否則，調用 interrupt方法是無法中斷線程的；需要通過其他方式；
其他方式是： 由中斷狀態來表示， 其狀態可以通過調用 interrupt 來設置。通過 Thread.interrupted() 來檢查中斷? 。
/*** 通過 Thread.interrupted() 來檢查中斷 * page 701*/
public class InterruptingIdiom {public static void main(String[] args) throws Exception {if(args.length != 1) {System.out.println("InterruptingIdiom-傻瓜式中斷");}Thread t = new Thread(new Blocked3()); t.start();TimeUnit.SECONDS.sleep(3); // 睡眠 t.interrupt(); // 中斷 }
}class NeedsCleanup {private final int id;public NeedsCleanup(int id) {this.id = id;System.out.println("NeedsCleanup " + id);}public void cleanup() {System.out.println("clean up " +id);}
}
/*** 在run()方法中創建的 NeedsCleanup 資源都必須在其后面緊跟 try-finally 子句， * 以確保 清理資源方法被調用 */
class Blocked3 implements Runnable {private volatile double d = 0.0; @Overridepublic void run() {try {int index = 1;// interrupted方法來檢查中斷狀態 while(!Thread.interrupted()) { // 只要當前線程沒有中斷 System.out.println("========== 第 " + index++ + " 次循環 =========="); NeedsCleanup n1 = new NeedsCleanup(1);try {System.out.println("sleeping-睡眠一秒");TimeUnit.SECONDS.sleep(1);NeedsCleanup n2 = new NeedsCleanup(2);try {System.out.println("calculating-高強度計算");for (int i=1; i<250000; i++) {d = d + (Math.PI + Math.E) / d;}System.out.println("finished time-consuming operation 完成耗時操作."); } finally {n2.cleanup(); // 清理 }} finally{n1.cleanup(); // 清理 }}System.out.println("exiting via while() test-從while循環退出 "); // 從while循環退出 } catch (InterruptedException e) {System.out.println("exiting via InterruptedException-從中斷InterruptedException退出 "); // 從中斷退出 }}}
/***
InterruptingIdiom-傻瓜式中斷
========== 第 1 次循環 ==========
NeedsCleanup 1
sleeping-睡眠一秒
NeedsCleanup 2
calculating-高強度計算
finished time-consuming operation 完成耗時操作.
clean up 2
clean up 1
========== 第 2 次循環 ==========
NeedsCleanup 1
sleeping-睡眠一秒
NeedsCleanup 2
calculating-高強度計算
finished time-consuming operation 完成耗時操作.
clean up 2
clean up 1
========== 第 3 次循環 ==========
NeedsCleanup 1
sleeping-睡眠一秒
clean up 1
exiting via InterruptedException-從中斷InterruptedException退出 */
?
【21.5】線程間的協作
1、當任務協作時，關鍵問題是任務間的握手。握手可以通過 Object.wait() Object.notify() 方法來安全實現。當然了 java se5 的并發類庫還提供了具有 await() 和 signal() 方法的Condition對象；
【21.5.1】wait()方法與notifyAll() 方法?
1、wait() 方法會在等待外部世界產生變化的時候將任務掛起，并且只有在 nofity() 或notifyall() 發生時，即表示發生了某些感興趣的事務，這個任務才會被喚醒去檢查鎖產生的變化。wait()方法提供了一種在任務之間對活動同步的方式。
還有，調用wait() 方法將釋放鎖，意味著另一個任務可以獲得鎖，所以該對象上的其他synchronized方法可以在線程A wait期間，被其他線程調用；?
2、有兩種形式的 wait() 調用
形式1： wait方法接收毫秒數作為參數，在wait()期間對象鎖是釋放的；通過 notify() notifyAll() 方法，或者時間到期后，從 wait() 恢復執行；?
形式2：wait方法不接受任何參數，這種wait將無線等待下去，直到線程接收到 notify或 notifyAll方法；?
補充1：wait方法，notify方法， notifyAll方法，都是基類Object的一部分，因為這些方法操作的鎖也是對象的一部分，而所有對象都是OBject的子類；?
補充2：實際上，只能在同步控制方法或同步控制塊里調用 wait， notify， notifyAll方法（因為不操作鎖，所有sleep方法可以在非同步控制方法里調用）。如果在非同步方法中調用 wait， notify， notifyAll方法， 編譯可以通過，但運行就報 IllegalMonitorStateException 異常，異常意思是： 在調用wait， notify， notifyAll方法前，必須獲取對象的鎖；?
（干貨——只能在同步控制方法或同步控制塊里調用 wait， notify， notifyAll方法）?
?
【荔枝】涂蠟與拋光： 拋光任務在涂蠟完成之前，是不能執行其工作的；而涂蠟任務在涂另一層蠟之前，必須等待拋光任務完成；
/*** 汽車上蠟與拋光* （拋光任務在涂蠟完成之前，是不能執行其工作的；而涂蠟任務在涂另一層蠟之前，必須等待拋光任務完成；） * page 705 */
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(1); // 睡眠5秒 exec.shutdown(); // 線程池關閉 }
}
// 汽車
class Car {private boolean waxOn = false; // 是否上蠟// 已上蠟 /*** notifyAll() 和 wait() 方法只能在 synchronized方法或synchronized塊中執行，因為獲取或釋放鎖 */public synchronized void waxed() { // 上蠟 waxOn = true; notifyAll(); // 喚醒所有調用 wait() 方法鎖阻塞的線程 // 為了使該任務從 wait() 中喚醒，線程必須重新獲得之前進入wait()時釋放的鎖。// 在這個鎖變得可用之前，這個任務是不會被喚醒的。}public synchronized void buffed() { // 拋光 waxOn = false; notifyAll();}public synchronized void waitForWaxing() throws InterruptedException { // 等待上蠟 while (waxOn == false) { // 若沒有上蠟，則等待 wait(); // 線程被掛起， 當前線程持有的car對象鎖被釋放 }}public synchronized void waitForBuffing() throws InterruptedException { // 等待拋光 while(waxOn == true) { // 若已上蠟，則等待拋光wait(); // 線程被掛起， 當前線程持有的car對象鎖被釋放}}
}
class WaxOn implements Runnable { // 上蠟線程（本線程先執行第1次上蠟，等待拋光，拋光線程第1次執行拋光后，本線程執行第2次上蠟......） private Car car; public WaxOn(Car c) {this.car = c; }@Overridepublic void run() {try {while(!Thread.interrupted()) {System.out.println("wax on !");TimeUnit.MILLISECONDS.sleep(200);car.waxed(); // 先上蠟完成 （把waxOn設置為true），喚醒等待上蠟的線程 car.waitForBuffing(); // 再等待拋光，當waxOn為ture，則拋光線程一直等待 }} catch (InterruptedException e ) {System.out.println("exiting via interrupt");}System.out.println("ending wax on task"); }
}
class WaxOff implements Runnable { // 拋光線程（本線程先等待上蠟，上蠟線程第1次執行后，本線程立即執行第1次拋光，接著本線程等待第2次上蠟......） private Car car; public WaxOff(Car c) {this.car = c; }@Overridepublic void run() {try {while(!Thread.interrupted()) {car.waitForWaxing(); // 先等待上蠟 , 當waxOn為false，則上蠟線程一直等待 System.out.println("wax off !"); // TimeUnit.MILLISECONDS.sleep(200);car.buffed(); // 拋光完成后，把waxOn設置為false，喚醒等待拋光的線程 }} catch (InterruptedException e ) {System.out.println("exiting via interrupt");}System.out.println("ending wax off task"); }
}
/*** wax on !
wax off !
wax on !
wax off !
...... */
補充：前面的實例強調必須用一個檢查感興趣的條件的while循環包圍wait方法。這很重要，因為：（為啥要用while包裹wait呢）
前面的示例強調必須用一個檢查感興趣的條件的while循環包圍wait()。這很重要，原因如下：（個人理解——比如有2個任務A，B都在等待資源R可用而阻塞，當R可用時，任務A和B均被喚醒，但任務A被喚醒后立即拿到了臨界資源或獲取了鎖，則任務B仍然需要再次阻塞，這就是while的作用）
用while 包圍wait方法的本質：檢查所有感興趣的條件，并在條件不滿足的情況下再次調用wait方法，讓任務再次阻塞；
3、錯失的信號：當兩個線程使用 notify/wait() 或 notifyAll()/ wait() 方法進行協作時，有可能會錯過某個信號；即 notify或 notifyAll發出的信號，帶有wait的線程無法感知到。
荔枝：
// T1:
synchronized(sharedMonitor) {<setup condition for T2>sharedMonitor.notify() // 喚醒所有等待線程
}
// T2:
while(someCondition) {// point 1 synchronized(sharedMonitor) {sharedMonitor.wait(); // 當前線程阻塞 }
}
當T2 還沒有調用 wait方法時，T1就發送了notify信號； 這個時候T2線程肯定接收不到這個信號；T1發送信號notify后，T2才調用wait方法，這時，T2將永久阻塞下去；因為他錯過了T1的notify信號；
T2正確的寫法如下：
// T2正確的寫法如下：
synchronized(sharedMonitor) {while(someCondition) {sharedMonitor.wait(); // 當前線程阻塞 }
}
如果T1先執行后釋放鎖；此時T2獲取鎖且檢測到 someCondition已經發生了變化，T2不會調用wait() 方法；?
如果T2先執行且調用了wait（）方法， 釋放了鎖； 這時T1后執行，然后調用notify()喚醒阻塞線程， 這時T2可以收到T1的 notify信號
?
【21.5.2】notify與notifyAll方法
/*** page 707* notify 與 notifyAll的對比， * notify 喚醒單個阻塞線程，而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()); // 運行5個任務， 只要task 任務一運行就會阻塞，除非被喚醒 }exec.execute(new Task2()); // 運行第6個任務， 只要task2 任務一運行就會阻塞，除非被喚醒 Timer timer = new Timer(); // 定時器 // 定時調度， 延遲400毫秒開始執行，兩次運行的時間間隔為500毫秒 timer.scheduleAtFixedRate(new TimerTask() {boolean prod = true; @Overridepublic void run() {if (prod) {System.out.println("\n notify() ");Task.blocker.prod(); // 喚醒單個阻塞線程 prod = false ;} else {System.out.println("\n notifyAll()");Task.blocker.prodAll(); // 喚醒所有阻塞線程 prod = true ;}}}, 400, 500);TimeUnit.SECONDS.sleep(5);timer.cancel(); // 關閉定時器，關閉所有線程，正在運行的任務除外 System.out.println("timer canceled");TimeUnit.MILLISECONDS.sleep(500); // 睡眠500毫秒 System.out.println("task2.blocker.prodAll()");Task2.blocker.prodAll(); // task2 喚醒所有阻塞線程 TimeUnit.MILLISECONDS.sleep(500); // 睡眠500毫秒System.out.println("\n shutting down");exec.shutdownNow(); // 關閉線程池 }
}
// 阻塞器
class Blocker {synchronized void waitingCall() {try {while(!Thread.interrupted()) {wait(); // 期初所有線程均阻塞，等待 notify 或 notifyAll 來喚醒 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();// 阻塞器 @Overridepublic void run() {blocker.waitingCall();// wait() 方法阻塞 }
}
// 任務2
class Task2 implements Runnable {static Blocker blocker = new Blocker(); // 阻塞器 @Overridepublic void run() {blocker.waitingCall(); // wait() 方法阻塞 }
}
/*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]
timer canceled
task2.blocker.prodAll()
Thread[pool-1-thread-6,5,main] shutting down */
補充：
// 阻塞器?
class Blocker {synchronized void waitingCall() {try {while(!Thread.interrupted()) {wait(); // 期初所有線程均阻塞，等待 notify 或 notifyAll 來喚醒?System.out.println(Thread.currentThread() + " ");}} catch (InterruptedException e ) {}}synchronized void prod() {notify();// 喚醒單個阻塞線程?}synchronized void prodAll() {notifyAll(); // 喚醒所有阻塞線程?}
}
Blocker.waitingCall 方法中的while循環， 有兩種方式可以離開這個循環：
【21.5.3】生產者與消費者
/*** page 709* 生產者（廚師chef）-與消費者（服務員WaitPerson） */
public class Restaurant { // 餐館 Meal meal ; ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 WaitPerson waitPerson = new WaitPerson(this); // 服務員 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;} @Overridepublic void run() {try {while(!Thread.interrupted()) {synchronized (this) {while(restaurant.meal == null) { // 沒有菜可以上，服務員等待 wait(); // 阻塞，直到 notify 或 notifyAll 喚醒 }}System.out.println("服務器取到餐=WaitPerson got " + restaurant.meal);synchronized (restaurant.chef) { // 廚師 restaurant.meal = null; restaurant.chef.notifyAll(); // 喚醒所有阻塞在 chef對象上的線程 }}} catch (InterruptedException e) {System.out.println("WaitPerson interrupted（服務員線程中斷）");}}
}
class Chef implements Runnable {// 廚師（生產者） private Restaurant restaurant; private int count = 0;public Chef(Restaurant r) {restaurant = r; }@Overridepublic void run() {try {while(!Thread.interrupted()) {synchronized (this) {while(restaurant.meal != null) { // 菜沒有被端走，廚師等待 wait(); // 阻塞，直到 notify 或 notifyAll 喚醒}}if (++count == 10) { // 廚師只做10個菜 System.out.println("out of food, closing。廚師只做10個菜，關閉線程池");restaurant.exec.shutdownNow(); // 關閉餐館的線程池，該池運行著廚師和服務員任務 }System.out.println("廚師說，上菜了，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");}}
}
/*
廚師說，上菜了，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。廚師只做10個菜，關閉線程池
廚師說，上菜了，order up!
WaitPerson interrupted（服務員線程中斷）
chef interrupted
*/
2、代碼解說， Restraurant是 WaitPerson和Chef的焦點，作為連接兩者的橋梁。他們知道在為哪個Restraurant工作，因為他們必須和這家飯店打交道，以便放置或拿取膳食。2.1、（干貨）再次提問：如果在等待一個訂單，一旦你被喚醒，這個訂單就必定是可以獲得的嗎？
3、使用顯式的Lock和 Condition 對象 ?
/*** page 711 * 使用顯式的Lock 和 Condition對象 */
public class WaxOMatic2 {public static void main(String[] args) throws InterruptedException {Car2 car = new Car2(); ExecutorService executorService = Executors.newCachedThreadPool();executorService.execute(new WaxOff2(car)); // 拋光executorService.execute(new WaxOn2(car)); // 打蠟 TimeUnit.SECONDS.sleep(1); // 睡眠5秒 executorService.shutdownNow(); }
}
class Car2 {private Lock lock = new ReentrantLock(); // 可重入鎖 private Condition condition = lock.newCondition(); // 獲取鎖的條件 private boolean waxOn = false; // 期初時，沒有上蠟public void waxed() { // 上蠟 lock.lock(); // 加鎖 try { waxOn = true; // 上蠟完成 condition.signalAll(); // 喚醒所有等待線程 } finally {lock.unlock(); // 解鎖 }}public void buffed() { // 拋光 lock.lock(); // 加鎖 try {waxOn = false; // 拋光完成，待上蠟 condition.signalAll();} finally {lock.unlock(); // 解鎖 }}public void waitForWaxing() throws InterruptedException { // 等待上蠟 lock.lock();try {while(waxOn == false) { // 還未上蠟，等待上蠟condition.await(); // 掛起 }} finally {lock.unlock();}}public void waitForBuffing() throws InterruptedException { // 等待拋光 lock.lock();try {while(waxOn == true) { // 上蠟完成，等待拋光 condition.await(); // 掛起 }} finally {lock.unlock(); }}
}
class WaxOn2 implements Runnable { // 打蠟任務 private Car2 car ; public WaxOn2(Car2 c) {this.car = c; }@Overridepublic void run() {try {while(!Thread.interrupted()) {System.out.println("wax on ");TimeUnit.MILLISECONDS.sleep(200);car.waxed(); // 打蠟完成 car.waitForBuffing(); // 等待拋光 }} catch(InterruptedException e ) {System.out.println("WaxOn2 exiting via interrupt");}System.out.println("WaxOn2 ending wax on task");}
}
class WaxOff2 implements Runnable {// 打蠟結束，開始拋光任務 private Car2 car; public WaxOff2(Car2 c) {this.car = c; }@Overridepublic void run() {try {while(!Thread.interrupted()) {car.waitForWaxing(); // 等待打蠟 System.out.println("wax off");TimeUnit.MILLISECONDS.sleep(200);car.buffed(); // 拋光完成 }} catch(InterruptedException e ) {System.out.println("WaxOff2 exiting via interrupt");}System.out.println("WaxOff2 ending wax off task");}
}
/*
wax on
wax off
wax on
wax off
wax on
WaxOff2 exiting via interrupt
WaxOff2 ending wax off task
WaxOn2 exiting via interrupt
WaxOn2 ending wax on task
*/
代碼解說：每個對lock()的調用都必須緊跟一個try-finally子句，用來保證在所有情況下都可以釋放鎖。在使用內建版本時，任務在可以調用 await()， signal(), signalAll() 方法前，必須擁有這個鎖。（干貨——不推薦使用Lock和Condition對象來控制并發）
【21.5.4】生產者與消費者隊列
/*** 阻塞隊列 * page 714 */
public class TestBlockingQueues {static void getKey() {try {// 從控制臺讀入用戶輸入new BufferedReader(new InputStreamReader(System.in)).readLine(); } catch (IOException e) {throw new RuntimeException(e); }}static void getKey(String msg) {System.out.println(msg);getKey(); }static void test(String msg, BlockingQueue<LiftOff> queue) {System.out.println(msg);LiftOffRunner runner = new LiftOffRunner(queue);Thread t = new Thread(runner);t.start();for (int i=0; i<3; i++) {runner.add(new LiftOff(3)); // 添加5個發射任務到阻塞隊列 } getKey("press enter " + msg);t.interrupt(); // 線程中斷 System.out.println("finished " + msg + " test");}public static void main(String[] args) {test("LinkedBlockingQueue", new LinkedBlockingQueue<LiftOff>()); // 鏈表阻塞隊列，無界 test("ArrayBlockingQueue", new ArrayBlockingQueue<LiftOff>(3)); // 數組阻塞隊列，固定長度 test("SynchronousQueue", new SynchronousQueue<LiftOff>()); // 同步隊列 }
}
// lift off 發射，起飛
class LiftOffRunner implements Runnable {private BlockingQueue<LiftOff> rockets; // 阻塞隊列，火箭隊列public LiftOffRunner(BlockingQueue<LiftOff> queue) {this.rockets = queue; }public void add(LiftOff lo) { // LiftOff 發射起飛任務 try {rockets.put(lo); // 往隊列里面放入 發射起飛任務 } catch (InterruptedException e) {System.out.println("interupted during put()");}}@Overridepublic void run() {try {while(!Thread.interrupted()) {LiftOff rocket = rockets.take(); // 從隊列中取出任務，運行，沒有任務，則阻塞 rocket.run(); }} catch (InterruptedException e) {System.out.println("waking from task()");}System.out.println("exiting LiftOffRunner"); }
}
/*
LinkedBlockingQueue
press enter LinkedBlockingQueue
#0(2),
#0(1),
#0(liftoff),
#1(2),
#1(1),
#1(liftoff),
#2(2),
#2(1),
#2(liftoff), finished LinkedBlockingQueue test
waking from task()
exiting LiftOffRunner
ArrayBlockingQueue
press enter ArrayBlockingQueue
#3(2),
#3(1),
#3(liftoff),
#4(2),
#4(1),
#4(liftoff),
#5(2),
#5(1),
#5(liftoff), finished ArrayBlockingQueue test
waking from task()
exiting LiftOffRunner
SynchronousQueue
#6(2),
#6(1),
#6(liftoff),
#7(2),
#7(1),
#7(liftoff),
#8(2),
press enter SynchronousQueue
#8(1),
#8(liftoff), finished SynchronousQueue test
waking from task()
exiting LiftOffRunner*/
【吐司BlockingQueue】
/*** 吐司制作程序-* 一臺機器有3個任務：第1制作吐司， 第2抹黃油，第3涂果醬，阻塞隊列-LinkedBlockingQueue * page 715 */
public class ToastOMatic {public static void main(String[] args) throws Exception {ToastQueue dryQueue = new ToastQueue(); // 烘干的吐司隊列 ToastQueue butterQueue = new ToastQueue(); // 涂黃油的吐司隊列 ToastQueue finishQueue = new ToastQueue(); // 制作完成的吐司隊列 /* 線程池 */ExecutorService exec = Executors.newCachedThreadPool();exec.execute(new Toaster(dryQueue)); // 吐司exec.execute(new Butterer(dryQueue, butterQueue)); // 黃油 exec.execute(new Jammer(butterQueue, finishQueue)); // 果醬 exec.execute(new Eater(finishQueue)); // 吃 TimeUnit.SECONDS.sleep(1);exec.shutdownNow(); // 發出中斷信號， 線程報 InterruptedException 中斷異常 , 所有線程均結束exec.shutdown(); }
}
class Toast { // 吐司類 public enum Status{DRY, BUTTERED, JAMMED}; // 枚舉類 dry-烘干， butter-黃油，jam-果醬 private Status status = Status.DRY; private final int id ;public Toast(int id) { // 編號 this.id = id; }public void butter() { // 抹黃油結束 status = Status.BUTTERED;}public void jam() { // 涂果醬結束 status = Status.JAMMED;}public Status getStatus() {return status; }public int getId() {return id; }public String toString() {return "toast " + id + " : " + status; }
}
class ToastQueue extends LinkedBlockingQueue<Toast> {} // 吐司隊列 class Toaster implements Runnable { // 第1個工序： 做吐司 private ToastQueue toastQueue; // 吐司隊列 private int count = 0; // 計數器 private Random rand = new Random(47);public Toaster(ToastQueue toastQueue) {this.toastQueue = toastQueue; } @Override public void run() {try {while(!Thread.interrupted()) { // 只要任務不中斷 TimeUnit.MILLISECONDS.sleep(100+ rand.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 { // 第2個工序：黃油 private ToastQueue dryQueue, butterQueue;public Butterer(ToastQueue dryQueue, ToastQueue butterQueue) { // 已烘干吐司隊列， 已抹黃油的吐司隊列 this.dryQueue = dryQueue;this.butterQueue = butterQueue; }@Overridepublic void run() {try {while(!Thread.interrupted()) {Toast t = dryQueue.take(); // 獲取烘干的吐司 t.butter(); // 抹黃油 System.out.println(t);butterQueue.put(t); // 往黃油隊列添加 }} catch (InterruptedException e) {System.out.println("butterer interrupted ");}System.out.println("butterer off");}
}
class Jammer implements Runnable { // 第3個工序，涂果醬 private ToastQueue butterQueue, finishQueue; public Jammer(ToastQueue butterQueue, ToastQueue finishQueue) {this.butterQueue = butterQueue;this.finishQueue = finishQueue;}@Overridepublic void run() {try {while(!Thread.interrupted()) {Toast t = butterQueue.take(); // 從抹黃油隊列中獲取吐司 t.jam(); // 涂果醬 System.out.println(t);finishQueue.put(t); // 添加到完成隊列 }} catch (InterruptedException e ) {System.out.println("jammer interrupted");}System.out.println("jammer off"); // 涂果醬完成 }
}
class Eater implements Runnable { // 消費者，吃吐司 private ToastQueue finishQueue; private int counter = 0;public Eater(ToastQueue finishQueue) {this.finishQueue = finishQueue;}@Overridepublic void run() {try {while(!Thread.interrupted()) {Toast t = finishQueue.take(); // 從吐司制作完成隊列中獲取吐司 if (t.getId() != counter++ || t.getStatus() != Toast.Status.JAMMED) {System.out.println(">>>> Error: " + t);System.exit(1);} else {System.out.println("chomp !" + t); // chomp-大聲咀嚼，吃吐司 }}} catch (InterruptedException e) {System.out.println("eater interrupted");}System.out.println("eat off"); // 吃飽回家 }
}
/*toast 0 : DRY
toast 0 : BUTTERED
toast 0 : JAMMED
chomp !toast 0 : JAMMED
toast 1 : DRY
toast 1 : BUTTERED
toast 1 : JAMMED
chomp !toast 1 : JAMMED
toast 2 : DRY
toast 2 : BUTTERED
toast 2 : JAMMED
chomp !toast 2 : JAMMED
eater interrupted
eat off
toaster interrupted
toaster off
butterer interrupted
butterer off
jammer interrupted
jammer off* */
【21.5.5】任務間使用管道進行輸入輸出
/*** 任務間使用管道進行輸入輸出 * page 718 */
public class PipedIO {public static void main(String[] args) throws Exception {Sender sender = new Sender(); // 發送器Receiver receiver = new Receiver(sender); // 接收器 ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 exec.execute(sender); // 發送 exec.execute(receiver); // 接收 TimeUnit.SECONDS.sleep(3);exec.shutdown(); // 關閉線程池 }
}
//發送者任務
class Sender implements Runnable { private Random rand = new Random(47); // 隨機數 private PipedWriter out = new PipedWriter(); // 管道輸出對象 public PipedWriter getPipedWriter() {return out; }@Overridepublic void run() {try {while(true) {for (char c = 'A'; c <= 'z'; c++) {out.write(c); // 把字符輸出到管道 TimeUnit.MILLISECONDS.sleep(rand.nextInt(500));}}} catch (IOException e) {System.out.println("\n" + e + " sender write exception ");} catch (InterruptedException e2) {System.out.println("\n" + e2 + " sender sleep interrupted. ");}}
}
// 接收者任務
class Receiver implements Runnable {private PipedReader in ; public Receiver(Sender sender) throws IOException {in = new PipedReader(sender.getPipedWriter()); // 管道輸入對象 }@Overridepublic void run() {try {while(true ) {System.out.print("read:" + (char)in.read() + ", ");// 從管道讀取數據 }} catch (IOException e ) {System.out.println("\n" + e + " receiver read exception.");}}
}
代碼解說1： 當Receiver調用read() 方法時，如果沒有更多的數據，管道將自動阻塞；（干貨） （干貨）
【21.6】死鎖
1、作為哲學家，他們很窮，只能買5根筷子（通俗講，筷子數量與哲學家數量相同）；他們坐在桌子周圍，每人之間放一根筷子，當一個哲學家要就餐的時候，這個哲學家必須同時擁有左邊和右邊的筷子。
/*** 死鎖-筷子* page719 */
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】
/*** 哲學家* page 718 */
public class Philosopher implements Runnable {private Chopstick left; // 左筷private Chopstick right; // 右筷 private int id = 1; // 編號 private int ponderFactor = 0; // 思考因素 private Random rand = new Random(47); // 隨機數發生器 private void pause() throws InterruptedException { // 暫停 if (ponderFactor == 0) return ; TimeUnit.MILLISECONDS.sleep(rand.nextInt(ponderFactor * 250)); // 睡眠 }public Philosopher(Chopstick left, Chopstick right, int ident, int ponder) {this.left = left; this.right = right; this.id = ident; this.ponderFactor = ponder; } @Overridepublic 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. " ); }}@Overridepublic String toString() {return "Philosopher-哲學家" + id; }
}
【代碼-DeadLockDiningPhilosophers】
/*** 發生死鎖的哲學家晚餐* page720 */
public class DeadLockDiningPhilosophers {public static void main(String[] args) throws Exception {int ponder = 0; // 把 ponder-思考時間 調整為0，發生死鎖 int size = 5; ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 Chopstick[] sticks = new Chopstick[size]; // 筷子數組 for (int i = 0; i < sticks.length; i++) {sticks[i] = new Chopstick(); // 初始化數組 }for (int i = 0; i < sticks.length; i++) {exec.execute(new Philosopher(sticks[i], sticks[(i+1)%size], i+1, ponder)); // 執行哲學家任務 }System.out.println("press enter to quit");System.in.read(); exec.shutdownNow(); }
}
/** 死鎖發生了
Philosopher-哲學家2 thinking
Philosopher-哲學家4 thinking
press enter to quit
Philosopher-哲學家1 thinking
Philosopher-哲學家3 thinking
Philosopher-哲學家3 grabbing right
Philosopher-哲學家1 grabbing right
Philosopher-哲學家5 thinking
Philosopher-哲學家4 grabbing right
Philosopher-哲學家4 grabbing left
Philosopher-哲學家2 grabbing right
Philosopher-哲學家5 grabbing right
Philosopher-哲學家1 grabbing left
Philosopher-哲學家3 grabbing left
Philosopher-哲學家5 grabbing left
Philosopher-哲學家2 grabbing left
*/
代碼解說：如果philosopher花費更多的時間思考而不是進餐（ponder值越大，思考時間越長），那么他們請求共享資源的可能性就會小許多，這樣你就會確信該程序不會死鎖，盡管他們并非如此。
2、死鎖發生條件： 當以下4個條件同事滿足時，死鎖發生；（干貨——死鎖發生的4個條件，同時滿足）
3、要防止死鎖，只需要破壞其中一個條件即可。最容易的方法是破壞第4個條件。
【21.7】新類庫中的構件
/*** count down-倒計時* latch-鎖存器 * page723 */
public class CountDownLatchDemo {static final int size = 5;public static void main(String[] args) {ExecutorService exec = Executors.newCachedThreadPool();CountDownLatch latch = new CountDownLatch(size); // 倒計時鎖存器 for (int i = 0; i < size; i++) {exec.execute(new WaitingTask(latch)); // 運行10個等待任務，直到鎖存器計數值為0 }for (int i = 0; i < size; i++) {exec.execute(new TaskPortion(latch)); // 運行任務，使得鎖存器計數值遞減 }System.out.println("launched all tasks"); // 啟動所有任務 exec.shutdown(); // 待所有線程執行完成，則線程池自動關閉 }
}
class TaskPortion implements Runnable { // 任務部分 private static int counter = 0; // 計數器private final int id = counter++;private static Random rand = new Random(47);private final CountDownLatch latch; // 遞減鎖存器 public TaskPortion(CountDownLatch latch) {this.latch = latch ; }@Overridepublic void run() {try {this.doWork(); // 做任務-睡眠 latch.countDown(); // 減少鎖存器計數 } catch (InterruptedException e ) {System.out.println(this + " interrupted"); } }public void doWork() throws InterruptedException {TimeUnit.MILLISECONDS.sleep(rand.nextInt(2000));System.out.println(this + " completed ");}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;}@Overridepublic void run() {try {latch.await(); // 使得當前線程等待或把自己攔住，直到鎖存器latch計數減至0， 或者本線程中斷 System.out.println("通過鎖存器障礙-latch barrier passed for " + this);} catch (InterruptedException e ) {System.out.println(this + " interrupted");}}@Override public String toString() {return String.format("waiting task %1$-3d", id); }
}
/*launched all tasks
1 completed
2 completed
4 completed
0 completed
3 completed
通過鎖存器障礙-latch barrier passed for waiting task 2
通過鎖存器障礙-latch barrier passed for waiting task 4
通過鎖存器障礙-latch barrier passed for waiting task 3
通過鎖存器障礙-latch barrier passed for waiting task 1
通過鎖存器障礙-latch barrier passed for waiting task 0
*/
【21.7.2】CyclicBarrier
/*** 循環障礙-賽馬比賽 * page724*/
public class HorseRace {static final int FINISH_LINE = 10; private List<Horse> horses = new ArrayList<>(); private ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 private CyclicBarrier barrier; public HorseRace(int number, final int pause) {// 當有number個線程等待時，barrier就會斷開。當其斷開時，給定的任務就會執行// ，由最后一個進入 barrier的線程提供資源執行。 barrier = new CyclicBarrier(number, new Runnable() { @Override public void run() { // 這個任務會多次執行，因為 當有number個線程等待時會發生多次 StringBuilder builder = new StringBuilder(); for (int i = 0; i < FINISH_LINE; i++) {builder.append("="); // 追加字符串 }System.out.println("builder");for (Horse horse : horses) {System.out.println(horse.tracks()) ; // 調用跟蹤方法 } for (Horse horse2 : horses) { // 遍歷任務 if (horse2.getStrides() >= FINISH_LINE) { System.out.println(horse2 + " own !"); exec.shutdownNow(); // 這里會終止掉所有線程的執行 return ;}}try {TimeUnit.MILLISECONDS.sleep(pause);} catch (InterruptedException e) {System.out.println("barrier-action sleep interrupted");}} }); for (int i = 0; i < number; i++) {Horse horse = new Horse(barrier);horses.add(horse);exec.execute(horse); // 執行賽馬任務 }}public static void main(String[] args) {int number = 5; // 5個馬 int pause = 100;// 暫停時間 new HorseRace(number, pause); }
}
class Horse implements Runnable { // 賽馬任務 private static int counter = 0; // 計數器 private final int id = counter++; // 編號 private int strides = 0; // 步數 private static Random rand = new Random(47);private static CyclicBarrier barrier; // 柵欄 public Horse(CyclicBarrier cyclicBarrier) {this.barrier = cyclicBarrier;}public synchronized int getStrides () {return strides; }@Overridepublic void run() {try {while(!Thread.interrupted()) {synchronized(this) {strides += rand.nextInt(3); // 步數自加 } barrier.await(); // 當前線程等待，直到所有線程在該 barrier等待為止 } } catch (InterruptedException e) {throw new RuntimeException(e);} catch (BrokenBarrierException e2) {throw new RuntimeException(e2);} }@Override public String toString() {return "horse " + id + " "; }public String tracks() { // 跟蹤方法 StringBuilder builder = new StringBuilder();for (int i = 0; i < getStrides(); i++) {builder.append("*"); }builder.append(id); return builder.toString(); }
}
/*builder
**0
**1
*2
**3
*4
builder
***0
****1
***2
**3
**4
builder
***0
*****1
***2
****3
**4
builder
*****0
******1
****2
*****3
***4
builder
*******0
*******1
****2
*****3
****4
builder
*******0
*********1
******2
******3
*****4
builder
********0
**********1
******2
******3
******4
horse 1 own !
*/
【21.7.3】 DelayQueue
/*** 阻塞隊列演示* page 726 */
public class DelayQueueDemo {public static void main(String[] args) {Random rand = new Random(47);ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 DelayQueue<DelayedTask> queue = new DelayQueue<>(); // 延遲隊列 for (int i = 0; i < 5; i++) {queue.put(new DelayedTask(rand.nextInt(5000))); // 往隊列添加延遲任務，延遲時間為5000內的隨機數，小于5000毫秒 }queue.add(new DelayedTask.EndSentinel(5000, exec)); // 再添加一個延遲任務（該任務，負責關閉線程池），延遲時間為5000毫秒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 delayInMilliseconds) {delta = delayInMilliseconds; // 延遲毫秒數 trigger = System.nanoTime() + TimeUnit.NANOSECONDS.convert(delta, TimeUnit.MILLISECONDS); // 時間單位轉換，有精度丟失 sequence.add(this);}@Override public long getDelay(TimeUnit unit) { // 獲取延遲時間 return unit.convert(trigger - System.nanoTime(), TimeUnit.NANOSECONDS);}@Overridepublic int compareTo(Delayed o) { // 時間比較 DelayedTask that = (DelayedTask) o; if (trigger < that.trigger) return -1; if (trigger > that.trigger) return 1; return 0;}@Overridepublic void run() {System.out.println(this + " ");}@Overridepublic 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 exec) {super(delay);this.exec = exec; }@Overridepublic void run() { for (DelayedTask task : sequence) { // 獲取每個延遲任務 System.out.println("task.summary() " + task.summary());}System.out.println(this + " calling shutdownNow() 立即關閉線程池-關閉所有線程");exec.shutdownNow(); // 關閉線程池 }}
}
class DelayedTaskConsumer implements Runnable { // 延遲任務消費者 private DelayQueue<DelayedTask> queue; public DelayedTaskConsumer(DelayQueue<DelayedTask> queue) {this.queue = queue; }@Overridepublic void run() {try {while(!Thread.interrupted()) {queue.take().run(); // 獲取隊列任務并運行 }} catch (InterruptedException e) {System.out.println("DelayedTaskConsumer interrupted");}System.out.println("finish DelayedTaskConsumer");}
}
/*[555 ] task 1
[961 ] task 4
[1693] task 2
[1861] task 3
[4258] task 0
task.summary() (0: 4258)
task.summary() (1: 555)
task.summary() (2: 1693)
task.summary() (3: 1861)
task.summary() (4: 961)
task.summary() (5: 5000)
[5000] task 5 calling shutdownNow() 立即關閉線程池-關閉所有線程
finish DelayedTaskConsumer
*/
代碼解說： 上述控制臺輸出信息為：
[555 ] task 1
【21.7.4】 PriorityBlockingQueue 優先級阻塞隊列
/*** 優先級阻塞隊列演示* page728 */
public class PriorityBlockQueueDemo {public static void main(String[] args) {ExecutorService exec = Executors.newCachedThreadPool(); // 線程吃 PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<>(); // 優先級阻塞隊列 exec.execute(new PrioritizedTaskProducer(queue, exec)); // 任務生產者 -創建41個任務exec.execute(new PrioritizedTaskConsumer(queue));}
}
// 優先級任務
class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {private Random rand = new Random(47);private static int counter = 0;private final int id = counter++; // 計數器 private int priority = 0; // 優先級 protected static List<PrioritizedTask> sequence = new ArrayList<>(); public PrioritizedTask(int priority) {this.priority = priority; sequence.add(this); }@Overridepublic int compareTo(PrioritizedTask o) { // 比較 // 值越大，優先級越高，越先執行 return priority < o.priority ? 1 : (priority > o.priority ? -1 : 0) ;}@Override public void run() { try {TimeUnit.MILLISECONDS.sleep(rand.nextInt(250)); // 做任務就是睡眠 } catch (InterruptedException e) {System.out.println("PrioritizedTask Interrupted"); }System.out.println(this); }@Overridepublic String toString() {return String.format("toString() = 線程優先級[%1$-3d]", priority) + " task-線程編號- " + id; }public String summary() {return "(summary() = 線程編號：" + id + ": 線程優先級：" + priority +")"; }public static class EndSentinel extends PrioritizedTask { // 哨兵任務 private ExecutorService exec; public EndSentinel(ExecutorService exec) {super(-1); // 優先級為-1， 值越小，越后執行 this.exec = exec; }@Overridepublic void run() {int count = 0;for (PrioritizedTask task : sequence) { // 遍歷每個任務，打印任務詳情 System.out.println(task.summary());if (++count % 5 == 0) {System.out.println(" --------------------我是換行符-------------------- ");}} System.out.println();System.out.println(this + " calling shutdownNow() 關閉線程池 ");this.exec.shutdownNow(); // 關閉所有任務 }}
}
// 任務生產者 -創建16個任務
class PrioritizedTaskProducer implements Runnable {private Random rand = new Random(47);private Queue<Runnable> queue; // 任務隊列 （優先級隊列 ）private ExecutorService exec; // 線程池 public PrioritizedTaskProducer(Queue<Runnable> queue, ExecutorService exec) {this.queue = queue; this.exec = exec; }@Overridepublic void run() { // 任務生產者 for (int i = 0; i < 10; i++) { queue.add(new PrioritizedTask(rand.nextInt(10))); // 往隊列添加任務，優先級小于10，比10先執行 Thread.yield(); // 當前線程讓出cpu 時間片 }try {for (int i = 0; i < 3; i++) {TimeUnit.MILLISECONDS.sleep(250); // 當前線程睡眠 queue.add(new PrioritizedTask(10)); // 再往隊列添加3個任務 ，其優先級為10，已添加13個}for (int i = 0; i < 2; i++) {queue.add(new PrioritizedTask(i)); // 再添加2個任務，其優先級為i，已添15個}// 添加任務EndSentinel，該任務會遍歷每個任務，打印任務詳情，并會關閉線程池queue.add(new PrioritizedTask.EndSentinel(exec));} catch (InterruptedException e) {System.out.println("PrioritizedTaskProducer interrupted");}System.out.println("finish PrioritizedTaskProducer");}
}
// 任務消費者
class PrioritizedTaskConsumer implements Runnable {private PriorityBlockingQueue<Runnable> queue; // PriorityBlockingQueue-優先級阻塞隊列 public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> queue) {this.queue = queue; }@Override public void run() {try {while(!Thread.interrupted()) {// 從優先級隊列中取出任務并執行， PrioritizedTask優先級任務 的優先級值越大，優先級越高，越先執行 queue.take().run(); }} catch (InterruptedException e) {System.out.println("PrioritizedTaskConsumer interrupted");}System.out.println("finish PrioritizedTaskConsumer");}
}
/*
toString() = 線程優先級[9 ] task-線程編號- 5
toString() = 線程優先級[8 ] task-線程編號- 0
toString() = 線程優先級[8 ] task-線程編號- 6
toString() = 線程優先級[7 ] task-線程編號- 9
toString() = 線程優先級[5 ] task-線程編號- 1
toString() = 線程優先級[3 ] task-線程編號- 2
toString() = 線程優先級[2 ] task-線程編號- 8
toString() = 線程優先級[1 ] task-線程編號- 4
toString() = 線程優先級[1 ] task-線程編號- 3
toString() = 線程優先級[0 ] task-線程編號- 7
toString() = 線程優先級[10 ] task-線程編號- 10
toString() = 線程優先級[10 ] task-線程編號- 11
finish PrioritizedTaskProducer
toString() = 線程優先級[10 ] task-線程編號- 12
toString() = 線程優先級[1 ] task-線程編號- 14
toString() = 線程優先級[0 ] task-線程編號- 13
(summary() = 線程編號：0: 線程優先級：8)
(summary() = 線程編號：1: 線程優先級：5)
(summary() = 線程編號：2: 線程優先級：3)
(summary() = 線程編號：3: 線程優先級：1)
(summary() = 線程編號：4: 線程優先級：1)--------------------我是換行符--------------------
(summary() = 線程編號：5: 線程優先級：9)
(summary() = 線程編號：6: 線程優先級：8)
(summary() = 線程編號：7: 線程優先級：0)
(summary() = 線程編號：8: 線程優先級：2)
(summary() = 線程編號：9: 線程優先級：7)--------------------我是換行符--------------------
(summary() = 線程編號：10: 線程優先級：10)
(summary() = 線程編號：11: 線程優先級：10)
(summary() = 線程編號：12: 線程優先級：10)
(summary() = 線程編號：13: 線程優先級：0)
(summary() = 線程編號：14: 線程優先級：1)--------------------我是換行符--------------------
(summary() = 線程編號：15: 線程優先級：-1)toString() = 線程優先級[-1 ] task-線程編號- 15 calling shutdownNow() 關閉線程池
finish PrioritizedTaskConsumer
*/
代碼解說：
【21.7.5】使用 ScheduledExecutor的溫室控制器 （干貨——ScheduledExecutor計劃調度器可用于系統后臺的周期性或定時跑批，如每日凌晨跑批，采用cron 表達式）
/*** 溫室調度器（定時調度，非常重要） * page 730 */
public class GreenHouseScheduler {private volatile boolean light = false; private volatile boolean water = false;private String thermostat = "day"; // 調溫器-day-白天 ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(10); // 調度線程池/*** 創建并執行一次操作，該操作在給定的延遲后變為啟用狀態。*/public void schedcule(Runnable event, long delay) { // 調度方法 scheduler.schedule(event, delay, TimeUnit.MILLISECONDS); }public void repeat(Runnable event, long initialDelay, long period) { // 重復執行任務 /*創建并執行一個周期性操作，該操作將在給定的初始延遲后首先啟用，然后在給定的周期內啟用； 即執行將在initialDelay，initialDelay + period，initialDelay + 2 *期間等之后開始，如果任務的任何執行遇到異常，則后續執行將被抑制，否則，該任務將僅通過執行者的取消或終止而終止 。如果此任務的任何執行花費的時間超過其周期，則后續執行可能會延遲開始，但不會同時執行。*/scheduler.scheduleAtFixedRate(event, initialDelay, period, TimeUnit.MILLISECONDS); } class LightOn implements Runnable { // 開燈任務 @Overridepublic void run() {light = true; System.out.println("turning on lights");}}class LightOff implements Runnable {// 關燈燈任務@Overridepublic void run() {light = false; System.out.println("turning off lights");}}class WaterOn implements Runnable { // 澆水任務 @Overridepublic void run() {water = true; System.out.println("turning on water");}}class WaterOff implements Runnable { // 停止澆水任務 @Overridepublic void run() {water = false; System.out.println("turning off water");}}class ThermostatNight implements Runnable { // 把溫度調整為夜晚模式 @Overridepublic void run() {System.out.println("Thermostat to night setting ");setThermostat("night");}}class ThermostatDay implements Runnable { // 把溫度調整為夜晚白晝模式 @Overridepublic void run() {System.out.println("Thermostat to day setting ");setThermostat("day");}}class Bell implements Runnable { // 鐘聲響起任務 @Overridepublic void run() {System.out.println("bing"); }}class Terminate implements Runnable { // 終止任務，關閉線程池 @Overridepublic void run() {System.out.println("terminating");scheduler.shutdownNow(); // 關閉線程池 // 聲明 同步列表 data // List<DataPoint> data = Collections.synchronizedList(new ArrayList<DataPoint>()); new Thread() {public void run() {for (DataPoint d : data) { // 遍歷并打印數據點 System.out.println(d);}}}.start(); }}static class DataPoint { // 數據點 Calendar time; // 日期float temperature; // 溫度float humidity; // 濕度 public DataPoint(Calendar time, float temperature, float humidity) {this.time = time;this.temperature = temperature;this.humidity = humidity;}@Overridepublic String toString() {return time.getTime() + String.format(", temperature: %1$.1f humidity: %2$.2f", temperature, humidity); }}private Calendar lastTime = Calendar.getInstance();{lastTime.set(Calendar.MINUTE, 30);lastTime.set(Calendar.SECOND, 30); }private float lastTemp = 65.0f; // 最新溫度private int tempDirection = 1; // 溫度方向， 正private float lastHumidity = 50.0f; // 最新濕度 private int humidityDirection = 1; // 濕度方向，正 private Random rand = new Random(47);/*Collections.synchronizedList - 返回由指定列表支持的同步（線程安全）列表。 為了保證串行訪問，至關重要的是，對后備列表的所有訪問都必須通過返回的列表來完成。*/List<DataPoint> data = Collections.synchronizedList(new ArrayList<DataPoint>());// 收集數據任務 class CollectData implements Runnable {@Override public void run() {System.out.println("collecting data");synchronized (GreenHouseScheduler.this) {lastTime.set(Calendar.MINUTE, lastTime.get(Calendar.MINUTE)+ 30); // 設置最新時間 if (rand.nextInt(5) == 4) {tempDirection *= -1; // 改變溫度方向， 正數加， 負數減 }lastTemp += tempDirection * (1.0f + rand.nextFloat()); // 最新溫度 if (rand.nextInt(5) == 4) {humidityDirection *= -1; // 改變濕度方向， 正數加， 負數減}lastHumidity += humidityDirection * rand.nextFloat(); // 計算最新濕度data.add(new DataPoint((Calendar)lastTime.clone(), lastTemp, lastHumidity)); // 添加數據信息 }}}// main public static void main(String[] args) {GreenHouseScheduler scheduler = new GreenHouseScheduler(); // 溫室調度器 scheduler.schedcule(scheduler.new Terminate(), 2000);// 3000毫秒后執行Terminate任務，關閉線程池 scheduler.repeat(scheduler.new Bell(), 0, 1000); // 0毫秒，0+1000毫秒，0+1000+1000毫秒.... 后執行Bell任務 scheduler.repeat(scheduler.new ThermostatNight(), 0, 1000); // 0毫秒，0+1000毫秒，0+1000+1000毫秒.... 后執行ThermostatNight任務scheduler.repeat(scheduler.new LightOn(), 0, 200); // 0毫秒，0+200毫秒，0+200+200.... 后執行LightOn任務scheduler.repeat(scheduler.new LightOff(), 0, 200); // 0毫秒，0+2毫秒，0+200+200毫秒.... 后執行LightOff任務scheduler.repeat(scheduler.new WaterOn(), 0, 200); // 0毫秒，0+200毫秒，0+200+200毫秒.... 后執行WaterOn任務scheduler.repeat(scheduler.new WaterOff(), 0, 200); // 0毫秒，0+200毫秒，0+200+200毫秒.... 后執行WaterOff任務scheduler.repeat(scheduler.new ThermostatDay(), 0, 200); // 0毫秒，0+200毫秒，0+200+200毫秒.... 后執行ThermostatDay 任務scheduler.repeat(scheduler.new CollectData(), 500, 500); // 0毫秒，0+500毫秒，0+500+500毫秒.... 后執行CollectData任務}public String getThermostat() { return thermostat;}public void setThermostat(String thermostat) {this.thermostat = thermostat;}
}
/*
bing
Thermostat to night setting
turning on lights
turning off lights
turning on water
turning off water
Thermostat to day setting
turning on lights
turning off lights
turning on water
Thermostat to day setting
turning off water
turning on lights
turning on water
turning off lights
Thermostat to day setting
turning off water
collecting data
turning on lights
turning off lights
turning off water
turning on water
Thermostat to day setting
turning on lights
turning off lights
turning on water
Thermostat to day setting
turning off water
bing
Thermostat to night setting
turning on lights
turning off lights
turning off water
collecting data
turning on water
Thermostat to day setting
turning on lights
Thermostat to day setting
turning off water
turning on water
turning off lights
turning on lights
turning off water
Thermostat to day setting
turning off lights
turning on water
collecting data
turning on lights
turning off lights
turning on water
turning off water
Thermostat to day setting
turning on lights
turning on water
turning off lights
turning off water
Thermostat to day setting
bing
turning on lights
Thermostat to day setting
Thermostat to night setting
terminating
collecting data
turning off water
turning on water
turning off lights
Sun Jul 05 17:00:30 CST 2020, temperature: 66.4 humidity: 50.05
Sun Jul 05 17:30:30 CST 2020, temperature: 68.0 humidity: 50.47
Sun Jul 05 18:00:30 CST 2020, temperature: 69.7 humidity: 51.42
Sun Jul 05 18:30:30 CST 2020, temperature: 70.8 humidity: 50.87
*/
public class ScheduledThreadPoolExecutorextends ThreadPoolExecutorimplements ScheduledExecutorService
代碼解說：volatile和 synchronized 都得到了應用，以防止任務之間的相互干涉。在持有 DataPoint的List中的所有方法都是 synchronized，這是因為在List被創建時，使用了 Collections工具 synchronizedList();
【21.7.6】Semaphore 而計數信號量允許n個任務同時訪問這個資源；
/*** Pool對象池-通過信號量Semaphore來管理 Semaphore-信號量(他管理著數量有限的對象，當要使用對象時可以簽出他們，* 在用戶使用完成時，將他們簽回) page 733*/
public class Pool<T> {private int size;private List<T> items = new ArrayList<>();private volatile boolean[] checkOut;private Semaphore available; // 計數信號量public Pool(Class<T> classObj, int size) {this.size = size;checkOut = new boolean[size];available = new Semaphore(size, true);for (int i = 0; i < size; i++) {try {items.add(classObj.newInstance());} catch (Exception e) {throw new RuntimeException(e);}}}// 如果沒有任何信號量許可證可用， available將阻塞調用。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 (!checkOut[i]) { // 簽出狀態為 false，則返回該對象checkOut[i] = true;return items.get(i);}}return null;}private synchronized boolean releaseItem(T item) { // 釋放對象int index = items.indexOf(item);if (index == -1)return false;if (checkOut[index]) {checkOut[index] = false;return true;}return false;}
}
/*** 創建代價高昂的對象類型，構造器運行起來很耗時* page 734 */
public class Fat {private volatile double d; private static int counter = 0;private final int id = counter++;public Fat() {for (int i = 0; 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; }
}
?
/*** 信號量演示（非常重要的荔枝） * page 734 * (* 一旦池中的所有對象被簽出，semaphore 將不允許執行任何簽出操作； * blocked的run()方法因此會被阻塞， 2秒鐘后，cancel()方法會被調用， 以此來掙脫Future的束縛* )*/
public class SemaphoreDemo {final static int size = 5; public static void main(String[] args) throws InterruptedException {final Pool<Fat> pool = new Pool<>(Fat.class, size); // 對象池，通過信號量來管理 ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 for (int i = 0; i < size; i++) {exec.execute(new CheckOutTask<Fat>(pool)); // 運行簽出任務 }System.out.println("all checkout tasks created");List<Fat> list = new ArrayList<>();for (int i = 0; i < size; i++) {Fat f = pool.checkOut(); // 簽出對象 System.out.println(i + " : main() thread check out");f.operation(); // "fat id: " + id list.add(f);}Future<?> blocked = exec.submit(new Runnable() {@Overridepublic void run() {try {pool.checkOut();// 開啟單個線程，簽出對象 } catch (InterruptedException e) {System.out.println("checkout() interrupted. "); }}});TimeUnit.SECONDS.sleep(2); //睡眠2秒 blocked.cancel(true); // 嘗試取消執行此任務 for (Fat f : list) {pool.checkIn(f);}for (Fat f : list) { // 冗余的簽入將被pool 忽略 pool.checkIn(f);} exec.shutdown(); // 關閉線程池 }
}
// 創建一個任務，先簽出Fat對象，持有一段時間后，再簽入，以此來測試Pool這個類
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; }@Overridepublic void run() {try {T item = pool.checkOut(); // 簽出對象，獲取信號量許可證System.out.println(this + " checked out " + item);System.out.println(this + " checking in " + item);pool.checkIn(item); // 簽入對象，釋放許可證歸還給信號量 } catch (InterruptedException e) {System.out.println("CheckOutTask interrupted");}}@Overridepublic String toString() {return "checkout task " + id + " "; }
}
/*checkout task 1 checked out fat id: 1
checkout task 4 checked out fat id: 4
checkout task 4 checking in fat id: 4
all checkout tasks created
checkout task 3 checked out fat id: 3
checkout task 3 checking in fat id: 3
checkout task 0 checked out fat id: 0
checkout task 0 checking in fat id: 0
checkout task 2 checked out fat id: 2
checkout task 2 checking in fat id: 2
0 : main() thread check out
fat id: 4
1 : main() thread check out
checkout task 1 checking in fat id: 1
fat id: 0
2 : main() thread check out
fat id: 1
3 : main() thread check out
fat id: 2
4 : main() thread check out
fat id: 3
checkout() interrupted. */
【21.7.7】 Exchanger 交換器
/*** 任務交換演示 * page 736*/
public class ExchangerDemo {static int size = 5; static int delay = 5; // secondspublic static void main(String[] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 Exchanger<List<Fat>> xc = new Exchanger<>(); // 交換器 List<Fat> producerList = new CopyOnWriteArrayList<>(); // 生產者列表 List<Fat> consumerList = new CopyOnWriteArrayList<>(); // 消費者列表 exec.execute(new ExchangerProducer<Fat>(xc, BasicGenerator.create(Fat.class), producerList));// 運行交換任務生產者 exec.execute(new ExchangerConsumer<Fat>(xc, consumerList)); // 運行交換任務消費者 TimeUnit.MILLISECONDS.sleep(5);exec.shutdownNow(); }
}
// 任務交換生產者
class ExchangerProducer<T> implements Runnable {private Generator<T> generator;private Exchanger<List<T>> exchanger;private List<T> holder; ExchangerProducer(Exchanger<List<T>> exchanger, Generator<T> gen, List<T> holder) {this.exchanger = exchanger; this.generator = gen; this.holder = holder; }@Overridepublic void run() {try {while (!Thread.interrupted()) {for (int i = 0; i < ExchangerDemo.size; i++) {holder.add(generator.next()); } System.out.println("producer, before exchange, holder = " + holder);// exchange方法-等待另一個線程到達此交換點（除非當前線程被中斷），然后將給定對象傳送給它，并在返回時接收其對象。holder = exchanger.exchange(holder);System.out.println("producer, after exchange, holder = " + holder);}} catch (InterruptedException e) {System.out.println("ExchangerProducer interrupted");}}
}
// 任務交換消費者
class ExchangerConsumer<T> implements Runnable {private Exchanger<List<T>> exchanger;private List<T> holder; private volatile T value ;ExchangerConsumer(Exchanger<List<T>> ex, List<T> holder) {this.exchanger = ex; this.holder = holder; }@Overridepublic void run() {try {while(!Thread.interrupted()) {System.out.println("consumer, before exchange, holder = " + holder);// exchange方法-等待另一個線程到達此交換點（除非當前線程被中斷），然后將給定對象傳送給它，并在返回時接收其對象。holder = exchanger.exchange(holder);System.out.println("consumer, after exchange, holder = " + holder);for (T x : holder) {value = x; holder.remove(x); }}} catch (InterruptedException e) {System.out.println(" ExchangerConsumer interrupted");}System.out.println("final value: " + value);}
}
/*
consumer, before exchange, holder = []
producer, before exchange, holder = [fat id: 0, fat id: 1, fat id: 2, fat id: 3, fat id: 4]
producer, after exchange, holder = []
consumer, after exchange, holder = [fat id: 0, fat id: 1, fat id: 2, fat id: 3, fat id: 4]
consumer, before exchange, holder = []
producer, before exchange, holder = [fat id: 5, fat id: 6, fat id: 7, fat id: 8, fat id: 9]
producer, after exchange, holder = []
consumer, after exchange, holder = [fat id: 5, fat id: 6, fat id: 7, fat id: 8, fat id: 9]
consumer, before exchange, holder = []
producer, before exchange, holder = [fat id: 10, fat id: 11, fat id: 12, fat id: 13, fat id: 14]
producer, after exchange, holder = []
consumer, after exchange, holder = [fat id: 10, fat id: 11, fat id: 12, fat id: 13, fat id: 14]
consumer, before exchange, holder = []
producer, before exchange, holder = [fat id: 15, fat id: 16, fat id: 17, fat id: 18, fat id: 19]
producer, after exchange, holder = []
consumer, after exchange, holder = [fat id: 15, fat id: 16, fat id: 17, fat id: 18, fat id: 19]
consumer, before exchange, holder = []ExchangerConsumer interrupted
final value: fat id: 19
producer, before exchange, holder = [fat id: 20, fat id: 21, fat id: 22, fat id: 23, fat id: 24]
ExchangerProducer interrupted */
代碼解說：
（干貨——引入了CopyOnWriteArrayList，允許在列表被遍歷時調用remove()方法，而不拋出異常 ModificationExcetpion）
【21.8】仿真
【21.9】性能調優
1、比較 synchronize 與 Lock的性能
/*** page 748* 啟動單個任務比較 synchronized關鍵字和 Lock和Atomic類的區別 */
public class SimpleMicroBenchmark {static long test(Incrementable incr) {long start = System.nanoTime();for (long i=0; i < 100000000L; i++) {incr.increment();}return System.nanoTime() - start; }public static void main(String[] args) {long synchTime = test(new SynchronizingTest());long lockTime = test(new LockingTest());System.out.printf("synchronized 花費多少納秒: %1$10d \n", synchTime); // 203974196 System.out.printf("lock 花費多少納秒: %1$10d \n", lockTime); // 164559713 System.out.printf("lock/synchronized: %1$.3f\n", lockTime/(double)synchTime); // 0.807 }
}
abstract class Incrementable {protected long counter = 0;public abstract void increment();
}
// synchronized 關鍵字性能測試
class SynchronizingTest extends Incrementable {public synchronized void increment() {++counter;}
}
// lock鎖性能測試
class LockingTest extends Incrementable {private Lock lock = new ReentrantLock();public void increment() {lock.lock();try {++counter;} finally {lock.unlock();}}
}
/*
synchronized 花費多少納秒: 2166063742
lock 花費多少納秒: 1725775548
lock/synchronized: 0.797
*/
以上代碼是單線程，不具體代表性；我們需要構建復雜程序，或多個任務來測試；
【代碼-SynchronizationComparison】
/*** page 749 * 啟動多個任務測試 synchronize， lock ， Atomic類性能 （經過驗證，Atomic原子類同步性能最佳 ）*/
public class SynchronizationComparison{static BaseLine baseLine = new BaseLine(); // 基線 static SynchronizedTest synch = new SynchronizedTest(); // synchronize測試 static LockTest lock = new LockTest(); // ReentrantLock 可重入鎖測試 static AtomicTest atomic = new AtomicTest(); // 原子類測試 static void test() { System.out.println("=============================="); System.out.printf("%-12s : %13d\n", "Cycles", Accumulator.cycles);// 運行任務測試總時長baseLine.timedTest(); // 基本測試，無任何同步方法 synch.timedTest(); // synchronize同步 lock.timedTest(); // lock 同步 atomic.timedTest(); // Atomic 類同步 // 比較兩種模擬器性能 Accumulator.report(synch, baseLine); // 1.65 Accumulator.report(lock, baseLine); // 2.31 Accumulator.report(atomic, baseLine); // 0.91 Accumulator.report(synch, lock); // 0.71 Accumulator.report(synch, atomic); // 1.82 Accumulator.report(lock, atomic); // 2.54 }public static void main(String[] args) {int iteration = 5; System.out.println("warm up");baseLine.timedTest();for (int i = 0; i < iteration; i++) {test();Accumulator.cycles *= 2; }Accumulator.exec.shutdown(); }
}
abstract class Accumulator { // 模擬器 public static long cycles = 50000L; // 循環次數 private static final int N = 4; public static ExecutorService exec = Executors.newFixedThreadPool(N*2); // 線程池
// CyclicBarrier也叫同步屏障，在JDK1.5被引入，可以讓一組線程達到一個屏障時被阻塞
// ，直到最后一個線程達到屏障時，所以被阻塞的線程才能繼續執行。 private static CyclicBarrier barrier = new CyclicBarrier(N*2+1); // 同步屏障 protected volatile int index = 0;protected volatile long value = 0;protected long duration = 0; // 持續時間 protected String id = "error";protected final static int SIZE = 100000 ;protected static int[] preLoaded = new int[SIZE]; static {Random rand = new Random(47); // 隨機數發生器 for (int i=0; i< SIZE; i++) {preLoaded[i] = rand.nextInt(); // 預加載}}public abstract void accumulate(); // 模擬方法，抽象，由子類實現 public abstract long read(); // 讀取方法 private class Modifier implements Runnable { // 修改器 @Overridepublic void run() { // 模板方法模式，由子類提供實現 for(long i=0; i<cycles; i++) {accumulate(); // 調用模擬方法 }try {barrier.await(); // 屏障阻塞，直到給定數量的線程都等待為止 } catch (Exception e) {throw new RuntimeException(e);}}}private class Reader implements Runnable { // 讀取器 private volatile long value; @Overridepublic void run() {for(long i=0; i<cycles; i++) {value = read(); }try {barrier.await(); // 屏障阻塞，直到給定數量的線程都等待為止 } catch (Exception e) {throw new RuntimeException(e);}}}public void timedTest() { // 時間測試 long start = System.nanoTime(); for (int i=0; i<N; i++) {exec.execute(new Modifier()); // 執行修改器 exec.execute(new Reader()); // 執行讀取器 }try {// 程序中必須有一個 CyclicBarrier， 因為需要確保所有任務在聲明每個測試完成之前都已經完成 barrier.await(); // 屏障阻塞，直到給定數量的線程都等待為止 } catch (Exception e) {throw new RuntimeException(e);}duration = System.nanoTime() - start; // 總時長 System.out.printf("%-13s:%13d\n", id, duration);}public static void report(Accumulator acc1, Accumulator acc2) { // 報告， System.out.printf("%-22s: %.2f\n", acc1.id + "/" + acc2.id, acc1.duration/(double)acc2.duration); }
}
class BaseLine extends Accumulator { // 基本測試，無任何同步方法 {id = "baseline"; }@Override public void accumulate() {if (index >= SIZE-1) index = 0;value += preLoaded[(index++)%SIZE];}@Overridepublic long read() {return value;}
}
class SynchronizedTest extends Accumulator { // synchronize同步 測試 {id = "Synchronized"; }@Overridepublic synchronized void accumulate() {if (index >= SIZE-1) index = 0;value += preLoaded[index++];}@Overridepublic long read() {return value;}
}
class LockTest extends Accumulator { // ReentrantLock可重入鎖同步 測試 {id = "lock";}private Lock lock = new ReentrantLock(); @Overridepublic void accumulate() {lock.lock();try {if (index >= SIZE-1) index = 0;value += preLoaded[index++];} finally {lock.unlock();}}@Overridepublic long read() {lock.lock();try {return value; } finally {lock.unlock(); }}
}
class AtomicTest extends Accumulator { // Atomic原子類同步 測試 {id = "atomic";}private AtomicInteger index = new AtomicInteger(0);private AtomicLong value = new AtomicLong(0);@Overridepublic void accumulate() {int i = index.getAndIncrement();value.getAndAdd(preLoaded[i%SIZE]);if (++i >= SIZE-1) {index.set(0);}}@Overridepublic long read() {return value.get();}
}
/*
warm up
baseline : 12811667
==============================
Cycles : 50000
baseline : 10401913
Synchronized : 18486698
lock : 26550332
atomic : 8931189
Synchronized/baseline : 1.78
lock/baseline : 2.55
atomic/baseline : 0.86
Synchronized/lock : 0.70
Synchronized/atomic : 2.07
lock/atomic : 2.97
==============================
Cycles : 100000
baseline : 18458982
Synchronized : 28172394
lock : 36321361
atomic : 14323233
Synchronized/baseline : 1.53
lock/baseline : 1.97
atomic/baseline : 0.78
Synchronized/lock : 0.78
Synchronized/atomic : 1.97
lock/atomic : 2.54
==============================
Cycles : 200000
baseline : 36408153
Synchronized : 50424697
lock : 71790482
atomic : 28702992
Synchronized/baseline : 1.38
lock/baseline : 1.97
atomic/baseline : 0.79
Synchronized/lock : 0.70
Synchronized/atomic : 1.76
lock/atomic : 2.50
==============================
Cycles : 400000
baseline : 68541253
Synchronized : 103632938
lock : 144097706
atomic : 53405164
Synchronized/baseline : 1.51
lock/baseline : 2.10
atomic/baseline : 0.78
Synchronized/lock : 0.72
Synchronized/atomic : 1.94
lock/atomic : 2.70
==============================
Cycles : 800000
baseline : 137235667
Synchronized : 180808536
lock : 283742763
atomic : 108986327
Synchronized/baseline : 1.32
lock/baseline : 2.07
atomic/baseline : 0.79
Synchronized/lock : 0.64
Synchronized/atomic : 1.66
lock/atomic : 2.60
*/
代碼解說：程序中有一個CyclicBarrier 循環屏障，因為我們希望確保所有的任務在聲明每個測試完成之前都已經完成了；
【互斥技術總結】 （干貨——Atomic類的使用場景） （干貨——推薦使用 synchronize進行并發控制）
【21.9.2】免鎖容器 CopyOnWriteArrayList好處是當多個迭代器同時遍歷和修改這個列表時，不會拋出 ConcurrentModificationException；
4、比較并發控制的list容器（干貨——測試并發編程下的list性能： CopyOnWriteArrayList性能 優于? SynchronizedList）
/*** 性能測試器* page 756* @param <C>*/
public abstract class Tester<C> {static int testReps = 10;static int testCycles = 10;static int containerSize = 10; // 容器大小/** 抽象方法-初始化容器 */abstract C containerInitializer();/** 抽象方法-開啟讀取和寫入任務 */abstract void startReadersAndWriters();C testContainer;String testId;/** 讀取線程個數 */int nReaders;/** 寫入線程個數 */ int nWriters;volatile long readResult = 0;volatile long readTime = 0;volatile long writeTime = 0;/*CountDownLatch的作用也是如此，在構造CountDownLatch的時候需要傳入一個整數n，* 在這個整數“倒數”到0之前，主線程需要等待在門口，而這個“倒數”過程則是由各個執行線程驅動的，* 每個線程執行完一個任務“倒數”一次?？偨Y來說，CountDownLatch的作用就是等待其他的線程都執行完任務，* 必要時可以對各個任務的執行結果進行匯總，然后主線程才繼續往下執行。*/CountDownLatch endLatch; // latch-門栓 static ExecutorService exec = Executors.newCachedThreadPool(); // 線程池 Integer[] writeData;/** 構造器 */ Tester(String testId, int nReaders, int nWriters) {this.testId = testId + " , " + nReaders + " reader thread, " + nWriters + " writer thread"; this.nReaders = nReaders;this.nWriters = nWriters; writeData = Generated.array(Integer.class, new RandomGenerator.Integer(), containerSize);for (int i=0; i< testReps; i++) {runTest();readTime = 0 ;writeTime = 0;}}void runTest() {endLatch = new CountDownLatch(nReaders+ nWriters);testContainer = containerInitializer();startReadersAndWriters();try {endLatch.await(); // 門栓等待，直到所有線程都執行完成 } catch (InterruptedException ex) {System.out.println("endLatch interrupted");}System.out.printf("%-100s %14d %14d\n", testId, readTime, writeTime);if (readTime !=0 && writeTime != 0) {System.out.printf("%-100s %14d\n", "readTime + writeTime = ", readTime + writeTime);}}abstract class TestTask implements Runnable {/** 開啟線程，運行test方法 */abstract void test();/** 存放結果，在synchronzid 靜態塊里執行 */abstract void putResults();long duration; public void run() {long startTime = System.nanoTime();test();duration = System.nanoTime() - startTime;synchronized (Tester.this) {putResults();}endLatch.countDown(); // 門栓減1，直到減為0，則門栓不等待 }}public static void initMain(String[] args) {testReps = new Integer(3);testCycles = new Integer(3);containerSize = new Integer(3);System.out.printf("%-100s %14s %14s\n", "type", "readTime", "write Time"); }
}
【代碼——ListComparisons——比較列表】
/*** 測試并發編程下的list性能： CopyOnWriteArrayList性能 優于 SynchronizedList* page 758 */
public class ListComparisons {public static void main(String[] args) {Tester.initMain(null);new SynchronizedArrayListTest(10, 0);new SynchronizedArrayListTest(9, 1);new SynchronizedArrayListTest(5, 5);new CopyOnWriteArrayListTest(10, 0);new CopyOnWriteArrayListTest(9, 1);new CopyOnWriteArrayListTest(5, 5);Tester.exec.shutdown();}
}
/** List測試類 */
abstract class ListTest extends Tester<List<Integer>> {ListTest(String testId, int nReaders, int nWriters) {super(testId, nReaders, nWriters);}class Reader extends TestTask { // 讀取任務 long result = 0;void test() {for (int i = 0; i < testCycles; i++) {for (int j = 0; j < containerSize; j++) {result += testContainer.get(j);}}}void putResults() {readResult += result;readTime += duration; }}class Writer extends TestTask { // 寫入任務 @Overridevoid test() {for (int i = 0; i < testCycles; i++) {for (int j = 0; j < containerSize; j++) {testContainer.set(i, writeData[j]);}}}@Overridevoid putResults() {writeTime += duration; }}/** 運行讀取任務和寫入任務 */void startReadersAndWriters() {for (int i = 0; i < nReaders; i++) {exec.execute(new Reader());}for (int i = 0; i < nWriters; i++) {exec.execute(new Writer());}}
}
/** 同步list-SynchronizedList*/
class SynchronizedArrayListTest extends ListTest {List<Integer> containerInitializer() {return Collections.synchronizedList(new ArrayList<Integer>(new CountingIntegerList(containerSize)));}SynchronizedArrayListTest(int nreaders, int nwriters) {super("synched arraylist", nreaders, nwriters);}
}
/** 同步list-CopyOnWriteArrayList*/
class CopyOnWriteArrayListTest extends ListTest {@OverrideList<Integer> containerInitializer() {/*CopyOnWriteArrayList好處是當多個迭代器同時遍歷和修改這個列表時* ，不會拋出 ConcurrentModificationException；*/ return new CopyOnWriteArrayList<Integer>(new CountingIntegerList(containerSize));} CopyOnWriteArrayListTest(int nreaders, int nwriters) {super("CopyOnWriteArrayListTest", nreaders, nwriters);}
}
/*
type readTime write Time
t = null
t = null
t = null
synched arraylist , 10 reader thread, 0 writer thread 86062 0
synched arraylist , 10 reader thread, 0 writer thread 140764 0
synched arraylist , 10 reader thread, 0 writer thread 535339 0
t = null
t = null
t = null
synched arraylist , 9 reader thread, 1 writer thread 238497 20422
readTime + writeTime = 258919
synched arraylist , 9 reader thread, 1 writer thread 188900 4376
readTime + writeTime = 193276
synched arraylist , 9 reader thread, 1 writer thread 192182 3647
readTime + writeTime = 195829
t = null
t = null
t = null
synched arraylist , 5 reader thread, 5 writer thread 86791 74393
readTime + writeTime = 161184
synched arraylist , 5 reader thread, 5 writer thread 605721 540446
readTime + writeTime = 1146167
synched arraylist , 5 reader thread, 5 writer thread 39385 76216
readTime + writeTime = 115601
t = null
t = null
t = null
CopyOnWriteArrayListTest , 10 reader thread, 0 writer thread 63456 0
CopyOnWriteArrayListTest , 10 reader thread, 0 writer thread 48866 0
CopyOnWriteArrayListTest , 10 reader thread, 0 writer thread 44126 0
t = null
t = null
t = null
CopyOnWriteArrayListTest , 9 reader thread, 1 writer thread 30997 35738
readTime + writeTime = 66735
CopyOnWriteArrayListTest , 9 reader thread, 1 writer thread 71475 21516
readTime + writeTime = 92991
CopyOnWriteArrayListTest , 9 reader thread, 1 writer thread 24434 21151
readTime + writeTime = 45585
t = null
t = null
t = null
CopyOnWriteArrayListTest , 5 reader thread, 5 writer thread 19692 680843
readTime + writeTime = 700535
CopyOnWriteArrayListTest , 5 reader thread, 5 writer thread 48503 622496
readTime + writeTime = 670999
CopyOnWriteArrayListTest , 5 reader thread, 5 writer thread 14587 756695
readTime + writeTime = 771282*/
代碼解說： synchronized ArrayList 無論讀者和寫入者的數量是多少，都具有大致相同的性能——讀取者與其他讀取者競爭鎖的方式與寫入者相同。但 CopyOnWriteArrayList 在沒有寫入者時，速度會快很多。通過測試，CopyOnWriteArrayList 性能優于 synchronized list，對列表寫入的影響并沒有超過短期同步整個列表的影響。
5、比較并發控制的map容器性能（干貨——測試并發編程下的Map性能： CurrentHashMap性能優于 synchronizedHashMap）
/*** 測試并發編程下的Map性能： CurrentHashMap性能優于synchronizedHashMap * page 758*/
public class MapComparisons {public static void main(String[] args) {Tester.initMain(null); new SynchronizedHashMapTest(10, 0);new SynchronizedHashMapTest(9, 1);new SynchronizedHashMapTest(5, 5);new ConcurrentHashMapTest(10, 0);new ConcurrentHashMapTest(9, 1);new ConcurrentHashMapTest(5, 5);Tester.exec.shutdown();}
}
/** Map測試 */
abstract class MapTest extends Tester<Map<Integer, Integer>> {MapTest(String testId, int nReaders, int nWriters) {super(testId, nReaders, nWriters); }/** 讀取器 */class Reader extends TestTask {long result = 0;void test() {for (int i = 0; i < testCycles; i++) {for (int j = 0; j < containerSize; j++) {result += testContainer.get(j);}}}@Overridevoid putResults() {readResult += result; readTime += duration; }}/** 寫入器 */ class Writer extends TestTask {long result = 0;void test() {for (int i = 0; i < testCycles; i++) {for (int j = 0; j < containerSize; j++) {testContainer.put(j, writeData[j]);}}}@Overridevoid putResults() {writeTime += duration; }}/** 運行讀取與寫入任務 */ void startReadersAndWriters() {for (int i = 0; i < nReaders; i++) {exec.execute(new Reader());}for (int i = 0; i < nWriters; i++) {exec.execute(new Writer());}}
}
/** 同步塊SynchronizedHashMap*/
class SynchronizedHashMapTest extends MapTest {Map<Integer, Integer> containerInitializer() {return Collections.synchronizedMap(new HashMap<Integer, Integer>(MapData.map(new CountingGenerator.Integer(), new CountingGenerator.Integer(), containerSize)));}SynchronizedHashMapTest(int nreaders, int nwriters) {super("SynchronizedHashMapTest", nreaders, nwriters);}
}
/** 同步HashMap */
class ConcurrentHashMapTest extends MapTest {Map<Integer, Integer> containerInitializer() {return new ConcurrentHashMap<Integer, Integer>( MapData.map(new CountingGenerator.Integer(), new CountingGenerator.Integer(), containerSize));}ConcurrentHashMapTest(int nreaders, int nwriters) {super("ConcurrentHashMapTest", nreaders, nwriters);}
}
/*
type readTime write Time
t = null
t = null
t = null
SynchronizedHashMapTest , 10 reader thread, 0 writer thread 207497 0
SynchronizedHashMapTest , 10 reader thread, 0 writer thread 487569 0
SynchronizedHashMapTest , 10 reader thread, 0 writer thread 253084 0
t = null
t = null
t = null
SynchronizedHashMapTest , 9 reader thread, 1 writer thread 126177 31361
readTime + writeTime = 157538
SynchronizedHashMapTest , 9 reader thread, 1 writer thread 131281 25892
readTime + writeTime = 157173
SynchronizedHashMapTest , 9 reader thread, 1 writer thread 98095 7658
readTime + writeTime = 105753
t = null
t = null
t = null
SynchronizedHashMapTest , 5 reader thread, 5 writer thread 37559 56889
readTime + writeTime = 94448
SynchronizedHashMapTest , 5 reader thread, 5 writer thread 49961 74393
readTime + writeTime = 124354
SynchronizedHashMapTest , 5 reader thread, 5 writer thread 64910 122531
readTime + writeTime = 187441
t = null
t = null
t = null
ConcurrentHashMapTest , 10 reader thread, 0 writer thread 46679 0
ConcurrentHashMapTest , 10 reader thread, 0 writer thread 48867 0
ConcurrentHashMapTest , 10 reader thread, 0 writer thread 41572 0
t = null
t = null
t = null
ConcurrentHashMapTest , 9 reader thread, 1 writer thread 48134 8023
readTime + writeTime = 56157
ConcurrentHashMapTest , 9 reader thread, 1 writer thread 55795 8023
readTime + writeTime = 63818
ConcurrentHashMapTest , 9 reader thread, 1 writer thread 28080 7294
readTime + writeTime = 35374
t = null
t = null
t = null
ConcurrentHashMapTest , 5 reader thread, 5 writer thread 17506 60170
readTime + writeTime = 77676
ConcurrentHashMapTest , 5 reader thread, 5 writer thread 17506 41207
readTime + writeTime = 58713
ConcurrentHashMapTest , 5 reader thread, 5 writer thread 14221 58348
readTime + writeTime = 72569*/
【21.9.3】樂觀加鎖
/*** page 760 * 樂觀加鎖測試 */
public class FastSimulation {/** 元素個數 */static final int N_ELEMENTS = 100;/** 每個元素的基因個數 */static final int N_GENES = 30;/** 進化者 */static final int N_EVOLVERS = 50;// 進化者 /** 網格 */ static final AtomicInteger[][] GRID = new AtomicInteger[N_ELEMENTS][N_GENES]; static Random rand = new Random(47);static class Evolver implements Runnable { // 進化者任務 @Overridepublic void run() {while(!Thread.interrupted()) { // 如果線程不中斷，重復執行 int element = rand.nextInt(N_ELEMENTS); // 獲取元素 for (int i = 0; i < N_GENES; i++) {/** 前一個元素 */int previous = element - 1; if (previous <0) previous = N_ELEMENTS - 1;/** 下一個元素 */ int next = element + 1; if (next >= N_ELEMENTS) next = 0;/** 舊值 */int oldValue = GRID[element][i].get();/** 新值 */ int newValue = oldValue + GRID[previous][i].get() + GRID[next][i].get();newValue /= 3; // Atomic.compareAndSet 方法比較重要：// 比較當前值與傳入的 old 值是否相同，相同則更新為 新值 if (!GRID[element][i].compareAndSet(oldValue, newValue)) {System.out.printf("oldvalue, changed from %d to %d \n", oldValue, newValue); }}}}}public static void main(String[] args) throws Exception {ExecutorService exec = Executors.newCachedThreadPool();for (int i=0; i<N_ELEMENTS; i++) {for (int j = 0; j < N_GENES; j++) {GRID[i][j] = new AtomicInteger(rand.nextInt(1000));}}for (int i = 0; i < N_EVOLVERS; i++) {exec.execute(new Evolver());}exec.shutdown(); }
}
/*
oldvalue, changed from 670 to 676
oldvalue, changed from 352 to 351
oldvalue, changed from 455 to 454
oldvalue, changed from 424 to 423
............
*/
【21.9.4】ReadWriteLock 讀寫鎖
/*** 讀寫任務列表-可重入鎖測試 page 763*/
public class ReaderWriterList<T> {private ArrayList<T> lockedList;/** 可重入讀寫鎖 */private ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);public ReaderWriterList(int size, T initialValue) {/** Collections.nCopies-返回由指定對象（這里是initialValue，初始值）的n(size)個副本組成的不可變列表。* 新分配的數據對象很小（它包含對數據對象的單個引用）。 該方法與List.addAll方法結合使用以增長列表很有用。* 返回的列表是可序列化的。*/lockedList = new ArrayList<T>(Collections.nCopies(size, initialValue));}/** 寫入 */public T set(int index, T element) {Lock wLock = lock.writeLock(); // 獲取寫鎖wLock.lock(); // 加鎖try {System.out.println("寫鎖數量 = " + lock.getWriteHoldCount());return lockedList.set(index, element);} finally {wLock.unlock(); // 解鎖}}public T get(int index) {Lock rLock = lock.readLock(); // 獲取讀鎖rLock.lock();// 加鎖try {if (lock.getReadLockCount() > 1) {System.out.println("讀鎖數量 = " + lock.getReadLockCount());}return lockedList.get(index);} finally {rLock.unlock(); // 解鎖}}public static void main(String[] args) {new ReaderWriterListTest(5, 1);}
}
/*** 讀寫任務列表測試*/
class ReaderWriterListTest {ExecutorService exec = Executors.newCachedThreadPool();private final static int SIZE = 10;private static Random rand = new Random(47);private ReaderWriterList<Integer> list = new ReaderWriterList<Integer>(SIZE, 0);/** 構造器，啟動讀取和寫入任務 */public ReaderWriterListTest(int readers, int writers) {for (int i = 0; i < readers; i++) {exec.execute(new Reader()); // 讀取任務}for (int i = 0; i < writers; i++) {exec.execute(new Writer()); // 寫入任務}}/** 寫入任務 */private class Writer implements Runnable {@Overridepublic void run() {try {for (int i = 0; i < SIZE; i++) {list.set(i, rand.nextInt()); // 獲取寫鎖TimeUnit.MICROSECONDS.sleep(100);}} catch (InterruptedException e) {e.printStackTrace();}System.out.println("writer finished, shutting down");exec.shutdown();}}/** 讀取任務 */private class Reader implements Runnable {@Overridepublic void run() {try {for (int i = 0; i < SIZE; i++) {list.get(i); // 獲取讀鎖TimeUnit.MICROSECONDS.sleep(1);}} catch (InterruptedException e) {e.printStackTrace();}System.out.println("reader finished, shutting down");}}
}
/** 讀鎖數量 = 3 讀鎖數量 = 3 讀鎖數量 = 4 讀鎖數量 = 3 寫鎖數量 = 1 寫鎖數量 = 1 寫鎖數量 = 1 寫鎖數量 = 1 寫鎖數量* = 1 讀鎖數量 = 2 讀鎖數量 = 2 寫鎖數量 = 1 寫鎖數量 = 1 讀鎖數量 = 3 讀鎖數量 = 4 讀鎖數量 = 4 讀鎖數量 = 4* 寫鎖數量 = 1 寫鎖數量 = 1 寫鎖數量 = 1 reader finished, shutting down reader finished,* shutting down writer finished, shutting down reader finished, shutting down* reader finished, shutting down reader finished, shutting down*/
【21.10】活動對象
【看個荔枝】有兩個方法，可以將方法調用排進隊列；
/*** 活動對象演示* page 764 */
public class ActiveObjectDemo {/** 線程池 */ private ExecutorService exec = Executors.newSingleThreadExecutor();private Random rand = new Random(47);/** 暫停方法，睡眠 */private void pause(int factor) {try {TimeUnit.MILLISECONDS.sleep(100+ rand.nextInt(factor));} catch (InterruptedException e) {System.out.println("sleep interrupt");} } /** 調用int方法 */public Future<Integer> calculateInt(final int x, final int y) {return exec.submit(new Callable<Integer>() {public Integer call() {System.out.println("starting x = " + x + ", y = " + y);pause(500);return x+ y ;}});}public Future<Float> calculateFloat(final float x, final float y) {return exec.submit(new Callable<Float>() {public Float call() {System.out.println("starting x = " + x + ", y = " + y);pause(2000);return x+ y; }});}public void shutdown() {exec.shutdown(); }public static void main(String[] args) {ActiveObjectDemo d1 = new ActiveObjectDemo(); /*在計算機中就是當你想要對一塊內存進行修改時，我們不在原有內存塊中進行寫操作* ，而是將內存拷貝一份，在新的內存中進行寫操作* ，寫完之后呢，就將指向原來內存指針指向新的內存，原來的內存就可以被回收掉嘛！*/List<Future<?>> results = new CopyOnWriteArrayList<Future<?>>();for (float f = 0.0f; f < 1.0f; f += 0.2f) {results.add(d1.calculateFloat(f, f));}for (int i = 0; i < 5; i++) {results.add(d1.calculateInt(i, i));}System.out.println("========== all asynch calls made ========== ");int index = 0;while(results.size() >0) { // while 循環，再放一層for循環，因為 可能f.isDone() 為false for (Future<?> f: results) {if (f.isDone()) {try {System.out.println("f.get(" + ++index +") = " + f.get()); } catch (Exception e) {throw new RuntimeException(e);}results.remove(f);}} d1.shutdown(); }}
}
/*
========== all asynch calls made ==========
starting x = 0.0, y = 0.0
f.get(1) = 0.0
starting x = 0.2, y = 0.2
f.get(2) = 0.4
starting x = 0.4, y = 0.4
f.get(3) = 0.8
starting x = 0.6, y = 0.6
f.get(4) = 1.2
starting x = 0.8, y = 0.8
f.get(5) = 1.6
starting x = 0, y = 0
starting x = 1, y = 1
f.get(6) = 0
f.get(7) = 2
starting x = 2, y = 2
f.get(8) = 4
starting x = 3, y = 3
f.get(9) = 6
starting x = 4, y = 4
f.get(10) = 8
*/
代碼解說：使用 CopyOnWriteArrayList 可以移除為了防止 ConcurrentModificationException而復制List的這種需求；小結： 小結2；有個活動對象， 你可以干一下事情：
?
【21.11】總結 【21.11.1】java線程進行并發編碼的基礎知識； 【21.11.2】什么時候使用并發，什么時候應該避免并發非常關鍵，使用并發的原因如下：
【21.11.3】 線程的好處 （干貨——線程的上下文切換是輕量級的，進程的上下文切換是重量級的）
【21.11.4】多線程的缺陷
?
總結
以上是生活随笔 為你收集整理的thinking-in-java(21)并发2 的全部內容，希望文章能夠幫你解決所遇到的問題。
如果覺得生活随笔 網站內容還不錯，歡迎將生活随笔 推薦給好友。
