JavaEE：多線程（3）：案例代碼

案例一：單例模式

餓漢模式

懶漢模式

思考：懶漢模式是否線程安全？

案例二：阻塞隊列

可以實現生產者消費者模型

削峰填谷

接下來我們自己實現一個阻塞隊列

1.先實現一個循環隊列

2. 引入鎖，實現線程安全

3.實現阻塞

實現生產者消費者模型

案例三：定時器

問題

線程安全

線程餓死

理解代碼過程

案例四：線程池

標準庫中的線程池：ThreadPoolExecutor

Executors工廠類

手敲線程池

多線程基礎知識要點

案例一：單例模式

是一種設計模式

軟件設計需要框架，這是硬性的規定；設計模式是軟性的規定。遵循好設計模式，代碼的下限就被兜住了

單例 = 單個實例（對象）

某個類在一個進程中只應該創建出一個實例（原則上不應該有多個）

使用單例模式可以對代碼進行一個更嚴格的校驗和檢查

實現單例模式~

餓漢模式

第1步：

class Singleton{private static Singleton instance = new Singleton();
}

這里的static指的是類屬性，而instance就是Singleton類對象持有的屬性

每個類的類對象只存在一個，類對象中的static屬性自然只有一個了

因此instance指向的這個對象，就是唯一的對象

第2步：

其他代碼要想使用這個類的實例就需要通過這個方法來進行獲取。不應該在其他代碼鐘重新new這個對象，而是使用這個方法獲取到現成的對象（已經創建好的對象）

第3步：奇淫巧計

這里直接把Singleton給private了，其他代碼根本沒辦法new

此時，無論你創建多少個對象，這些對象其實都是一樣的

餓漢模式下，實例是在類加載的時候就創建了，創建時機非常早，相當于程序一啟動，實例就創建了。“餓漢”形容“創建實例非常迫切，非常早”

欸！但是用非常規手段：反射就可以打破上述約定。我們可以用枚舉方法來創建單例模式

懶漢模式

創建實例的時機更晚，只到第一次使用的時候才會創建實例

“懶”的思想

比如有一個非常大的文件(10GB)，有一個編輯器，使用編輯器打開這個文件，如果是按照餓漢的方式，編輯器就要把這10GB先加載到內存里，然后再統一地展示。加載太多數據，用戶還得一點點看，沒辦法一下看那么多

如果按照懶漢的方式，編輯器就會制度取一小部分數據，把這部分數據先展示出來，隨著用戶翻頁之類的操作再繼續讀后面的數據。這樣效率可以提高

class SingletonLazy{private static SingletonLazy instance = null;//先初始化為null，不是立即初始化public static SingletonLazy getInstance(){if (instance == null){instance = new SingletonLazy();//首次調用getInstance才會創建出一個實例}return instance;}private SingletonLazy(){}
}

思考：懶漢模式是否線程安全？

這樣t1 new了一個對象，t2也new了一個對象，就會出現bug

所以，懶漢模式不是線程安全的

那怎么改成線程安全的呢？

1.加鎖，synchronized

2.把if和new兩個操作打包成一個原子

仍然是t1和t2兩個線程，t1先執行加鎖代碼，t2就被阻塞了，要等待t1釋放鎖才能繼續執行

而t1把instance修改之后，t2的if條件就不成立了，直接就返回了

emm，這段代碼還不夠完美...

在多線程里面，當第一個線程加了鎖，后面的線程再調用getInstance就是純粹的讀操作了，也就不會有線程問題了。那么沒有線程的代碼每次執行都要加鎖和解鎖，每次都會產生阻塞，效率巨低！

所以在synchronized外邊還得再套一層if，判定代碼是否要加鎖。仍然將instance是否為空作為判斷條件

第一個if判定是否加鎖

第二個if判定是否要創建對象?

🆗上面的代碼還有一點問題

涉及到指令重排序引起的線程安全問題

指令重排序是指調整原有代碼的執行順序，保證邏輯不變的前提下提高程序的效率

