JUC並行程式設計詳解（通俗易懂）
2023-01-05 21:02:11
一、JUC簡介
在Java5.0提供了java.util.concurrent包，簡稱JUC，即Java並行程式設計工具包。JUC更好的支援高並行任務。
具體的有以下三個包：
java.util.concurrent
java.util.concurrent.atomic
java.util.concurrent.locks
二、Lock鎖
1、傳統的synchronized鎖
/**
* synchronized售票例子
*/
public class SynSaleTicket {
//真正在公司開發，遵守oop思想，降低耦合性
//執行緒就是一個單獨的資源類，沒有任何附屬操作，裡面只包含屬性、方法
public static void main(String[] args) {
//並行：多個執行緒操作同一個資源
Ticket ticket = new Ticket();
//lambda表示式
new Thread(() -> {
for (int i = 0; i < 40; i++) {
ticket.sale();
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 40; i++) {
ticket.sale();
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 40; i++) {
ticket.sale();
}
}, "C").start();
}
}
//資源類
class Ticket {
//票數
private int number = 30;
//買票方法
//synchronized本質就是佇列+鎖
public synchronized void sale() {
if (number > 0) {
System.out.println(Thread.currentThread().getName() + "搶到了第" + (number--) + "張票，剩下" + number);
}
}
}
2、JUC包下的Lock介面
檢視jdk1.8官方檔案可以看到有三個實現類：
- ReentrantLock：可重入鎖（常用）
- ReentrantReadWriteLock.ReadLock：可重入讀鎖
- ReentrantReadWriteLock.WriteLock：可重入寫鎖
//lock鎖用法：
//1. 建立鎖物件 Lock l = ...;
//2. 加鎖 l.lock();
//3. 解鎖 try {} finally { l.unlock(); }
1. 公平鎖和非公平鎖
通俗的說公平鎖其實就是買票都需要排隊按隊伍順序遵循先來後到的原則獲得鎖，非公平鎖就是有人開VIP可以插隊獲得鎖。
- 公平鎖：多個執行緒按照申請鎖的順序去獲得鎖，執行緒會直接進入佇列去排隊，永遠都是佇列的第一位才能得到鎖。
- 優點：所有的執行緒都能得到資源，不會餓死在佇列中。
- 缺點：吞吐量會下降很多，佇列裡面除了第一個執行緒，其他的執行緒都會阻塞，cpu喚醒阻塞執行緒的開銷會很大。
- 非公平鎖：多個執行緒獲取鎖的時候，不會按照申請鎖的順序去獲得鎖，會直接嘗試獲取鎖，如果能獲取到鎖，就直接獲得鎖，如果獲取不到，再進入等待佇列乖乖等待。
- 優點：可以減少CPU喚醒執行緒的開銷，整體的吞吐效率會高點，CPU也不必取喚醒所有執行緒，會減少喚起執行緒的數量。
- 缺點：可能導致佇列中排隊的執行緒一直獲取不到鎖或者長時間獲取不到鎖，活活餓死。
//ReentrantLock無參構造，相當於ReentrantLock(false)
public ReentrantLock() {
sync = new NonfairSync();//預設是非公平鎖
}
//ReentrantLock有參構造，fair引數決定是否選為公平鎖，true是，false否
public ReentrantLock(boolean fair) {
sync = fair ? new FairSync()/*公平鎖*/ : new NonfairSync()/*非公平鎖/*;
}
package com.hao.demo01;
//使用juc包下的鎖
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* Lock售票例子
*/
public class LockSaleTicket {
//使用Lock鎖來解決買票問題
public static void main(String[] args) {
Ticket ticket = new Ticket();
new Thread(() -> {for (int i = 0; i < 40; i++) ticket.sale();}, "A").start();
new Thread(() -> {for (int i = 0; i < 40; i++) ticket.sale();}, "B").start();
new Thread(() -> {for (int i = 0; i < 40; i++) ticket.sale();}, "C").start();
}
}
class Ticket2 {
//票數
private int number = 30;
//Lock鎖
Lock lock = new ReentrantLock();
//買票方法
public void sale() {
lock.lock();//加鎖
try {
if (number > 0) {
System.out.println(Thread.currentThread().getName() + "搶到了第" + (number--) + "張票，剩下" + number);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();//解鎖
}
}
}
3、Synchronized 和 Lock鎖的區別
| 區別 | Synchronized | Lock |
|---|---|---|
| 是否關鍵字 | Synchronized是Java內建關鍵字 | Lock類是一個介面 |
| 是否可嘗試獲取鎖 | Synchronized無法判斷是否獲取鎖的狀態 | Lock可以判斷是否獲取到鎖 |
| 是否自動釋放鎖 | Synchronized會自動釋放鎖(a 執行緒執行完同步程式碼會釋放鎖；b 執行緒執行過程中發生異常會釋放鎖) | Lock需在finally中手工釋放鎖，否則容易造成執行緒死鎖 |
| 是否一直阻塞 | 用Synchronized關鍵字修飾的兩個執行緒1和執行緒2，如果當前執行緒1獲得鎖，執行緒2執行緒等待。如果執行緒1阻塞，執行緒2則會一直等待下去 | Lock鎖就不一定會等待下去，如果嘗試獲取不到鎖，執行緒可以不用一直等待就結束了 |
| 是否可重入、中斷、公平鎖 | Synchronized的鎖可重入、不可中斷、非公平 | Lock鎖可重入、可中斷、可公平（也可非公平） |
| 使用場合 | Synchronized鎖適合程式碼少量的同步問題 | Lock鎖適合大量同步的程式碼的同步問題 |
總體來說，Lock鎖比synchronized更加靈活，提供了更加豐富的API進行同步操作，也可以結合Condition條件實現比較複雜的執行緒間同步通訊。
三、執行緒通訊
1、傳統的Synchronized生產者與消費者
使用 Synchronized + wait + notify 這一套完成執行緒通訊。
/**
* 生產者和消費者問題！
*/
public class TestPC {
//使用多個執行緒操作同一個變數num=0
//執行緒A負責num+1，完事之後通知B操作。執行緒B負責num-1，完事之後通知A操作
public static void main(String[] args) {
Data data = new Data();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "B").start();
}
}
//資源類
class Data {
//變數
private int num = 0;
//加
public synchronized void increment() throws InterruptedException {
//判斷等待
if (num != 0) {
this.wait();
}
num++;
System.out.println(Thread.currentThread().getName() + "=>" + num);
//通知其他執行緒，+1執行完畢
this.notifyAll();
}
//減
public synchronized void decrement() throws InterruptedException {
//判斷等待
if (num == 0) {
this.wait();
}
num--;
System.out.println(Thread.currentThread().getName() + "=>" + num);
//通知其他執行緒，-1執行完畢
this.notifyAll();
}
}
如果此時再加兩個C、D執行緒，兩加兩減，發現執行出問題。
1. 問題誕生：虛假喚醒
A=>1
B=>0
C=>1
A=>2 //出現問題
C=>3
B=>2
B=>1
B=>0
C=>1
A=>2
Process finished with exit code 0
檢視官方檔案解釋Object類下的wait()方法
也就是說，這裡用if判斷的話，被喚醒後的執行緒將不會重新進入if判斷，程式碼直接執行if程式碼塊之後的程式碼，而使用while的話，被喚醒後的執行緒會重新判斷迴圈條件，如果不成立再執行while程式碼塊之後的程式碼。
解決辦法：將 if 換成 while
/**
* 生產者和消費者問題！
*/
public class TestPC {
//使用多個執行緒操作同一個變數num=0
//執行緒A負責num+1，完事之後通知B操作。執行緒B負責num-1，完事之後通知A操作
public static void main(String[] args) {
Data data = new Data();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "C").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "D").start();
}
}
//資源類
class Data {
//變數
private int num = 0;
//加
public synchronized void increment() throws InterruptedException {
//判斷等待
while (num != 0) {
this.wait();
}
num++;
System.out.println(Thread.currentThread().getName() + "=>" + num);
//通知其他執行緒，+1執行完畢
this.notifyAll();
}
//減
public synchronized void decrement() throws InterruptedException {
//判斷等待
while (num == 0) {
this.wait();
}
num--;
System.out.println(Thread.currentThread().getName() + "=>" + num);
//通知其他執行緒，-1執行完畢
this.notifyAll();
}
}
2、JUC版的生產者和消費者
思考，我們的傳統的Synchronized鎖被JUC版的Lock鎖替代了，那傳統的wait()和notify()有沒有被替換呢？
當然有滴！
我們檢視java.util.concurrent.locks包下的Lock介面下的方法
點Condition進入看
Condition介面屬於JUC包下的，繼續檢視官方檔案的解釋
繼續檢視Condition介面中的方法
其中有 await()等待 和 signal()喚醒 方法替換。
於是新的一套鎖方案出來了：Lock + await + signal
要用await()和signal()就必須得到Condition範例，看官方檔案怎麼說？
/**
* 生產者和消費者問題！
*/
public class TestLockPC {
public static void main(String[] args) {
Data2 data = new Data2();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.increment();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "C").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
try {
data.decrement();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}, "D").start();
}
}
class Data2 {
private int num = 0;
//建立Lock鎖範例
private final Lock lock = new ReentrantLock();
//建立Condition範例
private final Condition condition = lock.newCondition();
//加
public void increment() throws InterruptedException {
lock.lock();//加鎖
try {
//判斷等待
while (num != 0) {
condition.await();
}
num++;
System.out.println(Thread.currentThread().getName() + "=>" + num);
//通知其他執行緒，+1執行完畢
condition.signalAll();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();//解鎖
}
}
//減
public void decrement() throws InterruptedException {
lock.lock();//加鎖
try {
//判斷等待
while (num == 0) {
condition.await();
}
num--;
System.out.println(Thread.currentThread().getName() + "=>" + num);
//通知其他執行緒，-1執行完畢
condition.signalAll();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
lock.unlock();//解鎖
}
}
}
A=>1
B=>0
C=>1
B=>0
C=>1
B=>0
C=>1
D=>0
A=>1
D=>0
A=>1
D=>0
Process finished with exit code 0
乍一看，Condition也沒什麼牛逼的呀，還有什麼牛逼的功能嗎？
當然有滴！
我們看執行出來的結果執行緒還是隨機狀態的，A B C B C B C D A D A D，我們想要實現A執行完通知B，B執行完通知C，C執行完通知D，D執行完通知A，實現 A B C D 按順序執行。
使用Condition精準通知和喚醒執行緒。
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 精準通知和精準喚醒執行緒。
*/
public class TestAccuratePC {
public static void main(String[] args) {
Data3 data = new Data3();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
data.printA();
}
}, "A").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
data.printB();
}
}, "B").start();
new Thread(() -> {
for (int i = 0; i < 10; i++) {
data.printC();
}
}, "C").start();
}
}
class Data3 {
//建立Lock鎖範例
private final Lock lock = new ReentrantLock();
//建立Condition鎖範例
private final Condition condition1 = lock.newCondition();
private final Condition condition2 = lock.newCondition();
private final Condition condition3 = lock.newCondition();
private int num = 1;
//假如num等於1就讓A執行，2就讓B執行，3就讓C執行
public void printA() {
lock.lock();//加鎖
try {
while (num != 1) {
condition1.await();//等待
}
System.out.println(Thread.currentThread().getName() + "=>AAAAAA");
num = 2;
//喚醒指定的人，B
condition2.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();//解鎖
}
}
public void printB() {
lock.lock();//加鎖
try {
while (num != 2) {
condition2.await();//等待
}
System.out.println(Thread.currentThread().getName() + "=>BBBBBB");
num = 3;
//喚醒指定的人，C
condition3.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();//解鎖
}
}
public void printC() {
lock.lock();//加鎖
try {
while (num != 3) {
condition3.await();//等待
}
System.out.println(Thread.currentThread().getName() + "=>CCCCCC");
num = 1;
//喚醒指定的人，A
condition1.signal();
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();//解鎖
}
}
}
A=>AAAAAA
B=>BBBBBB
C=>CCCCCC
A=>AAAAAA
B=>BBBBBB
C=>CCCCCC
A=>AAAAAA
B=>BBBBBB
C=>CCCCCC
Process finished with exit code 0
四、8鎖現象
通過8個問題徹底理解鎖。
鎖只會鎖：物件、class
問題1
import java.util.concurrent.TimeUnit;
/**
* 題1：兩個同步方法，兩條執行緒分別執行，先列印 「傳簡訊」 還是 「打電話」
*/
public class Demo01 {
public static void main(String[] args) throws InterruptedException {
Phone phone = new Phone();
new Thread(() -> {
phone.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone.call();
}, "B").start();
}
}
class Phone{
public synchronized void sendSms(){
System.out.println("傳簡訊");
}
public synchronized void call(){
System.out.println("打電話");
}
}
結果是先傳簡訊後打電話。
如果你認為傳簡訊的方法先被呼叫就先執行，那你再看下面的情況。
問題2
import java.util.concurrent.TimeUnit;
/**
* 題2：兩個同步方法，兩條執行緒分別執行，傳簡訊睡3秒，先列印 「傳簡訊」 還是 「打電話」
*/
public class Demo01 {
public static void main(String[] args) throws InterruptedException {
Phone phone = new Phone();
new Thread(() -> {
phone.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone.call();
}, "B").start();
}
}
class Phone{
public synchronized void sendSms(){
//讓sendSms()方法睡3秒
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
public synchronized void call(){
System.out.println("打電話");
}
}
還是先傳簡訊後打電話。
原因：因為Synchronized鎖的物件是方法呼叫者，兩個方法用的同一把鎖，所以誰先拿到鎖，誰就先執行！
問題3
import java.util.concurrent.TimeUnit;
/**
* 題3：新增一個普通方法hello()，A執行緒呼叫sendSms()，B執行緒呼叫hello()
*/
public class Demo02 {
public static void main(String[] args) throws InterruptedException {
Phone2 phone = new Phone2();
new Thread(() -> {
phone.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone.hello();
}, "B").start();
}
}
class Phone2{
public synchronized void sendSms(){
try {
TimeUnit.SECONDS.sleep(3);//睡3秒
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
public synchronized void call(){
System.out.println("打電話");
}
//這裡沒有鎖，不用去拿鎖。
public void hello(){
System.out.println("hello");
}
}
結果：先列印hello，再列印傳簡訊
原因：hello()方法是一個普通方法，所以不受synchronized鎖的影響，不用等待鎖釋放。
問題4
import java.util.concurrent.TimeUnit;
/**
* 題4：兩個物件，物件1呼叫sendSms()，物件2呼叫call()
*/
public class Demo02 {
public static void main(String[] args) throws InterruptedException {
Phone2 phone1 = new Phone2();
Phone2 phone2 = new Phone2();
new Thread(() -> {
phone1.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone2.call();
}, "B").start();
}
}
class Phone2{
public synchronized void sendSms(){
try {
TimeUnit.SECONDS.sleep(3);//睡3秒
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
public synchronized void call(){
System.out.println("打電話");
}
}
結果：先打電話，再傳簡訊
原因：因為兩個物件有兩把鎖，各自用各自的鎖，傳簡訊的方法延遲了3秒，所以打電話的方法先執行。
問題5
import java.util.concurrent.TimeUnit;
/**
* 題5：兩個靜態同步方法，一個物件，先列印 「傳簡訊」 還是 「打電話」
*/
public class Test3 {
public static void main(String[] args) throws InterruptedException {
Phone3phone = new Phone3();
new Thread(() -> {
phone.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone.call();
}, "B").start();
}
}
//只有唯一一個class物件
class Phone3{
//Synchronized鎖的是方法的呼叫者！
//static靜態方法，類一載入就有了，鎖的是class
public static synchronized void sendSms(){
try {
TimeUnit.SECONDS.sleep(3);//睡3秒
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
public static synchronized void call(){
System.out.println("打電話");
}
}
結果：先傳簡訊，後打電話
問題6
import java.util.concurrent.TimeUnit;
/**
* 題6：兩個靜態同步方法，兩個物件，物件1呼叫sendSms()，物件2呼叫call()，先列印 「傳簡訊」 還是 「打電話」？
*/
public class Test3 {
public static void main(String[] args) throws InterruptedException {
Phone3 phone1 = new Phone3();
Phone3 phone2 = new Phone3();
new Thread(() -> {
phone1.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone2.call();
}, "B").start();
}
}
//只有唯一一個Class物件
class Phone3{
//Synchronized鎖的是方法的呼叫者！
//static靜態方法，類一載入就有了，鎖的是class
public static synchronized void sendSms(){
try {
TimeUnit.SECONDS.sleep(3);//睡3秒
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
public static synchronized void call(){
System.out.println("打電話");
}
}
結果：先傳簡訊，後打電話
原因：靜態方法是和類一起載入的，也就是說這個靜態方法是屬於這個類的，如果在static方法中使用synchronized鎖，鎖的將是Class物件，而Class物件又是全域性唯一的，不管new多少個物件，它的Class物件只有一個，所以誰先拿到了鎖誰就先執行。
問題7
import java.util.concurrent.TimeUnit;
/**
* 題7：一個靜態同步方法，一個普通同步方法，一個物件，先列印 「傳簡訊」 還是 「打電話」？
*/
public class Test4 {
public static void main(String[] args) throws InterruptedException {
Phone4 phone = new Phone4();
new Thread(() -> {
phone.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone.call();
}, "B").start();
}
}
class Phone4{
//靜態同步方法，鎖的是class類別範本
public static synchronized void sendSms(){
try {
TimeUnit.SECONDS.sleep(3);//睡3秒
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
//普通同步方法，鎖的是呼叫者
public synchronized void call(){
System.out.println("打電話");
}
}
結果：先打電話，後傳簡訊
問題8
import java.util.concurrent.TimeUnit;
/**
* 題8：一個靜態同步方法，一個普通同步方法，兩個物件，物件1呼叫sendSms()，物件2呼叫call()，先列印 「傳簡訊」 還是 「打電話」？
*/
public class Test4 {
public static void main(String[] args) throws InterruptedException {
Phone4 phone1 = new Phone4();
Phone4 phone2 = new Phone4();
new Thread(() -> {
phone1.sendSms();
}, "A").start();
TimeUnit.SECONDS.sleep(1);//JUC包下的sleep，睡1秒
new Thread(() -> {
phone2.call();
}, "B").start();
}
}
class Phone4{
//靜態同步方法，鎖的是class類別範本
public static synchronized void sendSms(){
try {
TimeUnit.SECONDS.sleep(3);//睡3秒
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("傳簡訊");
}
//普通同步方法，鎖的是呼叫者
public synchronized void call(){
System.out.println("打電話");
}
}
結果：先打電話，後傳簡訊
原因：靜態同步方法鎖的是class物件，普通同步方法鎖的是呼叫者，鎖的是不同物件，所以後者不需要等待前者釋放鎖。
小結：總之synchronized鎖就鎖兩個東西，一個是new出來的this物件，一個是類Class唯一的模板。
五、不安全的集合
1、List
我們通常使用的ArrayList就是執行緒不安全的，舉個簡單的例子
/**
* List集合
*/
public class TestList {
public static void main(String[] args) {
List<String> list = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
new Thread(() -> {
list.add(UUID.randomUUID().toString().substring(0, 5));
System.out.println(list);
}, String.valueOf(i)).start();
}
}
}
//並行修改異常
Exception in thread "4" java.util.ConcurrentModificationException
at java.util.ArrayList$Itr.checkForComodification(ArrayList.java:909)
at java.util.ArrayList$Itr.next(ArrayList.java:859)
at java.util.AbstractCollection.toString(AbstractCollection.java:461)
at java.lang.String.valueOf(String.java:2994)
at java.io.PrintStream.println(PrintStream.java:821)
at com.hnguigu.unsafe.TestList.lambda$main$0(TestList.java:20)
at java.lang.Thread.run(Thread.java:748)
Process finished with exit code 0
解決方案，有以下幾種！
import java.util.concurrent.CopyOnWriteArrayList;
public class TestList{
public static void main(String[] args) {
/**
* 解決方案
* 1. List<String> list = new Vector<>();
* 2. List<String> list = Collections.synchronizedList(new ArrayList<>());
* 3. List<String> list = new CopyOnWriteArrayList<>();
*/
List<String> list = new CopyOnWriteArrayList<>();
for (int i = 1; i <=10; i++) {
new Thread(() -> {
list.add(UUID.randomUUID().toString().substring(0,5));
System.out.println(list);
},String.valueOf(i)).start();
}
}
}
寫入時複製（CopyOnWrite，簡稱COW）思想是計算機程式設計領域中的一種通用優化策略。
CopyOnWrite容器即寫入時複製的容器。通俗的理解是當我們往一個容器新增元素的時候，不直接往當前容器新增，而是先將當前容器進行Copy，複製出一個新的容器，然後新的容器裡新增元素，新增完元素之後，再將原容器的參照指向新的容器。這樣做的好處是我們可以對CopyOnWrite容器進行並行的讀，而不需要加鎖，因為當前容器不會新增任何元素。所以CopyOnWrite容器也是一種讀寫分離的思想，讀和寫不同的容器。
讀的時候不需要加鎖，如果讀的時候有多個執行緒正在向CopyOnWriteArrayList新增資料，讀還是會讀到舊的資料。CopyOnWrite並行容器用於讀多寫少的並行場景。
CopyOnWriteArrayList 比 Vector厲害在哪裡？
CopyOnWriteArrayList底層採用了Lock鎖，是JDK層面的，效率高！
Vector底層採用了Synchronized加鎖的方式，保證了資料的安全性，但是效率低下！
解釋一下：Synchronized是Java內建的機制，是JVM層面的，效率低是因為底層操作依賴於作業系統，作業系統切換執行緒要從使用者態切換到核心態，花費很多時間。
優點：
- CopyOnWriteArrayList 並行安全且效能比 Vector 好。Vector 是增刪改查方法都加了synchronized 來保證同步，但是每個方法執行的時候都要去獲得鎖，效能就會大大下降，而 CopyOnWriteArrayList 只是在增刪改上加鎖，但是讀不加鎖，在讀方面的效能就好於 Vector。
缺點：
-
資料一致性問題。這種實現只是保證資料的最終一致性，不能保證資料的實時一致性。在新增到拷貝資料而還沒進行替換的時候，讀到的仍然是舊資料。
-
記憶體佔用問題。如果物件比較大，頻繁地進行替換會消耗記憶體，從而引發 Java 的 GC 問題，這個時候，我們應該考慮其他的容器，例如 ConcurrentHashMap。
2、Set
Set和List同樣是多執行緒下不安全的集合類，同樣會報並行修改異常！
/**
* Set集合
*/
public class TestSet {
public static void main(String[] args) {
Set<String> set = new HashSet<>();
for (int i = 1; i <= 20; i++) {
new Thread(() -> {
set.add(UUID.randomUUID().toString().substring(0, 5));
System.out.println(set);
}, String.valueOf(i)).start();
}
}
}
解決方案：
import java.util.concurrent.CopyOnWriteArraySet;
public class TestSet {
public static void main(String[] args) {
/**
* 解決方案
* set集合沒有可替換的集合
* 1. Set<String> set = Collections.synchronizedSet(new HashSet<>());
* 2. Set<String> set = new CopyOnWriteArraySet<>();
*/
Set<String> set = new CopyOnWriteArraySet<>();
for (int i = 1; i <= 20; i++) {
new Thread(() -> {
set.add(UUID.randomUUID().toString().substring(0, 5));
System.out.println(set);
}, String.valueOf(i)).start();
}
}
}
和麵試官談到這裡，一般都會問hashSet的底層實現原理。
底層其實就是用hashMap實現的
HashSet底層使用了雜湊表來支援的，特點：儲存快 往HashSet新增元素的時候，HashSet會先呼叫元素的HashCode方法得到元素的雜湊值，然後通過元素的雜湊值經過互斥或移位等運算，就可以算出該元素在雜湊表中的儲存位置。如果算出的元素儲存的位置目前沒有任何元素儲存，那麼該元素可以直接儲存在該位置上；如果算出的元素的儲存位置上目前已經有了其他的元素，那麼還會呼叫該元素的equals方法 ，與該位置的元素進行比較一次，如果過equals方法返回的是true，那麼該位置上的元素就會被視為重複元素，不允許被新增，如果false，則允許新增。
3、Map
和List、Set一樣，也會有並行修改異常！
import java.util.concurrent.ConcurrentHashMap;
/**
* Map集合
*/
public class TestMap {
public static void main(String[] args) {
//預設等價於：new HashMap<>(16,0.75);
//初始容量、載入因子
/**
* 解決方案
* 1. Map<String, String> hashMap = Collections.synchronizedMap(new HashMap<>());
* 2. Map<String, String> hashMap = new ConcurrentHashMap<>();
*/
Map<String, String> hashMap = new ConcurrentHashMap<>();
for (int i = 0; i < 20; i++) {
new Thread(()->{
hashMap.put(Thread.currentThread().getName(), UUID.randomUUID().toString().substring(0,5));
System.out.println(hashMap);
},String.valueOf(i)).start();
}
}
}
六、徹底瞭解Callable
Callable是JUC包下的，檢視以下官方檔案
得到結論：
- 返回結果
- 丟擲異常
- 方法不同，run()/call()
泛型的引數等於方法的返回值。
Thread構造引數只認識Runnable，那我們繼續檢視Runnable官方檔案
裡面有一個實現類：FutureTask，繼續往下看
FutureTask的構造方法接受Callable引數。我們可以通過這個適配類來啟動執行緒，這下就圓滿了！
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.FutureTask;
/**
* Callable
*/
public class TestCallable {
public static void main(String[] args) throws ExecutionException, InterruptedException {
//思考：怎麼啟動這個執行緒
MyThread myThread = new MyThread();
FutureTask task = new FutureTask(myThread);//適配類
new Thread(task,"A").start();//執行緒的啟動方式，有且只有一個
new Thread(task,"B").start();//結果會被快取，提高效率。
//這個get()方法可能會阻塞，因為它要等call()方法執行完畢，等待結果返回，一般放到最後一行執行，或者通過非同步通訊來處理！
Integer num = (Integer) task.get();//獲取Callable的返回值
System.out.println(num);
}
}
class MyThread implements Callable<Integer> {
@Override
public Integer call() {
System.out.println("call()");
//如果這裡是耗時操作，返回值就會等待
return 1024;
}
}
//輸出結果：
call()
1024
Process finished with exit code 0
注意點：1、結果有快取，2、返回值可能阻塞
七、JUC包下的常用輔助類
1、CountDownLatch
import java.util.concurrent.CountDownLatch;
/**
* CountDownLatch計數器
*/
public class TestCountDownLatch {
public static void main(String[] args) throws InterruptedException {
//它是一個減法計數器
CountDownLatch countDownLatch = new CountDownLatch(6);
for (int i = 1; i <= 6; i++) {
new Thread(()->{
System.out.println(Thread.currentThread().getName()+" Go Out！");
countDownLatch.countDown();//數量-1
},String.valueOf(i)).start();
}
countDownLatch.await();//等待計數器歸零，然後才向下執行。
System.out.println("Close Door");
//想象放學了，學生一個一個的走出教室，等待學生全部走完了，才能鎖門！
}
}
2、CyclicBarrier
看不懂？沒關係，我也看不懂，咱看程式碼！
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
/**
* CyclicBarrier計數器
*/
public class TestCyclicBarrier {
public static void main(String[] args) {
//它是一個加法計數器
//要求：集齊七顆龍珠召喚神龍！
CyclicBarrier cyclicBarrier = new CyclicBarrier(7,()->{
System.out.println("恭喜你！召喚神龍成功");
});
for (int i = 1; i <= 7; i++) {
//jdk1.8不需要加final
int finalI = i;
new Thread(()->{
System.out.println(Thread.currentThread().getName()+"獲得了第"+ finalI +"顆龍珠");
try {
cyclicBarrier.await();//等待集齊
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
//假如需要集齊8顆龍珠，只集齊了7顆，程式一直等待。
}
}
3、Semaphore
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* Semaphore號誌
*/
public class TestSemaphore {
public static void main(String[] args) {
//執行緒數量：停車位，限流
Semaphore semaphore = new Semaphore(3);
for (int i = 1; i <= 6; i++) {
int finalI = i;
new Thread(()->{
try {
semaphore.acquire();//acquire()得到許可證
System.out.println(Thread.currentThread().getName()+"搶到了車位");
TimeUnit.SECONDS.sleep(2);//睡2秒
System.out.println(Thread.currentThread().getName()+"釋放了車位");
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
semaphore.release();//release()釋放許可證
}
},String.valueOf(i)).start();
//想象搶車位，總共有3個車位，6臺車去搶，先搶到了3個車位的車用完之後就走了，剩下的3個去搶車位
}
}
}
semaphore.acquire()：獲得資源，如果資源已經使用完了，就等待資源釋放後再進行使用！
semaphore.release()：釋放資源，會將當前的號誌釋放+1，然後喚醒等待的執行緒！
作用： 多個共用資源互斥的使用！並行限流，控制最大的執行緒數！
八、讀寫鎖
讀寫鎖（Readers-Writer Lock）顧名思義是一把鎖分為兩部分：讀鎖和寫鎖，其中讀鎖允許多個執行緒同時獲得，因為讀操作本身是執行緒安全的，而寫鎖則是互斥鎖，不允許多個執行緒同時獲得寫鎖，並且寫操作和讀操作也是互斥的。總結來說，讀寫鎖的特點是：讀讀不互斥、讀寫互斥、寫寫互斥。
ReadWriteLock讀寫鎖：讀的時候可以被多執行緒共用，寫的時候只能一個執行緒去寫
/**
* ReadWriteLock讀寫鎖
*/
public class TestReadWriteLock {
public static void main(String[] args) {
MyCache myCache = new MyCache();
for (int i = 1; i <= 5; i++) {
int finalI = i;
new Thread(() -> {
myCache.put(String.valueOf(finalI), String.valueOf(finalI));
}, String.valueOf(i)).start();
}
for (int i = 1; i <= 5; i++) {
int finalI = i;
new Thread(() -> {
myCache.get(String.valueOf(finalI));
}, String.valueOf(i)).start();
}
}
}
//自定義快取
class MyCache {
//volatile保證了變數的可見性，後面會細講
private volatile Map<String, Object> map = new HashMap<>();
//沒加鎖導致執行緒插隊寫入。
//存--->寫
public void put(String key, Object value) {
System.out.println(Thread.currentThread().getName() + "開始寫入");
map.put(key, value);
System.out.println(Thread.currentThread().getName() + "寫入完成~");
}
//取--->讀
public void get(String key) {
System.out.println(Thread.currentThread().getName() + "開始讀取");
map.get(key);
System.out.println(Thread.currentThread().getName() + "讀取完成~");
}
}
2開始寫入
4開始寫入 //2還沒寫入完成，就被其他執行緒插進來了。
1開始寫入
3開始寫入
1寫入完成~
4寫入完成~
2寫入完成~
Process finished with exit code 0
注意：這裡我們可以採用重量鎖Synchronized和輕量鎖Lock來保證資料的安全。
但是我們這裡使用更加細粒度的ReadWriteLock讀寫鎖來完成。
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
//讀讀並行、讀寫互斥、寫寫互斥
public class TestReadWriteLock {
public static void main(String[] args) {
MyCache myCache = new MyCache();
for (int i = 1; i <= 5; i++) {
int finalI = i;
new Thread(() -> {
myCache.put(String.valueOf(finalI), String.valueOf(finalI));
}, String.valueOf(i)).start();
}
for (int i = 1; i <= 5; i++) {
int finalI = i;
new Thread(() -> {
myCache.get(String.valueOf(finalI));
}, String.valueOf(i)).start();
}
}
}
//自定義快取
class MyCache {
//volatile保證了變數的可見性，後面會細講
private volatile Map<String, Object> map = new HashMap<>();
//讀寫鎖
private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
//存--->寫
public void put(String key, Object value) {
readWriteLock.writeLock().lock();//加寫鎖
try {
System.out.println(Thread.currentThread().getName() + "開始寫入");
map.put(key, value);
System.out.println(Thread.currentThread().getName() + "寫入完成~");
} catch (Exception e) {
e.printStackTrace();
}finally {
readWriteLock.writeLock().unlock();//解寫鎖
}
}
//取--->讀
public void get(String key) {
readWriteLock.readLock().lock();//加讀鎖
try {
System.out.println(Thread.currentThread().getName() + "開始讀取");
map.get(key);
System.out.println(Thread.currentThread().getName() + "讀取完成~");
} catch (Exception e) {
e.printStackTrace();
}finally {
readWriteLock.readLock().unlock();//解讀鎖
}
}
}
1開始寫入
1寫入完成~
3開始寫入
3寫入完成~
4開始寫入
4寫入完成~
2開始寫入
2寫入完成~
5開始寫入
5寫入完成~
2開始讀取
1開始讀取
3開始讀取
3讀取完成~
1讀取完成~
5開始讀取
5讀取完成~
4開始讀取
4讀取完成~
2讀取完成~
Process finished with exit code 0
讀鎖與寫鎖互斥，加入讀鎖是為了防止讀的時候寫執行緒進來，打破原子性。
你們可能在外面聽到的讀寫鎖叫：
- 共用鎖（讀鎖）：多個執行緒共用資源。
- 排它鎖又或者獨佔鎖（寫鎖）：單個執行緒獨佔資源。
1、為什麼會有讀寫鎖？
Synchronized 和 ReentrantLock 都是獨佔鎖，即在同一時刻只有一個執行緒獲取到鎖。
然而在有些業務場景中，我們大多在讀取資料，很少寫入資料，這種情況下，如果仍使用獨佔鎖，效率將及其低下。
於是Java就提供了ReentrantReadWriteLock鎖來達到保證執行緒安全的前提下提高並行效率。
九、阻塞佇列
在瞭解阻塞佇列之前，先了解佇列。
1、什麼是佇列？
Queue：佇列，是一種特殊的線性表，是一種先進先出(FIFO)的資料結構。(FIFO:First In First Out)
在Java中，佇列是一種基本的集合型別。與List和Set是同級別的，都有共同父類別Collection介面。
它只允許在表的前端（front）進行刪除操作，而在表的後端（rear）進行插入操作。進行插入操作的端稱為隊尾，進行刪除操作的端稱為隊頭。
在現實生活中佇列也很常見，比如排隊買票，排隊上車，食堂打飯一個個等待處理，這些都是一種佇列模型。而且在程式設計中佇列使用也非常廣泛：比如多執行緒中等待處理的任務、排隊等待獲取某個鎖的執行緒等。
Queue的介面定義如下：
public interface Queue<E> extends Collection<E> {
//向佇列中新增一個元素；如果有空間則新增成功返回true，否則則丟擲IllegalStateException異常
boolean add(E e);
//向佇列中新增一個元素；如果有空間則新增成功返回true，否則返回false
boolean offer(E e);
//從佇列中刪除一個元素；如果元素存在則返回隊首元素，否則丟擲NoSuchElementException異常
E remove();
//從佇列中刪除一個元素；如果元素存在則返回隊首元素，否則返回null
E poll();
//從佇列獲取一個元素，但是不刪除；如果元素存在則返回隊首元素，否則丟擲NoSuchElementException異常
E element();
//從佇列獲取一個元素，但是不刪除；如果元素存在則返回隊首元素，否則返回null
E peek();
}
2、什麼是雙端佇列？
Deque：雙端佇列，是一種具有佇列和棧的性質的資料結構。是限定插入和刪除操作在表的兩端進行的線性表。（double-ended queue，雙端佇列）
在雙端佇列中，使用first和last來表示佇列的首、尾兩端，可以在雙端進行元素的插入和刪除。
offerFirst(A)表示在隊首進行元素插入，pollLast()表示在隊尾進行元素刪除。
可以看到Deque是Queue的子介面，這就表示除了擁有Queue介面的方法之外，還定義了一些特殊的雙端佇列方法。
public interface Deque<E> extends Queue<E> {
// 向隊首新增一個元素；如果有空間則新增成功返回true，否則則丟擲`IllegalStateException`異常
void addFirst(E e);
// 向隊尾新增一個元素；如果有空間則新增成功返回true，否則則丟擲`IllegalStateException`異常
void addLast(E e);
// 向隊首新增一個元素；如果有空間則新增成功返回true，否則返回false
boolean offerFirst(E e);
// 向隊尾新增一個元素；如果有空間則新增成功返回true，否則返回false
boolean offerLast(E e);
// 從隊首刪除一個元素；如果元素存在則返回隊首元素，否則丟擲`NoSuchElementException`異常
E removeFirst();
// 從隊尾刪除一個元素；如果元素存在則返回隊尾元素，否則丟擲`NoSuchElementException`異常
E removeLast();
// 從隊首刪除一個元素；如果元素存在則返回隊首元素，否則返回null
E pollFirst();
// 從隊尾刪除一個元素；如果元素存在則返回隊首元素，否則返回null
E pollLast();
// 從隊首獲取一個元素，但是不刪除；如果元素存在則返回隊首元素，否則丟擲`NoSuchElementException`異常
E getFirst();
// 從隊尾獲取一個元素，但是不刪除；如果元素存在則返回隊尾元素，否則丟擲`NoSuchElementException`異常
E getLast();
// 從隊首獲取一個元素，但是不刪除；如果元素存在則返回隊首元素，否則返回null
E peekFirst();
// 從隊尾獲取一個元素，但是不刪除；如果元素存在則返回隊尾元素，否則返回null
E peekLast();
// 如果元素o存在，則從佇列中刪除第一次出現的該元素
boolean removeFirstOccurrence(Object o);
// 如果元素o存在，則從佇列中刪除最後一次出現的該元素
boolean removeLastOccurrence(Object o);
// 其他方法省略....
}
它們其中有一個實現類：LinkedList是基於連結串列實現的List的一個資料集合，LinkedList還實現了Queue介面和Deque介面。也就是說我們還可以直接使用LinkedList來實現佇列的操作。這裡就不舉例了。
Java中Queue的實現其實是非執行緒安全的，如果在多執行緒環境下進行Queue的入隊和出隊操作，會產生不一致的情況。所以Java也提供了執行緒安全的佇列類——阻塞佇列BlockingQueue。
3、什麼是阻塞佇列？
在Java中，提供了兩種執行緒安全佇列的實現方式：一種是阻塞機制，另一種是非阻塞機制。
使用阻塞機制的佇列，是通過鎖實現的，在入隊和出隊時通過加鎖避免並行操作。而使用非阻塞機制的佇列，是通過使用CAS方式實現，比ConcurrentLinkedQueue。
BlockingQueue：阻塞佇列，是基於阻塞機制實現的執行緒安全的佇列。而阻塞機制的實現是通過在入隊和出隊時加鎖的方式避免並行操作。
BlockingQueue不同於普通的Queue的區別主要是：
- 通過在入隊和出隊時進行加鎖，保證了佇列執行緒安全
- 支援阻塞的入隊和出隊方法：當佇列滿時，會阻塞入隊的執行緒，直到佇列不滿；當佇列為空時，會阻塞出隊的執行緒，直到佇列中有元素。
BlockingQueue常用於生產者-消費者模型中，往佇列裡新增元素的是生產者，從佇列中獲取元素的是消費者；通常情況下生產者和消費者都是由多個執行緒組成；下圖所示則為一個最常見的生產者-消費者模型，生產者和消費者之間通過佇列平衡兩者的的處理能力、進行解耦等。
BlockingQueue介面定義如下：
public interface BlockingQueue<E> extends Queue<E> {
//入隊一個元素，有空間則新增並返回true，沒有則丟擲IllegalStateException異常。
boolean add(E e);
//入隊一個元素，有空間則新增並返回true，沒有則返回false。
boolean offer(E e);
//入隊一個元素，有空間則新增，沒有則一直阻塞等待。
void put(E e) throws InterruptedException;
//入隊一個元素，有空間則新增並返回true，沒有則等timeout時間，新增失敗則返回false
boolean offer(E e, long timeout, TimeUnit unit)
throws InterruptedException;
//出隊一個元素，有元素就出隊，沒有則一直阻塞等待。
E take() throws InterruptedException;
//出隊一個元素，有元素就出隊，沒有元素則等timeout時間，出隊失敗則返回false
E poll(long timeout, TimeUnit unit)
throws InterruptedException;
//返回該佇列剩餘的容量（如果沒有限制則返回Integer.MAX_VALUE）
int remainingCapacity();
//如果元素o在佇列中存在，則從佇列中刪除
boolean remove(Object o);
//判斷佇列中是否存在元素o
public boolean contains(Object o);
//將佇列中的所有元素出隊，並新增到給定的集合c中，返回出隊的元素數量
int drainTo(Collection<? super E> c);
//將佇列中的元素出隊，限制數量maxElements個，並新增到給定的集合c中，返回出隊的元素數量
int drainTo(Collection<? super E> c, int maxElements);
}
BlockingQueue主要提供了四類方法：
| 方法 | 丟擲異常 | 有返回值 | 阻塞等待 | 超時等待 |
|---|---|---|---|---|
| 入隊 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 出隊 | remove() | poll() | take() | poll(time,unit) |
| 獲取隊首元素 | element() | peek() | 沒有 | 沒有 |
BlockingQueue除了和Queue相同的丟擲異常和有返回值方法之外，還提供了兩種阻塞方法：
- 當佇列沒有空間/元素時，一直阻塞。
- 在指定時間嘗試入隊/出隊。
BlockingQueue實現類：
| 實現類 | 功能 |
|---|---|
| ArrayBlockingQueue | 基於陣列的阻塞佇列，維護了一個定長陣列，以便快取佇列中的資料物件，所以是有界佇列 |
| LinkedBlockingQueue | 基於連結串列的阻塞佇列，其內部也維持著一個資料緩衝佇列（該佇列由一個連結串列構成），預設是一個無界佇列；也可以通過構造方法中的capacity設定最大元素數量，所以也可以是有界佇列 |
| SynchronousQueue | 一種沒有緩衝的佇列，生產者產生的資料直接會被消費者獲取並且立刻消費，所以是無界佇列 |
| PriorityBlockingQueue | 基於優先順序的阻塞佇列，底層基於陣列實現，是一個無界佇列 |
| DelayQueue | 延遲佇列，其中的元素只有到了其指定的延遲時間，才能夠從佇列中出隊 |
其中在日常開發中用的比較多的是 ArrayBlockingQueue 和 LinkedBlockingQueue。
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* 阻塞佇列
*/
public class TestArrayBlockingQueue {
public static void main(String[] args) throws InterruptedException {
test4();
}
/**
* 丟擲異常
*/
public static void test1(){
//需要指定佇列的大小
ArrayBlockingQueue blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.add("a"));
System.out.println(blockingQueue.add("b"));
System.out.println(blockingQueue.add("c"));
//Exception in thread "main" java.lang.IllegalStateException: Queue full
//System.out.println(blockingQueue.add("d"));
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
System.out.println(blockingQueue.remove());
//Exception in thread "main" java.util.NoSuchElementException
//System.out.println(blockingQueue.remove());
//System.out.println(blockingQueue.element());//獲得佇列首個元素
}
/**
* 有返回值，不丟擲異常
*/
public static void test2(){
ArrayBlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.offer("張三"));
System.out.println(blockingQueue.offer("李四"));
System.out.println(blockingQueue.offer("王五"));
System.out.println(blockingQueue.offer("王五"));//false，不丟擲異常
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());//null
//System.out.println(blockingQueue.peek());//獲得佇列首個元素
}
/**
* 阻塞等待（一直等待）
*/
public static void test3() throws InterruptedException {
ArrayBlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);
blockingQueue.put("黃子澄");
blockingQueue.put("路哈");
blockingQueue.put("灌水同");
//blockingQueue.put("李成");//佇列沒有空間，程式一直等待。
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
System.out.println(blockingQueue.take());
//System.out.println(blockingQueue.take());//一直阻塞
}
/**
* 阻塞等待（超時等待）
*/
public static void test4() throws InterruptedException {
ArrayBlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);
System.out.println(blockingQueue.offer("唱"));
System.out.println(blockingQueue.offer("跳"));
System.out.println(blockingQueue.offer("rap"));
//System.out.println(blockingQueue.offer("雞你太美", 5, TimeUnit.SECONDS));//等待5秒後如果還是失敗則終止程式並返回false。
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll());
System.out.println(blockingQueue.poll(5,TimeUnit.SECONDS));//等待5秒如果還是失敗終止程式並返回null;
}
}
十、同步佇列
SynchronousQueue同步佇列同樣也是BlockingQueue介面的實現類，它是一種無緩衝的等待佇列，也就是說存入一個元素，必須等待取出來之後，才能繼續往裡面佇列存。
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
/**
* 同步佇列
*/
public class TestSynchronousQueue {
public static void main(String[] args) {
//宣告一個同步佇列
SynchronousQueue<String> blockingQueue = new SynchronousQueue<>();
new Thread(() -> {
try {
System.out.println(Thread.currentThread().getName() + "put 1");
blockingQueue.put(String.valueOf(1));
System.out.println(Thread.currentThread().getName() + "put 2");
blockingQueue.put(String.valueOf(2));
System.out.println(Thread.currentThread().getName() + "put 3");
blockingQueue.put(String.valueOf(3));
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "A").start();
new Thread(() -> {
try {
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName() + "=>" + blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName() + "=>" + blockingQueue.take());
TimeUnit.SECONDS.sleep(3);
System.out.println(Thread.currentThread().getName() + "=>" + blockingQueue.take());
} catch (InterruptedException e) {
e.printStackTrace();
}
}, "B").start();
}
}
A put 1
B=>1
A put 2
B=>2
A put 3
B=>3
Process finished with exit code 0
SynchronousQueue同步佇列與其他的BlockingQueue不同，同步佇列沒有容量，也可以稱為容量為1的佇列，put了一個元素，就必須從裡面先take出來，否則一直阻塞。
十一、執行緒池（重點）
之前我們是使用執行緒的時候就去建立一個執行緒，看起來很方便快捷，但是如果並行的執行緒數量較多時，而每一個執行緒且都只是執行一個時間很短的任務，這樣會引發一些問題：
-
執行緒的建立和銷燬都需要時間，執行緒數量過多時，耗費大量時間，影響效率。
-
頻繁的建立和銷燬執行緒會佔用大量的記憶體，還可能引發記憶體抖動，頻繁觸發GC，最直接的表現就是卡頓。長而久之，記憶體資源佔用過多或者記憶體碎片過多，系統甚至會出現OOM。
-
在作業系統中，CPU都是遵循時間片輪轉機制進行處理任務，執行緒數量過多時，必然會引發CPU頻繁的進行執行緒上下文切換。這個代價是非常昂貴的。
有沒有一種技術來優化這些資源的使用嘞？
1、池化技術
把一些能夠複用的東西（比如說資料庫連線、執行緒）放到池中，避免重複建立、銷燬的開銷，從而極大提高效能。
在Java中執行緒池就可以達到這樣的效果。
2、執行緒池的概念
執行緒池，本質上是一種物件池，用於管理執行緒資源。在任務執行前，需要從執行緒池中拿出執行緒來執行。在任務執行完成之後，需要把執行緒放回執行緒池。通過執行緒的這種反覆利用機制，可以有效地避免直接建立執行緒所帶來的壞處。
3、執行緒池的使用
四大方法、七大引數、四種拒絕策略。
1. 四大方法。
Java通過了JUC包下的Executors工具類提供了4種建立執行緒池的方式。
newFixedThreadPool
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 建立固定執行緒數量的執行緒池
* 建立一個固定大小的執行緒池，該方法可指定執行緒池的固定大小，對於超出的執行緒會在LinkedBlockingQueue佇列中等待
* 核心執行緒數可以指定，執行緒空閒時間為0
*/
public class TestCreatePool {
public static void main(String[] args) {
ExecutorService threadPool = Executors.newFixedThreadPool(5);
try {
for (int i = 0; i < 10; i++) {
//通過執行緒池開啟執行緒
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
} catch (Exception e) {
e.printStackTrace();
}finally {
//關閉執行緒池
threadPool.shutdown();
}
}
}
/*
pool-1-thread-2
pool-1-thread-5
pool-1-thread-4
pool-1-thread-3
pool-1-thread-3
pool-1-thread-3
pool-1-thread-1
pool-1-thread-4
pool-1-thread-5
pool-1-thread-2
*/
newCachedThreadPool
/**
* 可快取無界執行緒池
* 當執行緒池中的執行緒空閒時間超過60s則會自動回收空執行緒
* 當任務超過執行緒池的執行緒數則建立新執行緒。執行緒池的大小上限Integer.MAX_VALUE，約等於21億
*/
public class TestCreatePool {
public static void main(String[] args) {
ExecutorService threadPool = Executors.newCachedThreadPool();
try {
for (int i = 0; i < 10; i++) {
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
} catch (Exception e) {
e.printStackTrace();
}finally {
//關閉執行緒池
threadPool.shutdown();
}
}
}
/*
pool-1-thread-1
pool-1-thread-5
pool-1-thread-4
pool-1-thread-7
pool-1-thread-3
pool-1-thread-2
pool-1-thread-6
pool-1-thread-8
pool-1-thread-10
pool-1-thread-9
*/
newSingleThreadExecutor
/**
* 建立一個單執行緒化的執行緒池
* 該方法無引數，所有任務都儲存佇列LinkedBlockingQueue中，核心執行緒數為1，執行緒空閒時間為0
* 等待唯一的單執行緒來執行任務，並保證所有任務按照指定順序(FIFO或優先順序)執行
*/
public class TestCreatePool {
public static void main(String[] args) {
ExecutorService threadPool = Executors.newSingleThreadExecutor();
try {
for (int i = 0; i < 10; i++) {
threadPool.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
} catch (Exception e) {
e.printStackTrace();
}finally {
//關閉執行緒池
threadPool.shutdown();
}
}
}
/*
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
*/
newScheduledThreadPool
/**
* 建立一個定長的執行緒池
* 可以指定執行緒池核心執行緒數，支援定時及週期性任務的執行
*/
public class TestCreatePool {
public static void main(String[] args) {
ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);
//延遲5s後執行一次任務
threadPool.schedule(()->{
System.out.println(Thread.currentThread().getName() + ",5s");
},5,TimeUnit.SECONDS);
//定時任務：延遲2s後，每3s一次地週期性執行任務
ScheduledFuture<?> scheduledFuture = threadPool.scheduleAtFixedRate(() -> {
System.out.println(Thread.currentThread().getName() + ",3s");
}, 2, 3, TimeUnit.SECONDS);
//20秒之後關閉週期性執行任務
threadPool.schedule(()->{
scheduledFuture.cancel(true);
},20 ,TimeUnit.SECONDS);
//延遲0秒，等上個任務執行完才開始2s計時
threadPool.scheduleWithFixedDelay(()->{
System.out.println(new SimpleDateFormat("HH:mm:ss").format(new Date()););
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
},0,2,TimeUnit.SECONDS);
// schedule(Runnable command, long delay, TimeUnit unit)，延遲一定時間後執行Runnable裡的任務；
// scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)，延遲一定時間後，以間隔period時間的頻率週期性地執行任務；
// scheduleWithFixedDelay(Runnable command, long initialDelay, long delay,TimeUnit unit)，與 scheduleAtFixedRate()方法很類似，
/* 但是不同的是
scheduleAtFixedRate ，是以上一個任務開始的時間計時，period時間過去後，檢測上一個任務是否執行完畢，如果上一個任務執行完畢，則當前任務立即執行，如果上一個任務沒有執行完畢，則需要等上一個任務執行完畢後立即執行。
scheduleWithFixedDelay，是以上一個任務結束時開始計時，period時間過去後，立即執行。
*/
// 注意：執行緒池一旦關閉，週期任務不再執行。
}
}
看一下阿里巴巴開發手冊並行程式設計的規範
叫我們執行緒池不要使用Executors建立，要用ThreadPoolExecutor建立。
四大方法原始碼分析
//newFixedThreadPool
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
//newCachedThreadPool
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
//newSingleThreadExecutor
public static ExecutorService newSingleThreadExecutor() {
return new FinalizableDelegatedExecutorService
(new ThreadPoolExecutor(1, 1,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>()));
}
//newScheduledThreadPool
public ScheduledThreadPoolExecutor(int corePoolSize) {
super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
new DelayedWorkQueue());
}
本質還是建立一個ThreadPoolExecutor。
2. 七大引數
//ThreadPoolExecutor原始碼分析
public ThreadPoolExecutor(int corePoolSize,//核心執行緒大小
int maximumPoolSize,//最大核心執行緒大小
long keepAliveTime,//存活時間
TimeUnit unit,//超時單位
BlockingQueue<Runnable> workQueue,//阻塞佇列
ThreadFactory threadFactory,//執行緒工廠
RejectedExecutionHandler handler//拒絕策略) {
if (corePoolSize < 0 ||
maximumPoolSize <= 0 ||
maximumPoolSize < corePoolSize ||
keepAliveTime < 0)
throw new IllegalArgumentException();
if (workQueue == null || threadFactory == null || handler == null)
throw new NullPointerException();
this.acc = System.getSecurityManager() == null ?
null :
AccessController.getContext();
this.corePoolSize = corePoolSize;
this.maximumPoolSize = maximumPoolSize;
this.workQueue = workQueue;
this.keepAliveTime = unit.toNanos(keepAliveTime);
this.threadFactory = threadFactory;
this.handler = handler;
}
畫個圖理解每個引數的含義
3. 四大拒絕策略
跟蹤原始碼
檢視官方檔案解釋
具體是什麼意思我們通過程式碼測試。
4. 手動建立執行緒池
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
/**
* 使用ThreadPoolExecutor建立執行緒池
*/
public class TestThreadPool {
public static void main(String[] args) {
//工作中通過ThreadPoolExecutor手動建立執行緒池
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
2,//corePoolSize核心執行緒數
5,//maximumPoolSize最大執行緒數
5,//keepAliveTime執行緒存活時間
TimeUnit.SECONDS,//存活時間單位
new LinkedBlockingDeque<>(3),//workQueue阻塞佇列
Executors.defaultThreadFactory(),//threadFactory執行緒工廠一般使用預設就行
new ThreadPoolExecutor.AbortPolicy());//handler拒絕策略
try {
//使用執行緒池來建立執行緒
//迴圈次數就代表執行緒數
for (int i = 1; i <= 2; i++) {
poolExecutor.execute(()->{
System.out.println(Thread.currentThread().getName());
});
}
} catch (Exception e) {
e.printStackTrace();
}finally {
//執行緒池用完，程式結束，關閉執行緒池
poolExecutor.shutdown();
}
}
}
//第一次2個執行緒，正常處理（現有2個執行緒處理業務）
pool-1-thread-1
pool-1-thread-2
Process finished with exit code 0
//第二次5個執行緒，核心執行緒只有2個處理業務，剩下3個執行緒在阻塞佇列（現有2個執行緒處理）
pool-1-thread-1
pool-1-thread-2
pool-1-thread-1
pool-1-thread-2
pool-1-thread-1
Process finished with exit code 0
//第三次6個執行緒，核心執行緒和阻塞佇列都滿了，於是開啟1個最大執行緒視窗處理（現有3個執行緒處理業務）
pool-1-thread-1
pool-1-thread-3
pool-1-thread-2
pool-1-thread-3
pool-1-thread-1
pool-1-thread-2
Process finished with exit code 0
//第四次7個執行緒，開啟2個最大執行緒視窗處理（現有4個執行緒處理業務）
pool-1-thread-2
pool-1-thread-3
pool-1-thread-1
pool-1-thread-3
pool-1-thread-4
pool-1-thread-2
pool-1-thread-1
Process finished with exit code 0
//第五次8個執行緒，開啟3個最大執行緒視窗處理（現有5個執行緒處理業務）
pool-1-thread-1
pool-1-thread-4
pool-1-thread-4
pool-1-thread-5
pool-1-thread-3
pool-1-thread-2
pool-1-thread-2
pool-1-thread-3
Process finished with exit code 0
//第六次9個執行緒，最大執行緒視窗也慢了，就啟動AbortPolicy拒絕策略（現有5個執行緒處理業務）
pool-1-thread-2
pool-1-thread-4
pool-1-thread-3
pool-1-thread-2
pool-1-thread-5
pool-1-thread-1
pool-1-thread-3
pool-1-thread-4
java.util.concurrent.RejectedExecutionException: Task com.hnguigu.pool.TestThreadPool$$Lambda$1/1831932724@7699a589 rejected from java.util.concurrent.ThreadPoolExecutor@58372a00[Running, pool size = 5, active threads = 2, queued tasks = 0, completed tasks = 6]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
at com.hnguigu.pool.TestThreadPool.main(TestThreadPool.java:30)
Process finished with exit code 0
//CallerRunsPolicy拒絕策略
pool-1-thread-2
pool-1-thread-3
pool-1-thread-4
pool-1-thread-3
main
pool-1-thread-1
pool-1-thread-4
pool-1-thread-2
pool-1-thread-5
Process finished with exit code 0
//DiscardOldestPolicy拒絕策略
pool-1-thread-1
pool-1-thread-4
pool-1-thread-5
pool-1-thread-3
pool-1-thread-3
pool-1-thread-2
pool-1-thread-4
pool-1-thread-1
Process finished with exit code 0
//DiscardPolicy拒絕策略
pool-1-thread-1
pool-1-thread-5
pool-1-thread-2
pool-1-thread-4
pool-1-thread-3
pool-1-thread-3
pool-1-thread-1
pool-1-thread-5
Process finished with exit code 0
- new ThreadPoolExecutor.AbortPolicy()：拒絕處理該任務，並丟擲RejectedExecutionException異常
- new ThreadPoolExecutor.CallerRunsPolicy()：由呼叫執行緒處理該任務，如果呼叫執行緒是主執行緒，那麼主執行緒會呼叫執行器中的execute方法來執行該任務。
- new ThreadPoolExecutor.DiscardOldestPolicy()：嘗試丟棄佇列最前面的任務，然後重新嘗試執行任務（重複此過程）如果競爭失敗還是會被丟棄。
- new ThreadPoolExecutor.DiscardPolicy()：放棄當前任務，不會丟擲異常。
5. 調優
執行緒的本質就是為了執行任務，在計算機的世界裡，任務大致分為兩類，CPU密集型任務和 IO密集型任務。
- CPU密集型任務，比如公式計算、資源解碼等。這類任務要進行大量的計算，全都依賴CPU的運算能力，持久消耗CPU資源。所以針對這類任務，其實不應該開啟大量執行緒。因為執行緒越多，花線上程切換的時間就越多，CPU執行效率就越低，一般CPU密集型任務同時進行的數量等於CPU的核心數，最多再加個1。
- IO密集型任務，比如網路讀寫、檔案讀寫等。這類任務不需要消耗太多的CPU資源，絕大部分時間是在IO操作上。所以針對這類任務，可以開啟大量執行緒去提高CPU的執行效率，一般IO密集型任務同時進行的數量等於CPU的核心數的兩倍。
獲取CPU核心數：Runtime.getRuntime().availableProcessors()
public class TestThreadPool {
public static void main(String[] args) {
System.out.println("CPU核數："+Runtime.getRuntime().availableProcessors());
//工作中通過ThreadPoolExecutor手動建立執行緒池
ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
2,
Runtime.getRuntime().availableProcessors(),//最大核心執行緒數
5,
TimeUnit.SECONDS,
new LinkedBlockingDeque<>(3),
Executors.defaultThreadFactory(),//執行緒工廠一般使用預設就行
new ThreadPoolExecutor.DiscardPolicy());
}
}
當面試官問你，如何設定執行緒池最大核心執行緒數？
答：CPU密集型(CPU是幾核就設定多少)、IO密集型 (判斷程式中十分耗IO的執行緒，然後為其分配，耗IO的執行緒 * 2)---->屬於調優問題。
十二、四大函數式介面
新時代的程式設計師必須要會：lambda表示式、鏈式程式設計、函數式介面、Stream流式計算。
函數式介面：在java中是指有且僅只有一個抽象方法的介面。即適用於函數語言程式設計場景的介面。而Java中的，函數語言程式設計體現就是Lambda表示式，所以函數式介面就是可以適用於Lambda使用的介面。
某個介面上宣告了 @FunctionalInterface 註解，那麼編譯器就會按照函數式介面的定義來要求該介面。
四大函數式介面指的是Consumer、Function、Predicate、Supplier。位於java.util.function包下。
1、Function（函數型介面）
@FunctionalInterface
public interface Function<T, R> {
//接受一個引數T，返回一個結果R
R apply(T t);
}
public class Test {
public static void main(String[] args) {
//相當於一個工具類，可以根據自己的需求來實現其功能。
Function<String, String> fun1 = new Function<String, String>() {
@Override
public String apply(String s) {
return s.toUpperCase();
}
};
//lambda簡化後
Function<String, String> fun2 = s -> s.toUpperCase();
System.out.println(fun2.apply("hello,world"));//HELLO,WORLD
}
}
2、Predicate（斷定型介面）
@FunctionalInterface
public interface Predicate<T> {
//接受一個引數T，返回boolean型別的值
boolean test(T t);
}
public class TestPredicate {
public static void main(String[] args) {
//相當於一個工具類，可以根據自己的需求來實現其功能。
Predicate<String> fun1 = new Predicate<String>() {
@Override
public boolean test(String s) {
return s.isEmpty();
}
};
//lambda簡化後
Predicate<String> fun2 = s -> s.isEmpty();
System.out.println(fun2.test(""));//true
}
}
3、Consumer（消費型介面）
@FunctionalInterface
public interface Consumer<T> {
//接收一個引數T，沒有返回值
void accept(T t);
}
public class TestConsumer {
public static void main(String[] args) {
//因為沒有出參，常用於列印，傳簡訊等
Consumer<String> fun1 = new Consumer<String>() {
@Override
public void accept(String s) {
System.out.println(s);
}
};
//lambda簡化後
Consumer<String> fun2 = s -> System.out.println(s);
fun2.accept("你好啊，陳冠希！");//你好啊，陳冠希！
}
}
4、Supplier（供給型介面）
@FunctionalInterface
public interface Supplier<T> {
//沒有輸入引數，只有返回值T
T get();
}
public class TestSupplier {
public static void main(String[] args) {
//常用於符合條件時呼叫獲取結果；執行結果提前定義，但不執行。
Supplier<Double> fun1 = new Supplier<Double>() {
@Override
public Double get() {
return Math.PI;
}
};
//lambda簡化後
Supplier<Double> fun2 = () -> Math.PI;
System.out.println(fun2.get());//3.141592653589793
}
}
十三、Stream流
Stream 流是 Java 8 新提供給開發者的一組操作集合的 API，將要處理的元素集合看作一種流， 流在管道中傳輸， 並且可以在管道的節點上進行處理， 比如篩選、排序、聚合等。元素流在管道中經過中間操作（intermediate operation）的處理，最後由終端操作 (terminal operation) 得到前面處理的結果。Stream 流可以極大的提高開發效率，也可以使用它寫出更加簡潔明瞭的程式碼。
1、Stream流的建立
常見有三種建立stream流的方式：
- 集合：Collection.stream()
- 陣列：Arrays.stream()
- 靜態方法：Stream.of()
public class TestStream {
public static void main(String[] args) {
//1、使用Collection集合
List<String> list = Arrays.asList("a", "b", "c");
Stream<String> stream1 = list.stream();//建立一個順序流
Stream<String> stringStream = list.parallelStream();//建立一個並行流
//2、使用Arrays.stream(T[] array)方法用陣列建立流
int[] array = {1,2,3,4,5,6};
IntStream stream2 = Arrays.stream(array);
//3、使用Stream.of()靜態方法
Stream<Integer> stream3 = Stream.of(1,2,3,4,5,6);
}
}
stream流是序列流（順序流），由主執行緒按順序對流進行操作，而parallelStream是並行流，內部以多執行緒並行的方式對流進行操作。但前提是流中的資料處理沒有順序要求，如果流中的資料量足夠大，並行流可以加快處速度。除了直接建立並行流，還可以通過parallel()把順序流轉換成並行流：
2、Stream流的使用
java.util.stream包下的Stream介面檢視API檔案
以下案例使用到了：lambda、鏈式程式設計、函數式介面、流式計算
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* 使用者實體類
*/
//get、set、無參、有參
@Data
@NoArgsConstructor
@AllArgsConstructor
public class User {
private int id;
private String name;
private int age;
}
import java.util.Arrays;
import java.util.List;
/**
* 題目要求：一分鐘內完成此題，只能用一行程式碼完成！
* 現有5個使用者，進行篩選！
* 1、ID 必須是偶數
* 2、Age 必須大於23歲
* 3、Name 轉為大寫字母
* 4、Name 倒序
* 5、只輸出一個使用者
*/
public class Test {
public static void main(String[] args) {
User u1 = new User(1, "a", 21);
User u2 = new User(2, "b", 22);
User u3 = new User(3, "c", 23);
User u4 = new User(4, "d", 24);
User u5 = new User(6, "e", 25);
//1、將物件存入集合
List<User> list = Arrays.asList(u1, u2, u3, u4, u5);
//2、通過stream流篩選資料
list.stream()
.filter(user -> user.getId()%2==0)
.filter(user -> user.getAge() > 23)
.map(user -> {
user.setName(user.getName().toUpperCase());
return user;
})
.sorted((user1,user2)->user2.getName().compareTo(user2.getName()))
.limit(1)
.forEach(System.out::println);//方法參照
}
}
//輸出結果：User(id=4, name=D, age=24)
十四、Fork/Join
Fork/Join 框架是 Java7 提供了的一個用於並行執行任務的框架， 是一個把大任務分割成若干個小任務，最終彙總每個小任務結果後得到大任務結果的框架。
Fork/Join適用於巨量資料量計算，小資料量完全沒必要使用。
1、工作竊取
工作竊取（work-stealing）演演算法是指某個執行緒從其他佇列裡竊取任務來執行。
為什麼使用工作竊取演演算法呢？假如我們需要做一個比較大的任務，我們可以把這個任務分割為若干互不依賴的子任務，為了減少執行緒間的競爭，於是把這些子任務分別放到不同的佇列裡，併為每個佇列建立一個單獨的執行緒來執行佇列裡的任務，執行緒和佇列一一對應，比如 A 執行緒負責處理 A 佇列裡的任務。但是有的執行緒會先把自己佇列裡的任務幹完，而其他執行緒對應的佇列裡還有任務等待處理。幹完活的執行緒與其等著，不如去幫其他執行緒幹活，於是它就去其他執行緒的佇列裡竊取一個任務來執行。而在這時它們會存取同一個佇列，所以為了減少竊取任務執行緒和被竊取任務執行緒之間的競爭，通常會使用雙端佇列，被竊取任務執行緒永遠從雙端佇列的頭部拿任務執行，而竊取任務的執行緒永遠從雙端佇列的尾部拿任務執行。
工作竊取演演算法的優點是充分利用執行緒進行平行計算，並減少了執行緒間的競爭，其缺點是在某些情況下還是存在競爭，比如雙端佇列裡只有一個任務時。並且消耗了更多的系統資源，比如建立多個執行緒和多個雙端佇列。
2、Fork/Join的使用
需要用到兩個類：
-
ForkJoinTask：建立一個ForkJoin任務。因為它是一個抽象類，通常不直接使用，通常繼承它的子類來完成。
- RecursiveAction：遞迴事件，用於沒有返回值的任務。
- RecursiveTask：遞迴任務：用於有返回值的任務。
-
ForkJoinPool：ForkJoinTask通過ForkJoinPool來執行。任務分割出的子任務會新增到當前工作執行緒所維護的雙端佇列中，進入佇列的頭部。當一個工作執行緒的佇列裡暫時沒有任務時，它會隨機從其他工作執行緒的佇列的尾部獲取一個任務。
舉個例子：現需要計算一個超大的累加數（10_0000_0000）的和，你會怎麼做？
- 辦法一：用一個迴圈在一個執行緒內完成（效率低）
- 辦法二：把計算的大任務拆分成多個小任務，並行執行。（效率高）
- 辦法三：使用stream並行流（效率封頂！）
分析：使用Fork/Join框架首先考慮如何分割任務，設定一個閾值10000，如果我們要計算的數值超過閾值，那麼就使用Fork/Join來分割任務計算，沒有超過就正常計算。
我們來看如何使用Fork/Join對巨量資料進行並行求和：
import java.util.concurrent.RecursiveTask;
/**
* 1、設定閾值
* 2、ForkJoinTask建立子任務
* 3、ForkJoinPool執行任務
*/
public class ForkJoinDemo extends RecursiveTask<Long> {
//因為數值比較大所以使用Long型別
private static final int THRESHOLD = 10000;//設定閾值
private Long start;
private Long end;
public ForkJoinDemo(Long start, Long end) {
this.start = start;
this.end = end;
}
@Override
protected Long compute() {
Long sum = 0L;
if ((end - start) <= THRESHOLD) {
//如果任務足夠小，我們就正常計算。
for (Long i = start; i <= end; i++) {
sum += i;
}
} else {
//如果任務大於閾值，就分裂子任務計算。
Long middle = (start + end) >> 1; //中間值
//建立分割任務，將一個任務拆分成兩個子任務
ForkJoinDemo f1 = new ForkJoinDemo(start, middle);
ForkJoinDemo f2 = new ForkJoinDemo(middle + 1, end);
//壓入執行緒佇列，非同步執行子任務
f1.fork();
f2.fork();
//等子任務執行完，得到其結果
Long j1 = f1.join();
Long j2 = f2.join();
//最後合併結果
sum = j1 + j2;
}
return sum;
}
}
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.stream.LongStream;
/**
* 測試
*/
public class Test {
public static void main(String[] args) {
//test1();//sum=500000000500000000,耗時=6412
//test2();//sum=500000000500000000,耗時=6101
test3();//sum=500000000500000000,耗時=126
//注意：這裡使用Long包裝類，jvm會幫我們自動裝箱自動拆箱會消費時間
}
//普通方法
public static void test1() {
Long startTime = System.currentTimeMillis();
Long sum = 0L;
for (Long i = 1L; i <= 10_0000_0000; i++) {
sum += i;
}
Long endTime = System.currentTimeMillis();
System.out.println("sum=" + sum + ",耗時=" + (endTime - startTime));
}
//使用ForkJoin的方法
public static void test2() {
try {
Long startTime = System.currentTimeMillis();
//建立ForkJoinPool
ForkJoinPool forkJoinPool = new ForkJoinPool();
//建立任務
ForkJoinTask<Long> forkJoinTask = new ForkJoinDemo(1L, 10_0000_0000L);
//提交任務
ForkJoinTask<Long> submit = forkJoinPool.submit(forkJoinTask);
Long sum = submit.get();
Long endTime = System.currentTimeMillis();
System.out.println("sum=" + sum + ",耗時=" + (endTime - startTime));
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
//使用stream並行流的方法
public static void test3() {
Long startTime = System.currentTimeMillis();
long sum = LongStream.rangeClosed(1L, 10_0000_0000L).parallel().reduce(0, Long::sum);
Long endTime = System.currentTimeMillis();
System.out.println("sum=" + sum + ",耗時=" + (endTime - startTime));
}
}
Fork/Join是一種基於「分治」的演演算法：通過分解任務，並行執行，最後合併結果得到最終結果。
十五、非同步回撥
非同步回撥：在發起一個非同步任務的同時指定一個函數，在非同步任務完成時會呼叫這個函數，這個函數就叫回撥函數。
Future介面表示一個可能還沒有完成的非同步任務的結果，針對這個結果可以新增Callback以便在任務執行成功或失敗後作出相應的操作。
使用 Future 獲得非同步執行結果時，要麼呼叫阻塞方法get()，要麼輪詢看isDone()是否為true，這兩種方法都不是很好，因為主執行緒也會被迫等待。
從Java 8開始引入了 CompletableFuture，它針對 Future 做了改進，可以傳入回撥物件，當非同步任務完成或者發生異常時，自動呼叫回撥物件的回撥方法。
CompletableFuture 是 Future 的實現類。一個completableFuture物件代表著一個任務。
1、建立無返回值的非同步任務
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class Demo01{
public static void main(String[] args) throws ExecutionException, InterruptedException {
//runAsync執行一個非同步任務，沒有返回值
CompletableFuture<Void> future = CompletableFuture.runAsync(()->{
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName());
});
System.out.println("1111111");
future.get();//阻塞獲取執行結果
System.out.println("2222222");
}
}
/*
輸出結果：
1111111
ForkJoinPool.commonPool-worker-9
2222222
*/
2、建立有返回值的非同步任務
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
public class Demo01{
public static void main(String[] args) throws ExecutionException, InterruptedException {
//supplyAsync執行一個非同步任務，有返回值
CompletableFuture<Integer> completableFuture = CompletableFuture.supplyAsync(()-> {
System.out.println(Thread.currentThread().getName());
//int i = 1 / 0;//異常則回撥exceptionally
return 1024;
});
//success回撥函數
CompletableFuture<Integer> whenComplete = completableFuture.whenComplete((t, u) -> {
System.out.println("t=>" + t);//正常返回值
System.out.println("u=>" + u);//丟擲異常的錯誤資訊
}).exceptionally((e) -> {
//error回撥函數
System.out.println(e.getMessage());
//非同步任務執行異常則返回500
return 500;
});
System.out.println(whenComplete.get());//獲取非同步任務執行結果
}
}
/*
正常輸出結果：
ForkJoinPool.commonPool-worker-9
t=>1024
u=>null
1024
異常輸出結果：
ForkJoinPool.commonPool-worker-9
t=>null
u=>java.util.concurrent.CompletionException: java.lang.ArithmeticException: / by zero
java.lang.ArithmeticException: / by zero
500
*/
十六、JMM
JMM就是Java記憶體模型(java memory model)。因為在不同的硬體生產商和不同的作業系統下，記憶體的存取有一定的差異，所以會造成相同的程式碼執行在不同的系統上會出現各種問題。所以java記憶體模型(JMM)遮蔽掉各種硬體和作業系統的記憶體存取差異，以實現讓java程式在各種平臺下都能達到一致的並行效果。
JMM也是屬於JVM的一部分，只是JMM是一種抽象的概念，是一組規則，並不實際存在。JMM規定了記憶體主要劃分為主記憶體和工作記憶體兩種。
JVM在設計時候考慮到，如果Java執行緒每次讀取和寫入變數都直接操作主記憶體，對效能影響比較大，所以每條執行緒擁有各自的工作記憶體，從主記憶體中拷貝一份變數放入工作記憶體中，執行緒對變數的讀和寫都是在自己的工作記憶體中操作，執行完相關的指令和操作，最後寫回主記憶體，而不是直接操作主記憶體中的變數。
但這樣會引發一個問題：當某個執行緒修改了自己工作記憶體中的變數，對其他的執行緒是不可見的，就會導致執行緒不安全問題。
因此JMM制定了一套標準來保證開發者在編寫多執行緒程式的時候，能夠控制什麼時候記憶體會被同步給其他執行緒。
1、記憶體互動操作
記憶體互動操作有8種，虛擬機器器實現必須保證每一個操作都是原子的，不可在分的（對於double和long型別的變數來說，load、store、read和write操作在某些平臺上允許例外）
- lock（鎖定）：作用於主記憶體的變數，把一個變數標識為執行緒獨佔狀態
- unlock（解鎖）：作用於主記憶體的變數，它把一個處於鎖定狀態的變數釋放出來，釋放後的變數才可以被其他執行緒鎖定
- read（讀取）：作用於主記憶體變數，它把一個變數的值從主記憶體傳輸到執行緒的工作記憶體中，以便隨後的load動作使用
- load（載入）：作用於工作記憶體的變數，它把read操作從主記憶體中變數放入工作記憶體中
- use（使用）：作用於工作記憶體中的變數，它把工作記憶體中的變數傳輸給執行引擎，每當虛擬機器器遇到一個需要使用到變數的值，就會使用到這個指令
- assign（賦值）：作用於工作記憶體中的變數，它把一個從執行引擎中接受到的值放入工作記憶體的變數副本中
- store（儲存）：作用於主記憶體中的變數，它把一個從工作記憶體中一個變數的值傳送到主記憶體中，以便後續的write使用
- write（寫入）：作用於主記憶體中的變數，它把store操作從工作記憶體中得到的變數的值放入主記憶體的變數中
JMM對這八種指令的使用，制定瞭如下規則：
- 不允許read和load、store和write操作之一單獨出現。即使用了read必須load，使用了store必須write
- 不允許執行緒丟棄他最近的assign操作，即工作變數的資料改變了之後，必須告知主記憶體
- 不允許一個執行緒將沒有assign的資料從工作記憶體同步回主記憶體
- 一個新的變數必須在主記憶體中誕生，不允許工作記憶體直接使用一個未被初始化的變數。就是懟變數實施use、store操作之前，必須經過assign和load操作
- 一個變數同一時間只有一個執行緒能對其進行lock。多次lock後，必須執行相同次數的unlock才能解鎖
- 如果對一個變數進行lock操作，會清空所有工作記憶體中此變數的值，在執行引擎使用這個變數前，必須重新load或assign操作初始化變數的值
- 如果一個變數沒有被lock，就不能對其進行unlock操作。也不能unlock一個被其他執行緒鎖住的變數
- 對一個變數進行unlock操作之前，必須把此變數同步回主記憶體（執行store和write操作）
2、三大模型特徵
原子性
一次操作或者多次操作，要麼所有的操作全部都得到執行並且不會受到任何因素的干擾而中斷，要麼都不執行。
在 Java 中，可以藉助synchronized 、各種 Lock 以及各種原子類實現原子性。
synchronized 和各種 Lock 可以保證任一時刻只有一個執行緒存取該程式碼塊，因此可以保障原子性。各種原子類是利用 CAS (compare and swap) 操作（可能也會用到 volatile或者final關鍵字）來保證原子操作。
可見性
當一個執行緒對共用變數進行了修改，那麼另外的執行緒都是立即可以看到修改後的最新值。
在 Java 中，可以藉助synchronized 、volatile 以及各種 Lock 實現可見性。
-
volatile關鍵字要求被修改之後的變數要求立即更新到主記憶體，每次使用前從主記憶體處進行讀取。
-
synchronized保證unlock之前必須先把變數重新整理回主記憶體。
-
final修飾的欄位在構造器中一旦完成初始化，並且構造器沒有this逸出，那麼其他執行緒就能看到final欄位的值。
有序性
在本執行緒內觀察，所有的操作都是有序的；而在一個執行緒內觀察另一個執行緒，所有操作都是無序的。前半句指 as-if-serial 語意：執行緒內似表現為序列，後半句是指：「指令重排序現象」和「工作記憶體與主記憶體同步延遲現象」。處理器為了提高程式的執行效率，提高並行效率，可能會對程式碼進行優化。編譯器認為，重排序後的程式碼執行效率更優。這樣一來，程式碼的執行順序就未必是編寫程式碼時候的順序了，在多執行緒的情況下就可能會出錯。
Java 語言提供了 volatile 和 synchronized 兩個關鍵字來保證執行緒之間操作的有序性。
-
volatile關鍵字是使用記憶體屏障達到禁止指令重排序，以保證有序性。
-
synchronized是一個執行緒lock之後，必須unlock後，其他執行緒才可以重新lock，使得被synchronized包住的程式碼塊在多執行緒之間是序列執行的。
同步規定：
- 執行緒解鎖前，必須把共用變數的值重新整理回主記憶體。
- 執行緒加鎖前，必須將主記憶體的最新值讀取到自己的工作記憶體。
- 加鎖和解鎖是同一把鎖。
十七、Volatile
volatile在java語言中是一個關鍵字，用於修飾變數。被volatile修飾的變數後，表示這個變數在不同執行緒中是共用，編譯器與執行時都會注意到這個變數是共用的，因此不會對該變數進行重排序。
特點：1.保證了可見性 2.不保證原子性 3.禁止指令重排。
1、驗證volatile保證可見性
public class TestVolatile {
//標誌位
static boolean flag= true;
public static void main(String[] args) throws InterruptedException {
//子執行緒
new Thread(()->{
while(flag){}
}).start();
TimeUnit.SECONDS.sleep(2);
//main執行緒
flag = false;
System.out.println(flag);//false
}
}
以上的情況，按道理說程式應該是停止的。但事實並不是這樣。程式輸出false，程式死迴圈。
flag變數被拷貝到子執行緒和main執行緒的工作記憶體，子執行緒開始迴圈，此時的flag=true，此時的main執行緒修改flag=false，並寫入了主記憶體，但是子執行緒還是在迴圈，並不知道其他執行緒修改了主記憶體中的變數。
但是此時用volatile關鍵字修飾變數，main執行緒修改完flag變數之後，會立即將變數寫回主記憶體，並且告知子執行緒，子執行緒發現自己的變數失效後，會重新去主記憶體中存取flag變數，此時flag=false，迴圈退出。
//加入volatile保證可見性。
volatile static boolean flag = true;
2、驗證volatile不保證原子性
//以下幾句程式碼能保證原子性嗎？
int i = 2;
int j = i;
i++;
i = i + 1;
第一句是基本型別賦值操作，必定是原子性操作。
第二句先讀取i的值，再賦值到j，兩步操作，不能保證原子性。
第三和第四句其實是等效的，先讀取i的值，再+1，最後賦值到i，三步操作了，不能保證原子性。
JMM只能保證基本的原子性，如果要保證一個程式碼塊的原子性，提供了monitorenter 和 moniterexit 兩個位元組碼指令，也就是 synchronized 關鍵字。因此在 synchronized 塊之間的操作都是原子性的。
public class TestVolatile2 {
//volatile不保證多執行緒條件下的原子性
private volatile static int num = 0;
public static void add(){
num++;//不是原子性操作
}
public static void main(String[] args) {
//理論應該是num = 20000
for (int i = 1; i <= 20; i++) {
new Thread(()->{
for (int j = 0; j < 1000; j++) {
add();
}
}).start();
}
//保證執行緒全部執行完
while(Thread.activeCount()>2){
//如果存活的執行緒大於2，main gc，那就禮讓
Thread.yield();
}
System.out.println(Thread.currentThread().getName()+"=>"+num);
}
}
我們發現所有的執行緒跑完了num的值還是沒有20000，我們之前學了鎖，給add()方法加上synchronized是不是就可以保證原子性了，確實是這樣的。但是我們這裡是要驗證volatile關鍵字是否保證原子性。我們給num關鍵字加上volatile關鍵字，num的值依舊到不了20000，證明了volatile不保證原子性。
這時候面試官就為難你了.
如果不加lock和synchronized怎麼保證原子性？
num++看上去是一行程式碼，其實底層拆分了三行，第一步讀取num的值，第二步num+1，第三步給num賦值。
JUC剩下最後一個java.util.concurrent.atomic包沒講了，這裡直接派上用場。
使用java.util.concurrent.atomic包下的原子類解決問題
import java.util.concurrent.atomic.AtomicInteger;
public class TestVolatile2 {
//原子類的Integer
private static AtomicInteger num = new AtomicInteger();
public synchronized static void add(){
num.getAndIncrement();//atomicInteger+1的方法
}
public static void main(String[] args) {
//理論應該是num = 20000
for (int i = 1; i <= 20; i++) {
new Thread(()->{
for (int j = 0; j < 1000; j++) {
add();
}
}).start();
}
//保證執行緒全部執行完
while(Thread.activeCount()>2){
//如果存活的執行緒大於2，main gc，那就禮讓
Thread.yield();
}
System.out.println(Thread.currentThread().getName()+"=>"+num);//20000
}
}
看方法原始碼，Unsafe類裡面全是native方法，這些類的底層都和作業系統掛鉤！它的num++，是在記憶體中修改值，底層是CAS保證的原子性。
3、禁止指令重排詳解
為了提升執行速度/效能，計算機在執行程式程式碼的時候，會對指令進行重排序。
簡單來說就是系統在執行程式碼的時候並不一定是按照你寫的程式碼的順序依次執行。
我們去執行程式會經歷這樣的過程。
原始碼 ---> 編譯器優化重排 --> 指令並行重排 --> 記憶體系統重排 --> 執行
int x = 1; //1行
int y = 2; //2行
x = x + 2; //3行
y = x * x; //4行
//我們期望的執行順序是 1234，可能執行的時候是 2134 1324
//可不可能是4123？ 不可能的嗷
//處理器在進行指令重排：會考慮，資料之間的依賴性。
指令重排序在單執行緒是沒有問題的，不會影響執行結果，而且還提高了效能。但是在多執行緒的環境下就不能保證一定不會影響執行結果了。
現在 x y a b 預設都是0
| 執行緒A | 執行緒B |
|---|---|
| x = a | y = b |
| b = 1 | a = 2 |
正常執行結果：x = 0，y = 0，指令重排後
| 執行緒A | 執行緒B |
|---|---|
| b = 1 | a = 2 |
| x = a | y = b |
指令重排後執行結果：x = 2，y = 1
volatile可以禁止指令重排
怎麼實現的呢？
指令並行重排和記憶體系統重排都屬於是處理器級別的指令重排序。對於處理器，通過插入記憶體屏障的方式來禁止特定型別的處理器重排序。
記憶體屏障（Memory Barrier，或有時叫做記憶體柵欄，Memory Fence）是一種 CPU 指令，用來禁止處理器指令發生重排序（像屏障一樣），從而保障指令執行的有序性。另外，為了達到屏障的效果，它也會使處理器寫入、讀取值之前，將主記憶體的值寫入快取記憶體，清空無效佇列，從而保障變數的可見性。
volatile禁止重排序原理
在每個volatile讀操作後插入LoadLoad屏障，在讀操作後插入LoadStore屏障。
在每個volatile寫操作的前面插入一個StoreStore屏障，後面插入一個SotreLoad屏障。
面試官：那麼你知道在哪裡用這個記憶體屏障用得最多呢？單例模式
十八、徹底玩轉單例模式
單例模式（Singleton Pattern）是 Java 中最簡單的設計模式之一。這種型別的設計模式屬於建立型模式，它提供了一種建立物件的最佳方式。
這種模式涉及到一個單一的類，該類負責建立自己的物件，同時確保只有單個物件被建立。這個類提供了一種存取其唯一的物件的方式，可以直接存取，不需要範例化該類的物件。
注意：
1、單例類只能有一個範例。
2、單例類必須自己建立自己的唯一範例。
3、單例類必須給所有其他物件提供這一範例。
意圖：保證一個類僅有一個範例，並提供一個存取它的全域性存取點。
主要解決：一個全域性使用的類頻繁地建立與銷燬。
何時使用：當您想控制範例數目，節省系統資源的時候。
如何解決：判斷系統是否已經有這個單例，如果有則返回，如果沒有則建立。
關鍵程式碼：建構函式是私有的。
優點：
- 在記憶體裡只有一個範例，減少了記憶體的開銷，尤其是頻繁的建立和銷燬範例（比如管理學院首頁頁面快取）。
- 避免對資源的多重佔用（比如寫檔案操作）
缺點：沒有介面，不能繼承，與單一職責原則衝突，一個類應該只關心內部邏輯，而不關心外面怎麼樣來範例化。
注意事項：getInstance() 方法中需要使用同步鎖 synchronized (Singleton.class) 防止多執行緒同時進入造成 instance 被多次範例化。
/**
* 單例模式
*/
public class SingleObject {
//建立 SingleObject 的一個物件
private static SingleObject instance = new SingleObject();
//私有化建構函式，這樣該類就不會被範例化
private SingleObject(){}
//獲取唯一可用的物件
public static SingleObject getInstance(){
return instance;
}
public void showMessage(){
System.out.println("Hello,World！");
}
}
public class SingletonPatternDemo {
public static void main(String[] args) {
//不合法的建構函式
//編譯時錯誤：建構函式 SingleObject() 是不可見的
//SingleObject object = new SingleObject();
//獲取唯一可用的物件
SingleObject object = SingleObject.getInstance();
//顯示訊息
object.showMessage();
//Hello World!
}
}
1、單例模式的幾種實現方式
單例模式的實現有多種方式，如下所示：
1. 餓漢式（可用）
優點：沒有加鎖，執行效率會提高。
缺點：在類裝載的時候就完成範例化，沒有達到Lazy Loading的效果。如果從始至終從未使用過這個範例，則會造成記憶體的浪費。
public class Hungry {
private final static Hungry INSTANCE = new Hungry();
private Hungry(){}
public static Hungry getInstance(){
return INSTANCE;
}
}
2. 懶漢式（不可用）
優點：這種方式Lazy Loading的效果很明顯
缺點：因為沒有加鎖 synchronized，所以嚴格意義上它並不算單例模式。不要求執行緒安全，在多執行緒不能正常工作。
public class LazyMan {
private static LazyMan lazyMan;
private LazyMan(){}
public static LazyMan getInstance(){
if(lazyMan == null){
lazyMan = new LazyMan();
}
return lazyMan;
}
}
單執行緒下確實是單例的，看看多執行緒。
/**
* 多執行緒測試懶漢式的單例模式
*/
public class LazyMan {
private static LazyMan lazyMan;
private LazyMan(){
//列印執行緒
System.out.println(Thread.currentThread().getName());
}
public static LazyMan getInstance(){
if(lazyMan == null){
lazyMan = new LazyMan();
}
return lazyMan;
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
LazyMan.getInstance();
}).start();
}
}
}
//多執行緒下出現多個範例
Thread-1
Thread-2
Thread-4
Thread-0
Process finished with exit code 0
解決辦法第一個想到就是給方法加鎖。但是給方法加synchronized鎖，會導致很大的效能開銷，並且加鎖其實只需要在第一次初始化的時候用到，之後的呼叫都沒必要再進行加鎖。
所以我們使用synchronized塊鎖類
public class LazyMan {
private static LazyMan lazyMan;
private LazyMan(){
System.out.println(Thread.currentThread().getName());
}
public static LazyMan getInstance(){
if(lazyMan == null){
synchronized (LazyMan.class){
lazyMan = new LazyMan();
}
}
return lazyMan;
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
LazyMan.getInstance();
}).start();
}
}
}
這種情況還是會出現多範例，如果兩個執行緒同時進入了if判斷，其中A執行緒搶到了鎖，B執行緒就在這裡等待，等A執行緒執行鎖塊里程式碼完成之後就釋放鎖，B執行緒就拿鎖，繼續執行鎖塊程式碼，從而導致建立了多個範例。繼續優化
雙檢鎖/雙重校驗鎖（DCL，即 double-checked locking）（推薦）
優點：這種方式採用雙鎖機制，安全且在多執行緒情況下能保持高效能。
public class LazyMan {
//加入volatile禁止指令重排
private volatile static LazyMan lazyMan;
private LazyMan(){
System.out.println(Thread.currentThread().getName());
}
public static LazyMan getInstance(){
if(lazyMan == null){
synchronized (LazyMan.class){
if(lazyMan == null){
lazyMan = new LazyMan();
}
}
}
return lazyMan;
}
public static void main(String[] args) {
for (int i = 0; i < 10; i++) {
new Thread(()->{
LazyMan.getInstance();
}).start();
}
}
}
為什麼要加入volatile關鍵字：因為 「lazyMan = new LazyMan」 不是原子性操作，底層會經過三步操作：
- 為物件分配記憶體空間。
- 執行構造方法初始化物件。
- 將這個物件指向分配的記憶體空間。
由於編譯器為了效能原因，可能會將第二步和第三步操作交換順序，也就是指令重排。順序就變成這樣了：
- 為物件分配記憶體空間。
- 將這個物件指向分配的記憶體空間。
- 執行構造方法初始化物件。
如果A執行緒執行到第二步操作的時候，此時進來一個B執行緒，B執行緒進入if判斷的時候檢查該物件不是null值，直接return返回去，而此時的物件還未完成初始化。所以B執行緒就會存取到一個未初始化的物件。加入volatile解決問題。
3.登記式/靜態內部類（推薦）
這個時候就炫技，我靜態內部類玩得挺6，我用靜態內部類來實現單例模式。
優點：避免了執行緒不安全，延遲載入，效率高。
這種方式跟餓漢式方式採用的機制類似，但又有不同。兩者都是採用了類裝載的機制來保證初始化範例時只有一個執行緒。不同的地方在餓漢式方式是隻要Singleton類被裝載就會範例化，沒有Lazy-Loading的作用，而靜態內部類方式在Singleton類被裝載時並不會立即範例化，而是在需要範例化時，呼叫getInstance方法，才會裝載SingletonInstance類，從而完成Singleton的範例化。
類的靜態屬性只會在第一次載入類的時候初始化，所以在這裡，JVM幫助我們保證了執行緒的安全性，在類進行初始化時，別的執行緒是無法進入的。
public class Holder {
private Holder(){}
//呼叫getInstance()方法的時候才會範例化
public static Holder getInstance(){
return InnerClass.INSTANCE;
}
//靜態內部類去範例化
public static class InnerClass{
private static final Holder INSTANCE = new Holder();
}
}
但是這幾種單例模式都是不安全的，因為我們之前學過一種非常牛逼的技術叫「反射」，使用反射可用破解這幾種單例模式。
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
/**
* 反射破解懶漢式單例模式
*/
public class Singleton {
private volatile static Singleton instance;
private Singleton(){}
public static Singleton getInstance(){
if(instance == null){
synchronized (Singleton.class){
if(instance == null){
instance = new Singleton();
}
}
}
return instance;
}
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
//正常方式
Singleton instance1 = Singleton.getInstance();
//反射破壞
Constructor<Singleton> constructor = Singleton.class.getDeclaredConstructor(null);//獲取指定形參的構造器
constructor.setAccessible(true);//關閉安全檢測，也就是忽略private
Singleton instance2 = constructor.newInstance();//反射建立範例
System.out.println(instance1);
System.out.println(instance2);
}
}
com.hnguigu.single.Singleton@74a14482
com.hnguigu.single.Singleton@1540e19d
Process finished with exit code 0
我們發現確實通過反射破壞了單例模式，怎麼解決呢？因為反射是走的無參構造器，於是我們可用在無參構造器加鎖判斷。
public class Singleton {
private volatile static Singleton instance;
private Singleton(){
synchronized (Singleton.class){
if(instance!=null){
throw new RuntimeException("不要試圖使用反射破壞異常");
}
}
}
public static Singleton getInstance(){
if(instance == null){
synchronized (Singleton.class){
if(instance == null){
instance = new Singleton();
}
}
}
return instance;
}
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
//正常方式
Singleton instance1 = Singleton.getInstance();
//反射破壞
Constructor<Singleton> constructor = Singleton.class.getDeclaredConstructor(null);//獲取指定形參的構造器
constructor.setAccessible(true);//關閉安全檢測，也就是忽略private
Singleton instance2 = constructor.newInstance();//反射建立範例
System.out.println(instance1);
System.out.println(instance2);
}
}
確實報異常了，剛剛是有一個物件是以正常方式建立的，那如果我兩個物件都是反射方式建立的呢？
public class Singleton {
private volatile static Singleton instance;
private Singleton(){
synchronized (Singleton.class){
if(instance!=null){
throw new RuntimeException("不要試圖使用反射破壞異常");
}
}
}
public static Singleton getInstance(){
if(instance == null){
synchronized (Singleton.class){
if(instance == null){
instance = new Singleton();
}
}
}
return instance;
}
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
//反射破壞
Constructor<Singleton> constructor = Singleton.class.getDeclaredConstructor(null);//獲取指定形參的構造器
constructor.setAccessible(true);//關閉安全檢測，也就是忽略private
Singleton instance1 = constructor.newInstance();//反射建立範例
Singleton instance2 = constructor.newInstance();//反射建立範例
System.out.println(instance1);
System.out.println(instance2);
}
}
com.hnguigu.single.Singleton@74a14482
com.hnguigu.single.Singleton@1540e19d
Process finished with exit code 0
還是遭到破壞了，我們可用通過紅綠燈來防止破壞，設定一個標誌位來判斷
public class Singleton {
private volatile static Singleton instance;
//標誌位
private static boolean flag = false;
private Singleton(){
synchronized (Singleton.class){
if(flag == false){
flag = true;
}else{
throw new RuntimeException("不要試圖使用反射破壞異常");
}
}
}
public static Singleton getInstance(){
if(instance == null){
synchronized (Singleton.class){
if(instance == null){
instance = new Singleton();
}
}
}
return instance;
}
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
//正常方式
//Singleton instance1 = Singleton.getInstance();
//反射破壞
Constructor<Singleton> constructor = Singleton.class.getDeclaredConstructor(null);//獲取指定形參的構造器
constructor.setAccessible(true);//關閉安全檢測，也就是忽略private
Singleton instance2 = constructor.newInstance();//反射建立範例
Singleton instance1 = constructor.newInstance();//反射建立範例
System.out.println(instance1);
System.out.println(instance2);
}
}
如果不通過反編譯的情況下，是找不到這個標誌位的，我們還可以對標誌位進行加密處理，使其單例變得更安全。
但是再牛逼的加密方式，也會有解密辦法，假設我們拿到了標誌位，是可以通過反射來轉換標誌位的。
但是 「道高一尺魔高一丈」，正義終將戰勝邪惡！我們檢視反射建立範例原始碼
有一個 「Cannot reflectively create enum objects」 異常，意思就是無法以反射方式建立列舉物件，條件是必須是列舉型別。
4、列舉（推薦）
藉助JDK1.5中新增的列舉來實現單例模式。它不僅能避免多執行緒同步問題，而且還自動支援序列化機制，防止反序列化重新建立新的物件，絕對防止多次範例化。
public enum Singleton {
INSTANCE;
public void whateverMethod() {
}
}
我們驗證列舉類物件不能被反射建立，檢視列舉類裡的構造器
public enum EnumSingle {
INSTANCE;
public EnumSingle getInstance(){
return INSTANCE;
}
}
class Test{
public static void main(String[] args) throws NoSuchMethodException, InvocationTargetException, InstantiationException, IllegalAccessException {
EnumSingle instance1 = EnumSingle.INSTANCE;
Constructor<EnumSingle> constructor = EnumSingle.class.getDeclaredConstructor(String.class,int.class);
EnumSingle instance2 = constructor.newInstance();
System.out.println(instance1);
System.out.println(instance2);
}
}
Exception in thread "main" java.lang.IllegalArgumentException: Cannot reflectively create enum objects
at java.lang.reflect.Constructor.newInstance(Constructor.java:417)
at com.hnguigu.single.Test.main(EnumSingle.java:25)
Process finished with exit code 1
十九、深入理解CAS
CAS全稱叫「Compare-And-Swap」，直譯就是 「比較並交換」。它是一條CPU的原子指令（並行原語），其作用就是讓CPU先比較兩個值是否相等，如果相等則更新為新值，這個過程是原子的。
CAS並行原語體現在JAVA語言中就是sun.misc.Unsafe類中的各個方法。呼叫UnSafe類中的CAS方法，JVM會幫我們實現出CAS組合指令。這是一種完全依賴於硬體的功能。
1、使用
CAS操作是原子性的，所以多執行緒並行使用CAS更新資料時，可以不使用鎖。JDK中大量使用了CAS來更新資料而防止加鎖(synchronized 重量級鎖)來保持原子更新。
我們來看下AtomicInteger類的核心原始碼：
public class AtomicInteger extends Number implements java.io.Serializable {
private static final long serialVersionUID = 6214790243416807050L;
// 設定為使用Unsafe.compareAndSwapInt進行更新
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final long valueOffset;
/*
Unsafe類負責執行CAS並行原語，由JVM轉換為組合。
Java無法操作記憶體，C++可以操作記憶體，Java可以呼叫C++（native方法）。
Unsafe類就相當於Java的後門，可以通過這個類來操作記憶體
*/
static {
try {
valueOffset = unsafe.objectFieldOffset
(AtomicInteger.class.getDeclaredField("value"));
} catch (Exception ex) { throw new Error(ex); }
}
private volatile int value;
public AtomicInteger(int initialValue) {
value = initialValue;
}
}
public class CASDemo {
public static void main(String[] args) {
AtomicInteger atomicInteger = new AtomicInteger(2020);
//boolean compareAndSet(int expect, int update)
//expect：期望，update：更新
//如果實際的值和我期望的值是相同的，那麼就更新，否則就不更新
System.out.println(atomicInteger.compareAndSet(2020, 2021));//true
System.out.println(atomicInteger.get());//2021
System.out.println(atomicInteger.compareAndSet(1999, 2021));//false
System.out.println(atomicInteger.get());//2021
atomicInteger.getAndIncrement();//++操作
}
}
以getAndIncrement()的原始碼為例：
public final int getAndIncrement() {
return unsafe.getAndAddInt(this, valueOffset, 1);
}
//引數：var1當前物件，var2該變數值在記憶體中的偏移地址，var4需要增加的值大小
public final int getAndAddInt(Object var1, long var2, int var4) {
int var5;//代表要交換的值。
do {
//getIntVolatile獲取記憶體地址中的值
var5 = this.getIntVolatile(var1, var2);
} while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));
return var5;
}
compareAndSwapInt該方法中使用了自旋鎖以保證其原子性。
假設主記憶體值為 v 等於10，此時有 T1、T2兩個執行緒進入到該方法，根據 Java 記憶體模型（JMM）我們可以知道，執行緒 T1 和執行緒 T2 都會將主記憶體的值10拷貝到自己的工作記憶體。
1、當執行緒 T1 和執行緒 T2 都通過getIntVolatile(var1, var2)賦值給了變數 var5 之後，執行緒 T1 被掛起；
2、執行緒 T2 呼叫方法compareAndSwapInt，因為當中的期望值 var5 和當前主記憶體值相同，比較成功，更新當前記憶體的值為 11，返回 true，退出迴圈；
3、執行緒 T1 被喚醒，在執行compareAndSwapInt方法的時候，由於當前記憶體的值已經為11，和 工作記憶體 var5 的值10不同了，所以比較不成功，返回 false，繼續執行迴圈；
4、執行緒 T1 重新從主記憶體獲取當前的最新值11賦值給 var5；
5、執行緒 T1 繼續進行比較，若此時沒有其他執行緒對主記憶體的進行修改，比較更新成功 ，退出迴圈；否則繼續執行步驟4。
雖然CAS沒有加鎖保證了一致性，並行性有所提高 ，但是也產生了一系列的問題，比如迴圈時間長開銷大、只能保證一個共用變數的原子操作、會產生ABA問題。
2、ABA問題
使用 CAS 會產生 ABA 問題，這是因為 CAS 演演算法是在某一時刻取出記憶體值然後在當前的時刻進行比較，中間存在一個時間差，在這個時間差裡就可能會產生 ABA 問題。
ABA問題的過程就是：現有 T1 T2 執行緒從記憶體中獲取值為 A，T2將值改成了B，然後又改回成A，T2退出。T1進行操作，使用預期值和記憶體中的值比較，發現都是A，修改成功然後退出。CAS進行檢查時則會發現它的值沒有發生變化，但是實際上卻變化了。我們稱這種現象為ABA問題。
解決思路：可以新增版本號，每次值更新的時候將版本號+1。
對應的思想就是樂觀鎖。
3、原子參照
帶版本號的原子操作。
從Java 1.5開始，java.util.concurrent.atomic包裡提供了一個類AtomicStampedReference。其中的compareAndSet方法如下
public boolean compareAndSet(V expectedReference,//期望被修改的值
V newReference,//新的值
int expectedStamp,//期望被修改的版本號
int newStamp) {//新的版本號
Pair<V> current = pair;
return
expectedReference == current.reference &&
expectedStamp == current.stamp &&
((newReference == current.reference &&
newStamp == current.stamp) ||
casPair(current, Pair.of(newReference, newStamp)));
}
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicStampedReference;
/**
* ABA問題
*/
public class ABADemo {
/*
* AtomicStampedReference：注意如果泛型是基本資料型別的包裝類，注意物件的參照問題
* 正常業務操作下，這裡面比較的是一個個物件
*/
private static AtomicStampedReference<Integer> stampedReference = new AtomicStampedReference<>(100, 1);
public static void main(String[] args) {
new Thread(() -> {
int stamp = stampedReference.getStamp();//版本號
System.out.println("當前執行緒的名字：" + Thread.currentThread().getName() + "，版本號：" + stamp + "，值：" + stampedReference.getReference());
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
stampedReference.compareAndSet(100, 101, stampedReference.getStamp(), stampedReference.getStamp() + 1);
System.out.println("當前執行緒的名字：" + Thread.currentThread().getName() + "，版本號：" + stampedReference.getStamp() + "，值：" + stampedReference.getReference());
stampedReference.compareAndSet(101, 100, stampedReference.getStamp(), stampedReference.getStamp() + 1);
System.out.println("當前執行緒的名字：" + Thread.currentThread().getName() + "，版本號：" + stampedReference.getStamp() + "，值：" + stampedReference.getReference());
System.out.println("T1完成CAS操作~");
}, "T1").start();
new Thread(() -> {
int stamp = stampedReference.getStamp();//版本號
System.out.println("當前執行緒的名字：" + Thread.currentThread().getName() + "，版本號：" + stamp + "，值：" + stampedReference.getReference());
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
//版本號對不上，修改不成功！
boolean flag = stampedReference.compareAndSet(100, 66, stamp, stamp + 1);//false
System.out.println("當前執行緒的名字：" + Thread.currentThread().getName() + "，修改成功與否：" + flag + "，最新版本號：" + stampedReference.getStamp() + "，最新值：" + stampedReference.getReference());
}, "T2").start();
}
}
當前執行緒的名字：T1，版本號：1，值：100
當前執行緒的名字：T2，版本號：1，值：100
當前執行緒的名字：T1，版本號：2，值：101
當前執行緒的名字：T1，版本號：3，值：100
T1完成CAS操作~
當前執行緒的名字：T2，修改成功與否：false，最新版本號：3，最新值：100
Process finished with exit code 0
二十、徹底吃透各種鎖
1、公平鎖和非公平鎖
這裡前面講過就不再贅述。
2、可重入鎖
可重入鎖：某個執行緒已經獲得某個鎖，還可以再次獲得鎖而不會出現死鎖。
/**
* 可重入鎖的案例，synchronized
*/
public class WhatReentrantLock {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(()->{
phone.sendSms();
},"A").start();
new Thread(()->{
phone.sendSms();
},"B").start();
/*
A=>sms
A=>call
B=>sms
B=>call
*/
}
}
class Phone{
public synchronized void sendSms(){
System.out.println(Thread.currentThread().getName()+"=>"+"sms");
call();
}
public synchronized void call(){
System.out.println(Thread.currentThread().getName()+"=>"+"call");
}
}
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
/**
* 可重入鎖，ReentrantLock
*/
public class WhatReentrantLock2 {
public static void main(String[] args) {
Phone phone = new Phone();
new Thread(()->{
phone.sendSms();
},"A").start();
new Thread(()->{
phone.sendSms();
},"B").start();
/*
A=>sms
A=>call
B=>sms
B=>call
*/
}
}
class Phone2{
Lock lock = new ReentrantLock();
public void sendSms(){
//這是兩把鎖，兩把鑰匙
//lock鎖必須配對，否則就會死鎖
lock.lock();
try {
System.out.println(Thread.currentThread().getName()+"=>"+"sms");
call();//這裡也是一把鎖
} catch (Exception e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
public void call(){
lock.lock();
try {
System.out.println(Thread.currentThread().getName()+"=>"+"call");
} catch (Exception e) {
e.printStackTrace();
}finally {
lock.unlock();
}
}
}
lock鎖必須配對，相當於lock和 unlock 必須數量相同；
3、自旋鎖
由於多執行緒的核心是CPU的時間分片，所以同一時刻只能有一個執行緒獲取到鎖。那麼沒有獲取到鎖的執行緒應該怎麼辦？
通常有兩種處理方式：一種是沒有獲取到鎖的執行緒就一直迴圈等待判斷該資源是否已經釋放鎖，這種鎖叫做自旋鎖，它不用將執行緒阻塞起來(NON-BLOCKING)；還有一種處理方式就是把自己阻塞起來，等待重新排程請求，這種叫做互斥鎖。
自旋鎖：當一個執行緒嘗試去獲取某一把鎖的時候，如果這個鎖此時已經被別人獲取(佔用)，該執行緒將會等待，間隔一段時間後會再次嘗試獲取。這種採用迴圈加鎖 -> 等待的機制被稱為自旋鎖(spinlock)。
- 優點：如果持有鎖的執行緒能在短時間內釋放鎖資源，那麼避免了使用者程序和核心切換的消耗。
- 缺點：如果持有鎖的執行緒長時間佔用鎖執行同步塊，其他執行緒就一直佔著CPU資源不進行釋放，進而會影響整體系統的效能。
解決辦法：可以給自旋鎖設定一個自旋時間，等時間一到立即釋放自旋鎖。
自己建立一個自旋鎖
import java.util.concurrent.atomic.AtomicReference;
/**
* 自旋鎖
*/
public class SpinLock {
private AtomicReference<Thread> atomicReference = new AtomicReference<>();
//加鎖
public void myLock(){
Thread thread = Thread.currentThread();
//自旋鎖
while(!atomicReference.compareAndSet(null,thread)){}
System.out.println(thread.getName()+"=>lock");
}
//解鎖
public void myUnLock(){
Thread thread = Thread.currentThread();
if(!atomicReference.compareAndSet(thread,null)){
throw new RuntimeException("釋放鎖失敗！");
}
System.out.println(thread.getName()+"=>unlock");
}
}
import java.util.concurrent.TimeUnit;
/**
* 測試自旋鎖
*/
public class TestSpinLock {
public static void main(String[] args) throws InterruptedException {
//ReentrantLock reentrantLock = new ReentrantLock();
//reentrantLock.lock();
//reentrantLock.unlock();
//自己寫的鎖，底層使用自旋鎖，cas實現
SpinLock lock = new SpinLock();
new Thread(()->{
lock.myLock();
try {
TimeUnit.SECONDS.sleep(3);
} catch (Exception e) {
e.printStackTrace();
}finally {
lock.myUnLock();
}
},"T1").start();
TimeUnit.SECONDS.sleep(1);//讓t1先獲得鎖
new Thread(()->{
lock.myLock();
try {
TimeUnit.SECONDS.sleep(1);
} catch (Exception e) {
e.printStackTrace();
}finally {
lock.myUnLock();
}
},"T2").start();
}
}
T1=>lock
T1=>unlock
T2=>lock
T2=>unlock
//t1進來獲得鎖之後，t2進入自旋，等待t1釋放鎖後，t2退出自旋然後獲得鎖
4、死鎖
多個執行緒各自佔有一些共用資源，並且互相等待其他執行緒佔有的資源才能執行，而導致兩個或多個執行緒在等待對方釋放鎖資源，都停止執行的情形，某一個程式碼塊同時擁有兩個以上物件的鎖時，就可能會發生「死鎖」的問題。
import java.util.concurrent.TimeUnit;
public class DeadLockDemo {
public static void main(String[] args) {
Makeup makeup1 = new Makeup(0, "灰姑娘");
Makeup makeup2 = new Makeup(1, "白雪公主");
makeup1.start();
makeup2.start();
//最終結果：程式僵持執行著
}
}
//口紅
class Lipstick{
String name = "迪奧口紅";
}
//鏡子
class Mirror{
String name = "魔鏡";
}
//化妝
class Makeup extends Thread{
//使用static保證只有一份
static Lipstick lipstick = new Lipstick();
static Mirror mirror = new Mirror();
int choice;//選擇
String girlName;//選擇化妝的人
Makeup(int choice,String girlName){
this.choice = choice;
this.girlName = girlName;
}
@Override
public void run() {
try {
makeup();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//化妝
private void makeup() throws InterruptedException {
if(choice==0){
synchronized (lipstick){//獲得口紅的鎖
System.out.println(this.girlName + "--->獲得" + lipstick.name);
TimeUnit.SECONDS.sleep(1);
synchronized (mirror){//一秒鐘之後想要鏡子的鎖
System.out.println(this.girlName + "--->獲得" + mirror.name);
}
}
}else{
synchronized (mirror){//獲得鏡子的鎖
System.out.println(this.girlName + "--->獲得" + mirror.name);
TimeUnit.SECONDS.sleep(2);
synchronized (lipstick){//兩秒鐘之後想要口紅的鎖
System.out.println(this.girlName + "--->獲得" + lipstick.name);
}
}
}
}
}
灰姑娘拿著口紅的鎖不釋放，隨後一秒鐘後又要魔鏡的鎖，白雪公主拿著魔鏡的鎖不釋放，兩秒鐘後又要口紅的鎖，雙方都不釋放已經使用完了的鎖資源，僵持形成死鎖。
產生死鎖的四個必要條件：
- 互斥條件：一個資源每次只能被一個程序使用。
- 請求與保持條件：一個程序因請求資源而阻塞時，對以獲得的資源保持不放。
- 不剝奪條件：程序已獲得的資源，在未使用完畢之前，不能被強行搶走。
- 迴圈等待條件：若干程序之間形成一種頭尾相接的迴圈等待資源關係。
上面就是形成死鎖的必要條件，只需要解決其中任意一個或者多個條件就可以避免死鎖的發生。
死鎖排查
jps程序狀態工具 jps.exe 工具是 jdk 自帶的，在 %JAVA_HOME%/bin 目錄下。
第一步：開啟idea提供terminal終端命令列，使用jps -l檢視程序
第二步：使用jstack 程序號檢視堆疊資訊
一般情況資訊在最後面
好文要頂！