為什么調整代碼執行順序可以提高程序效率？

比如我們去超市買東西，我們需要買黃瓜，胡蘿卜，西紅柿，土豆。我們就有很多種去不同攤位的路徑選擇，每種選擇的最終總購買時間不一樣。這就相當于程序的效率。

這行代碼可以分成三個大步驟

1.申請一段內存空間

2.在這個內存上調用構造方法，創建出這個實例

3.把這個內存地址賦給Instance引用變量

假設有t1和t2兩個線程

t1線程按照1 3 2的執行順序，就會出現問題

解決上述問題核心思路：volatile

volatile有兩個功能：

1）保證內存可見性，每次訪問變量必須要重新讀取內存，而不會優化到寄存器/緩存中

2）禁止指令重排序，針對這個volatile修飾的變量的讀寫操作的相關指令，是不能被重排序的

這樣修改之后，針對instance變量的讀寫操作就不會出現重排序

案例二：阻塞隊列

特點：1.線程安全；2.阻塞

如果一個已經滿了的隊列進行入隊列，此時入隊列操作就會阻塞，一直阻塞到隊列不滿之后

如果一個已經空的隊列進行出隊列，出隊列操作就會阻塞，一直阻塞到隊列有元素為止

可以實現生產者消費者模型

這個模型可以更好地解耦合(把代碼的耦合程度從高降低)

實際開發中，往往會用到分布式系統，服務器整個功能不是由一個服務器完成的，而是每個服務器負責一部分功能。通過服務器之間的網絡通信，最終完成整個功能

在這個案例中，A和B,C之間的耦合性比較強，一旦B或者C掛了一個，A也就跟著掛了

如果引入生產者消費者模型

這個阻塞隊列不是簡單的數據結構，而是基于這個數據結構實現的服務器程序，又被部署到單獨的主機上

削峰填谷

為啥當請求多了的時候，服務器就容易掛？

因為服務器處理每個請求都是要消耗硬件資源（包括但不限于CPU，內存，硬盤，網絡帶寬），上述任何一種硬件資源達到瓶頸，服務器都會掛

因為B和C抗壓能力比較弱，所以我們可以用一個阻塞隊列來承擔峰值請求

阻塞隊列：數據結構

消息隊列：基于阻塞隊列實現服務器程序

Java標準庫里線程的阻塞隊列

BlickingQueue: ArrayBlockingQueue, LinkedBlockingQueue, PriorityBlockingQueue

        BlockingDeque<String> queue = new ArrayBlockingQueue<>(100);queue.put("aaa");

put和offer都是入隊列，但是put帶有阻塞功能，而offer沒帶阻塞功能，隊列滿了會返回布爾結果

        String elem = queue.take();System.out.println("elem = "+elem);

take用來出隊列，帶有阻塞功能?

接下來我們自己實現一個阻塞隊列

1.先實現一個循環隊列

class MyBlockingQueue{private String[] elems = null;private int head = 0;private int tail = 0;private int size = 0;public MyBlockingQueue(int capacity){elems = new String[capacity];}public void put(String elem){//新的元素放到tail指向的位置上if (size >= elems.length){//隊列滿了，需要下面這個代碼阻塞return;}//新的元素要放到tail指向的元素上elems[tail] = elem;tail++;if(tail >= elems.length){tail = 0;}size++;}public String take(){if(size == 0){//隊列空了，需要下面這個代碼阻塞return null;}String elem = elems[head];head ++;if(head >= elems.length){head = 0;}size--;return null;}
}

2. 引入鎖，實現線程安全

    private static Object locker = new Object();public MyBlockingQueue(int capacity){elems = new String[capacity];}public void put(String elem){synchronized (locker){//新的元素放到tail指向的位置上if (size >= elems.length){//隊列滿了，需要下面這個代碼阻塞return;}//新的元素要放到tail指向的元素上elems[tail] = elem;tail++;if(tail >= elems.length){tail = 0;}size++;}}public String take(){String elem = null;synchronized (locker){if(size == 0){//隊列空了，需要下面這個代碼阻塞return null;}elem = elems[head];head ++;if(head >= elems.length){head = 0;}size--;return elem;}}
}

3.實現阻塞

對于滿了的情況，用wait方法阻塞，在出隊列成功之后再進行喚醒

隊列空的情況，在入隊列成功后的線程中喚醒

    public void put(String elem) throws InterruptedException {synchronized (locker){//新的元素放到tail指向的位置上while (size >= elems.length){//隊列滿了，需要下面這個代碼阻塞locker.wait();}//新的元素要放到tail指向的元素上elems[tail] = elem;tail++;if(tail >= elems.length){tail = 0;}size++;//入隊列成功后喚醒locker.notify();}}public String take() throws InterruptedException {String elem = null;synchronized (locker){while (size == 0){//隊列空了，需要下面這個代碼阻塞locker.wait();}elem = elems[head];head ++;if(head >= elems.length){head = 0;}size--;//出隊列成功后喚醒locker.notify();}return elem;}

這里的if為什么改成while了呢？

因為if只能判定一次條件，有時候一旦程序進入阻塞之后再被喚醒，中間隔的時間會很長，這個間隔過程變數很多，可能這個入隊列的條件無法再滿足了。

欸那改成while之后，就是wait喚醒之后再判定一次條件，wait之前判定一次，喚醒之后再判定一次（就是多做一次確定）。再次確認發現隊列還是滿的，那就繼續等待。

實現生產者消費者模型

    public static void main(String[] args) {MyBlockingQueue queue = new MyBlockingQueue(1000);//生產者Thread t1 = new Thread(()->{int n = 1;while(true){try {queue.put(n + "");System.out.println("生產元素 " + n);n++;Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});//消費者Thread t2 = new Thread(()->{while(true){try {String n = queue.take();System.out.println("消費元素 " + n);} catch (InterruptedException e) {throw new RuntimeException(e);}}});t1.start();t2.start();}

實際開發中，生產者和消費者往往不僅僅是一個線程，也可能是一個獨立的服務器程序

案例三：定時器

可以設定一個時間，時間到了的時候，定時器自動執行某個邏輯（比如，寫博客定時發布）

用法：定義一個timer添加多個任務，每個任務同時會帶有一個時間

Timer里面內置了前臺線程，因為timer不知道你的代碼是否還會添加新的任務進來，仍然嚴正以待

需要使用cancel來主動結束

現在我們來手搓一個定時器

需要有什么？1.一個可以幫我們掐時間的線程；2.一個能幫我們存儲任務的優先級隊列

因為每個任務都帶有delay時間的，用優先級隊列可以先執行時間小的，后執行時間大的

掃描線程就不必遍歷了，只需要關注隊首元素是否到時間（隊首沒到時間，其他元素也沒到時間）

計時任務

任務優先級邏輯（時間小的優先級越高）

計時器

問題

線程安全

由于我們在主線程中對隊列的元素進行添加，而掃描線程對已完成的元素進行刪除，兩個線程操作同一個優先級隊列變量，會有線程安全問題

此時需要加鎖來解決線程安全問題

schedule方法（主線程要調用）的加鎖

?判斷：以下哪種加鎖方法是正確的

   //第一種public MyTimer(){t = new Thread(()->{//掃描線程就需要循環的反復掃描隊首元素，然后判定隊首任務時間是否到達//時間到了就執行任務并刪除這個任務//時間沒到就啥都不干synchronized (locker){while (true){if (queue.isEmpty()){continue;}MyTimerTask task = queue.peek();//獲取當前時間long curTime = System.currentTimeMillis();if(curTime >= task.getTime()){//當前時間已經到了任務時間，就可以執行任務了queue.poll();task.run();}else{//時間還沒到，暫時先不執行continue;}} }});//第二種public MyTimer(){t = new Thread(()->{//掃描線程就需要循環的反復掃描隊首元素，然后判定隊首任務時間是否到達//時間到了就執行任務并刪除這個任務//時間沒到就啥都不干while (true){synchronized (locker){if (queue.isEmpty()){continue;}MyTimerTask task = queue.peek();//獲取當前時間long curTime = System.currentTimeMillis();if(curTime >= task.getTime()){//當前時間已經到了任務時間，就可以執行任務了queue.poll();task.run();}else{//時間還沒到，暫時先不執行continue;}}}});}

第一種方法，把鎖放到while外面，如果while沒有結束的話，鎖永遠都釋放不了，主線程調用schedule方法就永遠上不了鎖。所以我們要采用第二種方法，把鎖加到while里面，才有釋放鎖的機會

線程餓死

上面的第二種方法雖然解決了線程安全問題，但是這部分代碼執行速度很快，解鎖之后就立即重新加鎖，導致其他線程想通過schedule加鎖都加不上，所以我們需要使用wait來解決

一旦由新的任務加入，wait就會被喚醒，因為不知道加入的任務是不是最早的任務，所以我們用task.getTime() - curTime來獲取任務時間

沒有新的任務，時間到了。按照原定計劃，執行之前的這個最早的任務即可

執行結果

理解代碼過程

理解peek:

優先級隊列：無論添加多少元素，這里的peek都是得到時間最小的值。

理解run方法

👇

👇（Runnable作為描述任務的主體）

👇

main方法里面寫出任務具體執行代碼

案例四：線程池

池是什么？

池就相當于一個共享資源，是對資源的整合和調配，節省存儲空間，當需要的時候可以直接在池中取，用完之后再還回去。比如，如果你喝水，你可以拿杯子去水龍頭接。如果很多人喝水，那就只能排隊去接。

Java常用的池有常量池，數據庫連接池，線程池，進程池，內存池

最開始進程能夠解決并發編程的問題，但是因為頻繁創建銷毀進程的成本太高了，引入了線程這種輕量級進程。但是如果創建銷毀線程的頻率進一步提高，這里的開銷也不能忽視

那怎么優化線程創建銷毀效率呢？

1.引入輕量級線程--纖程/協程

協程本質是程序員在用戶態代碼中進行調度，不是靠內核的調度器來調度的

協程運行在線程之上，當一個協程執行完成后，可以選擇主動讓出，讓另一個協程運行在當前線程之上。可以節省很多調度上的開銷。

?線程里有協程這句話是不嚴謹的，因為協程本身不是系統級別的概念，是用戶代碼中基于線程封裝出來的，有不同的實現方法，可能n個協程對應1個線程，也可能n個協程對應m個線程

2.引入線程池。把要使用的線程提前創建好，用完了也不要直接釋放而是備下次使用，就節省創建/銷毀線程的開銷

從線程池里取線程（純用戶態代碼）比從系統申請更高效！

比如一個事情一個人自己就能完成，就更可控，更高效，這種相當于純用戶態代碼

但是如果這個事情這個人要拜托其他人來完成，不知道委托人要花多少時間，就不可控，更低效。相當于去系統申請線程

標準庫中的線程池：ThreadPoolExecutor

構造方法（面試題常考）

標準庫提供的線程池，持有的線程個數并不是一成不變的，會根據當前的任務量自適應線程個數?

?核心線程數（規定一個線程池里最少有多少個線程）

最大線程數（規定一個線程最大有多少個線程）

某個線程超過保持存活時間閾值就會被銷毀掉

和定時器類似，線程池中可以持有多個任務

?-- 線程工廠

通過這個工廠類創建線程對象（Thread對象），在這個類里面提供了方法，讓方法封裝new Thread的操作，同時給Thread設置一些屬性

設計模式：工廠模式。通過專門的工廠類/對象來創建指定的對象

例子：
//平面上的一個點class Point{public Point(double x, double y){...}//通過笛卡爾坐標構造這個點//還可以用三角函數轉換笛卡爾坐標//x = r * cos(a); y = r * sin(a)public Point(double r, double a){...}//通過極坐標系來構造點（半徑，角度）}
上面代碼能編譯通過嗎？不能。因為不能構成重載（因為形參類型和個數相同了）

為了讓上面代碼通過，就可以引入工廠模式
class Point{//工廠方法public static Point makePointByXY(double x, double y){Point p = new Point();p.setX(x);p.setY(y);return p;}public static Point makePointByRA(double r, double a){Point p = new Point();p.setR(r);p.setA(a);return p;}Point p = Point.makePointByXY(x, y);Point p = Point.makePointByRA(r, a);
}
通過靜態方法封裝new操作，在方法內部設定不同的屬性完成對象初始化，這個構造對象的過程就是工廠模式

拒絕策略

在線程池中，有一個阻塞隊列，能夠容納的元素有上限，當任務隊列已經滿了，如果繼續往隊列中添加任務，線程池中就會拒絕添加

四種拒絕策略

第一種：繼續添加任務，直接拋出異常

第二種：新的任務由添加任務的線程負責執行（線程池不會執行）--誰攬的活誰干

第三種：丟棄最老的任務

第四種：丟棄最新的任務

Executors工廠類

通過這個類創建不同的線程池對象

例子

    public static void main(String[] args) {ExecutorService service = Executors.newFixedThreadPool(4);service.submit(new Runnable() {@Overridepublic void run() {}});}

啥時候使用Executors，啥時候使用ThreadPoolExecutor

Executors方便只是簡單用一下，ThreadPoolExecutor希望高度定制化

線程池里最好有多少個線程？（具體情況具體分析，回答具體數字就是錯誤的）

線程里的任務分成兩種

CPU密集型任務：這個線程大部分時間都在CPU上運行/計算。比如在線程run里面計算1+2+3+...+10w。

IO密集型任務：這個線程大部分時間都在等待IO，不需要去CPU上運行。比如線程run里加scanner，讀取用戶輸入。

如果一個進程中，所有的線程都是CPU密集型，每個線程所有的工作都在CPU上執行。此時，線程數目就不應該超過N（CPU邏輯核心數）。——每個線程都要占一個核，超過N就失控了

如果一個進程中，所有的線程都是IO密集型，每個線程大部分工作都在等待IO，CPU消耗非常少。此時線程數目就可以很多，遠遠超過N。——一個線程工作，其他線程休息，不霸占CPU核

手敲線程池

1.提供構造方法，指定創建多少個線程

2.在構造方法中，把這些線程都創建好

3.有一個阻塞隊列，能夠持有要執行的任務

4.提供submit方法，能夠添加新的任務

寫的過程中遇到問題

run變量捕獲到i之后，正常情況i是不能變的，但是i因為循環造成改變，引發編譯器異常

此處的n就是一個實時final變量，每次循環就創建一個不可變的n，這個n是可以被捕獲的

package Thread;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;class MyThreadPoolExecutor{private List<Thread> threadList = new ArrayList<>();//創建一個用來保存任務的隊列private BlockingQueue<Runnable> queue = new ArrayBlockingQueue<>(1000);//通過n指定創建多少個線程public MyThreadPoolExecutor(int n){for (int i = 0; i < n; i++) {Thread t = new Thread(()->{while (true) {try {//此處take帶有阻塞功能，如果此處隊列為空，take就會阻塞Runnable runnable = queue.take();//取出一個任務就執行一個任務runnable.run();} catch (InterruptedException e) {e.printStackTrace();}}});t.start();threadList.add(t);}}public void submit(Runnable runnable) throws InterruptedException{queue.put(runnable);}
}
public class ThreadDemo11 {public static void main(String[] args) throws InterruptedException{MyThreadPoolExecutor executor = new MyThreadPoolExecutor(4);for (int i = 0; i < 1000; i++) {int n = i;executor.submit(new Runnable() {@Overridepublic void run() {System.out.println("執行任務 "+ n + ", 當前線程為：" + Thread.currentThread().getName());}});}}
}

更體現多線程執行順序不確定