java多執行緒總結（系列一）
阿新 • • 發佈：2019-07-11
1、Java中的Thread建立和狀態
(1)新建（New）
建立後尚未啟動。ex：Thread t = new Thread(){//重寫run方法};
(2)可執行（Runnable）
可能正在執行，也可能正在等待 CPU 時間片。包含了作業系統執行緒狀態中的 Running 和 Ready。
(3)阻塞（Blocked）
等待獲取一個排它鎖，如果其執行緒釋放了鎖就會結束此狀態。
(4)無限期等待（Waiting）
等待其它執行緒顯式地喚醒，否則不會被分配 CPU 時間片。
| 進入方法 | 退出方法 |
|---|---|
| 沒有設定 Timeout 引數的 Object.wait() 方法 | Object.notify() / Object.notifyAll() |
| 沒有設定 Timeout 引數的 Thread.join() 方法 | 被呼叫的執行緒執行完畢 |
| LockSupport.park() 方法 | LockSupport.unpark(Thread) |
(5)限期等待（Timed Waiting）
無需等待其它執行緒顯式地喚醒，在一定時間之後會被系統自動喚醒。
呼叫 Thread.sleep() 方法使執行緒進入限期等待狀態時，常常用“使一個執行緒睡眠”進行描述。
呼叫 Object.wait() 方法使執行緒進入限期等待或者無限期等待時，常常用“掛起一個執行緒”進行描述。
睡眠和掛起是用來描述行為，而阻塞和等待用來描述狀態。
阻塞和等待的區別在於，阻塞是被動的，它是在等待獲取一個排它鎖。而等待是主動的，通過呼叫 Thread.sleep() 和 Object.wait() 等方法進入。
| 進入方法 | 退出方法 |
|---|---|
| Thread.sleep() 方法 | 時間結束 |
| 設定了 Timeout 引數的 Object.wait() 方法 | 時間結束 / Object.notify() / Object.notifyAll() |
| 設定了 Timeout 引數的 Thread.join() 方法 | 時間結束 / 被呼叫的執行緒執行完畢 |
| LockSupport.parkNanos() 方法 | LockSupport.unpark(Thread) |
| LockSupport.parkUntil() 方法 | LockSupport.unpark(Thread) |
(6)死亡（Terminated）
可以是執行緒結束任務之後自己結束，或者產生了異常而結束。
下面是Thread類的start方法原始碼(模板設計模式的應用)
public synchronized void start() {
/**
* threadStatus==0表示執行緒還沒有start，所以不能兩次啟動thread，否則丟擲異常IllegalThreadStateException
*/
if (threadStatus != 0)
throw new IllegalThreadStateException();
/* 將執行緒新增到所屬的執行緒組中，並減少執行緒組中unstart的執行緒數 */
group.add(this);
boolean started = false;
try {
start0();//呼叫本地方法，本地方法中呼叫run方法
started = true;
} finally {
try {
if (!started) {
group.threadStartFailed(this);
}
} catch (Throwable ignore) {
/* */
}
}2、Runnable介面
重寫Thread類的run方法和實現Runnable介面的run方法的不同：
①繼承Thread類之後就不能繼承別的類，而實現Runnable介面更靈活
②Thread類的run方法是不能共享的，執行緒A不能把執行緒B的run方法當做自己的執行單元，而使用Runnable介面能夠實現這一點，即使用同一個Runnable的例項構造不同的Thread例項
3、守護執行緒
守護執行緒是程式執行時在後臺提供服務的執行緒，不屬於程式中不可或缺的部分。當程式中沒有一個非守護執行緒的時候，JVM也會退出，即當所有非守護執行緒結束時，程式也就終止，同時會殺死所有守護執行緒。main() 屬於非守護執行緒。
線上程啟動之前使用 setDaemon() 方法可以將一個執行緒設定為守護執行緒。如果一個執行緒已經start，再呼叫setDaemon方法就會丟擲異常IllegalThreadStateException
public class TestThread4 {
public static void main(String[] args) {
//thread t作為當前處理業務的執行緒，其中建立子執行緒用於檢測某些事項
Thread t = new Thread() {
@Override
public void run() {
Thread innerThread = new Thread() {
@Override
public void run() {
try {
while (true) {
System.out.println("我是檢測XX的執行緒");
TimeUnit.SECONDS.sleep(1);//假設每隔 1s check
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
innerThread.setDaemon(true);//設定為守護執行緒
innerThread.start();
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
};
t.start();
//父執行緒t 執行完退出之後，他的子執行緒innerThread也會退出，
//如果innerThread沒有設定為守護執行緒，那麼這個Application就會一直執行不會退出
//設定為守護執行緒，那麼這個Application完成任務或者被異常打斷執行，這個子執行緒也會退出而不會繼續執行check
//t.setDaemon(true);//Exception in thread "main" java.lang.IllegalThreadStateException
}
}4、Join方法
(1)join方法使用
線上程中呼叫另一個執行緒的 join() 方法，會將當前執行緒掛起，而不是忙等待，直到呼叫join方法的執行緒結束。對於以下程式碼，Work實現Runnable介面，並在構造方法傳遞呼叫join的執行緒，所以儘管main中start方法的呼叫是t3、t2、t1但是實際上的執行順序還是t1先執行，然後t2，最後t3（t2會等待t1執行結束，t3會等待t2執行結束）
package demo1;
public class TestThread5 {
public static void main(String[] args) {
Thread t1 = new Thread(new Work(null));
Thread t2 = new Thread(new Work(t1));
Thread t3 = new Thread(new Work(t2));
t3.start();
t2.start();
t1.start();
}
}
class Work implements Runnable {
private Thread prevThread;
public Work(Thread prevThread) {
this.prevThread = prevThread;
}
public void run() {
if(null != prevThread) {
try {
prevThread.join();
for (int i = 0; i < 50; i++) {
System.out.println(Thread.currentThread().getName() + "-" + i);
}
} catch (InterruptedException e) {
e.printStackTrace();
}
} else {
System.out.println(Thread.currentThread().getName());
}
}
}(2)join方法原理
下面是join方法的原始碼
//預設的join方法
public final void join() throws InterruptedException {
join(0);
}
public final synchronized void join(long millis)
throws InterruptedException {
long base = System.currentTimeMillis();
long now = 0;
if (millis < 0) {
throw new IllegalArgumentException("timeout value is negative");
}
if (millis == 0) { //預設直接使用join方法的時候，會進入這個分支
while (isAlive()) {
wait(0); //注意這裡呼叫的是wait方法，我們就需要知道wait方法的作用（呼叫此方法的當前執行緒需要釋放鎖，並等待喚醒）
}
} else { //設定join方法的引數(即呼叫者執行緒等待多久，目標執行緒還沒有結束的話就不等待了)
while (isAlive()) {
long delay = millis - now;
if (delay <= 0) {
break; //結束等待
}
wait(delay); //設定的時間未到之前，繼續等待
now = System.currentTimeMillis() - base;
}
}
} 在上面的方法中，我們看到public final synchronized void join(long millis){}這是一個synchronized方法，瞭解過synchronized的知道，修飾方法的代表鎖住物件例項(this)，那麼這個this是什麼呢，（呼叫者執行緒CallThread呼叫thread.join()方法，this就是當前的執行緒物件）。
CallThread執行緒呼叫join方法的時候，先持有當前的鎖(執行緒物件thread)，然後在這裡這裡wait住了（釋放鎖，並進入阻塞等待狀態），那麼CallThread什麼時候被notify呢，下面是jvm的部分原始碼，可以看到notify的位置。當thread執行完畢，jvm就是喚醒阻塞在thread物件上的執行緒（即剛剛說的CallThread呼叫者執行緒），那麼這個呼叫者執行緒就能繼續執行下去了。
總結就是，假設當前呼叫者執行緒是main執行緒，那麼：
①main執行緒在呼叫方法之後，會首先獲取鎖——thread物件例項；②進入其中一個分支之後，會呼叫wait方法，釋放鎖，並阻塞等待thread執行完之後被喚醒；③當thread執行完畢，會呼叫執行緒的exit方法，在下面的程式碼中我們能看出其中呼叫ensure_join方法，而在ensure_join中就會喚醒阻塞等待的thread物件上的執行緒；④main執行緒被notify，然後繼續執行
// 位於/hotspot/src/share/vm/runtime/thread.cpp中
void JavaThread::exit(bool destroy_vm, ExitType exit_type) {
// ...
// Notify waiters on thread object. This has to be done after exit() is called
// on the thread (if the thread is the last thread in a daemon ThreadGroup the
// group should have the destroyed bit set before waiters are notified).
// join（ensure_join）
ensure_join(this);
// ...
}
static void ensure_join(JavaThread* thread) {
// We do not need to grap the Threads_lock, since we are operating on ourself.
Handle threadObj(thread, thread->threadObj());
assert(threadObj.not_null(), "java thread object must exist");
ObjectLocker lock(threadObj, thread);
// Ignore pending exception (ThreadDeath), since we are exiting anyway
thread->clear_pending_exception();
// Thread is exiting. So set thread_status field in java.lang.Thread class to TERMINATED.
java_lang_Thread::set_thread_status(threadObj(), java_lang_Thread::TERMINATED);
// Clear the native thread instance - this makes isAlive return false and allows the join()
// to complete once we've done the notify_all below
java_lang_Thread::set_thread(threadObj(), NULL);
// thread就是當前執行緒，就是剛才例子中說的t執行緒
lock.notify_all(thread);
// Ignore pending exception (ThreadDeath), since we are exiting anyway
thread->clear_pending_exception();
}5、Stop方法
(1)stop弊端
破壞原子邏輯，如下面的例子中，MultiThread實現了Runnable介面,run方法中加入了synchronized同步程式碼塊,表示內部是原子邏輯,a的值會先增加後減少,按照synchronized的規則,無論啟動多少個執行緒,打印出來的結果都應該是a=0。但是如果有一個正在執行的執行緒被stop,就會破壞這種原子邏輯.(下面面main方法中程式碼)
public class TestStop4 {
public static void main(String[] args) {
MultiThread t = new MultiThread();
Thread testThread = new Thread(t,"testThread");
// 啟動t1執行緒
testThread.start();
//（1）這裡是保證執行緒t1先線獲得鎖，執行緒t1啟動,並執行run方法,由於沒有其他執行緒同步程式碼塊的鎖,
//所以t1執行緒執行自加後執行到sleep方法開始休眠,此時a=1.
try {
TimeUnit.MILLISECONDS.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
//（2）JVM又啟動了5個執行緒,也同時執行run方法,由於synchronized關鍵字的阻塞作用,這5個執行緒不能執行自增和自減操作,等待t1執行緒釋放執行緒鎖.
for (int i = 0; i < 5; i++) {
new Thread(t).start();
}
//（3）主執行緒執行了t1.stop方法,終止了t1執行緒,但是由於a變數是執行緒共享的,所以其他5個執行緒獲得的a變數也是1.（synchronized的可見性保證）
testThread.stop();
//（4）其他5個執行緒獲得CPU的執行機會,打印出a的值.
}
}
class MultiThread implements Runnable {
int a = 0;
private Object object = new Object();
public void run() {
// 同步程式碼塊，保證原子操作
synchronized (object) {
// 自增
a++;
try {
// 執行緒休眠0.1秒
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 自減
a--;
String tn = Thread.currentThread().getName();
System.out.println(tn + ":a =" + a);
}
}
}(2)實現“stop”執行緒
下面是兩種優雅的關閉執行緒的方式
①使用自定義標誌位配合volatile實現
public class ThreadStop1 {
private static class Work extends Thread {
private volatile boolean flag = true;
public void run() {
while (flag) {
//do something
}
}
public void shutdown() {
this.flag = false;
}
}
public static void main(String[] args) {
Work w = new Work();
w.start();
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
w.shutdown();
}
}②使用中斷
public class ThreadStop2 {
private static class Work extends Thread{
public void run() {
while (true) {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
break; //別的執行緒中斷該執行緒
}
}
}
}
public static void main(String[] args) {
Work w = new Work();
w.start();
try {
TimeUnit.SECONDS.sleep(1);
w.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
6、synchronized
可以參考我總結的Java中Lock和synchronized。這裡在說一下synchronized的相關知識
(1)synchronized用法
①當synchronized作用在例項方法時，監視器鎖（monitor）便是物件例項（this）；
②當synchronized作用在靜態方法時，監視器鎖（monitor）便是物件的Class例項，因為Class資料存在於永久代，因此靜態方法鎖相當於該類的一個全域性鎖；
③當synchronized作用在某一個物件例項時，監視器鎖（monitor）便是括號括起來的物件例項；
public class TestSynchronized1 {
//靜態程式碼塊
static {
synchronized (TestSynchronized1.class) {
System.out.println(Thread.currentThread().getName() + "==>static code");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
//static
private synchronized static void m1() {
System.out.println(Thread.currentThread().getName() + "==>method:m1");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//this
private synchronized void m2() {
System.out.println(Thread.currentThread().getName() + "==>method:m2");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//普通方法
private void m3() {
System.out.println(Thread.currentThread().getName() + "==>method:m3");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//static
private synchronized static void m4() {
System.out.println(Thread.currentThread().getName() + "==>method:m4");
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
final TestSynchronized1 test = new TestSynchronized1();
new Thread() {
@Override
public void run() {
test.m2();
}
}.start();
new Thread() {
@Override
public void run() {
TestSynchronized1.m1();
}
}.start();
new Thread() {
@Override
public void run() {
test.m3();}
}.start();
new Thread() {
@Override
public void run() {
TestSynchronized1.m4();
}
}.start();
}
}
(2)synchronized的原理
(關於鎖與原子交換指令硬體原語可以參考從同步原語看非阻塞同步 的介紹)
①Java 虛擬機器中的同步(Synchronization)基於進入和退出管程(Monitor)物件實現。同步程式碼塊是使用monitorenter和monitorexit來實現的，同步方法 並不是由 monitorenter 和 monitorexit 指令來實現同步的，而是由方法呼叫指令讀取執行時常量池中方法的 ACC_SYNCHRONIZED 標誌來隱式實現的(但是實際上差不多)。monitorenter指令是在編譯後插入同步程式碼塊的起始位置，而monitorexit指令是在方法結束處和異常處，每個物件都有一個monitor與之關聯，當一個monitor被持有後它就會處於鎖定狀態。
②synchronized用的鎖是存在Java物件頭（非陣列型別包括Mark Word、型別指標，陣列型別多了陣列長度）裡面的，物件頭中的Mark Word儲存物件的hashCode，分代年齡和鎖標記位，型別指標指向物件的元資料資訊，JVM通過這個指標確定該物件是那個類的例項等資訊。
③當在物件上加鎖的時候，資料是記錄在物件頭中，物件頭中的Mark Word裡儲存的資料會隨著鎖標誌位的變化而變化（無鎖、輕量級鎖00、重量級鎖10、偏向鎖01）。當執行synchronized的同步方法或者同步程式碼塊時候會在物件頭中記錄鎖標記，鎖標記指向的是monitor物件（也稱為管程或者監視器鎖）的起始地址。由於每個物件都有一個monitor與之關聯，monitor和與關聯的物件一起建立（當執行緒試圖獲取鎖的時候）或銷燬，當monitor被某個執行緒持有之後，就處於鎖定狀態。
④Hotspot虛擬機器中的實現，通過ObjectMonitor來實現的。
(3)關於synchronized的說明
①synchronized能夠保證原子性、可見性、有序性
②在定義同步程式碼塊的時候，不要將常量物件作為鎖物件
③synchronized鎖的是物件，而不是引用
④當同步方法出現異常的時候會自動釋放鎖，不會影響其他執行緒的執行
⑤synchronized是可重入的
(2)實現一個會發生死鎖的程式
public class TestDeadLock {
private final Object lock1 = new Object();
private final Object lock2 = new Object();
private void m1() {
synchronized (lock1) {
System.out.println(Thread.currentThread().getName() + " get lock:lock1");
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (lock2) {
System.out.println(Thread.currentThread().getName() + " get lock:lock2");
}
}
}
private void m2() {
synchronized (lock2) {
System.out.println(Thread.currentThread().getName() + " get lock:lock2");
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
synchronized (lock1) {
System.out.println(Thread.currentThread().getName() + " get lock:lock1");
}
}
}
public static void main(String[] args) {
final TestDeadLock test = new TestDeadLock();
new Thread(){
@Override
public void run() {
test.m1();
}
}.start();
new Thread(){
@Override
public void run() {
test.m2();
}
}.start();
}
}
7、執行緒間通訊與生產者消費者模型
(1)wait/notify的訊息通知機制
結合上面說到的synchronized，在上圖中，ObjectMonitor中有兩個佇列（EntryList、WaitSet(Wait Set是一個虛擬的概念)）以及鎖持有者Owner標記，其中WaitSet是哪些呼叫wait方法之後被阻塞等待的執行緒佇列，EntryList是ContentionList中能有資格獲取鎖的執行緒佇列。當多個執行緒併發訪問同一個同步程式碼時候，首先會進入EntryList，當執行緒獲得鎖之後monitor中的Owner標記會記錄此執行緒，並在該monitor中的計數器執行遞增計算代表當前鎖被持有鎖定，而沒有獲取到的執行緒繼續在EntryList中阻塞等待。如果執行緒呼叫了wait方法，則monitor中的計數器執行賦0運算，並且將Owner標記賦值為null，代表當前沒有執行緒持有鎖，同時呼叫wait方法的執行緒進入WaitSet佇列中阻塞等待，直到持有鎖的執行執行緒呼叫notify/notifyAll方法喚醒WaitSet中的執行緒，喚醒的執行緒進入EntryList中等待鎖的獲取。除了使用wait方法可以將修改monitor的狀態之外，顯然持有鎖的執行緒的同步程式碼塊執行結束也會釋放鎖標記，monitor中的Owner會被賦值為null，計數器賦值為0。如下圖所示是一個執行緒狀態轉換圖。
①wait
該方法用來將當前執行緒置入BLOCKED狀態（或者進入WAITSET中），直到接到通知或被中斷為止。在呼叫 wait()之前，執行緒必須要獲得該物件的MONITOR，即只能在同步方法或同步塊中呼叫 wait()方法。呼叫wait()方法之後，當前執行緒會釋放鎖。如果呼叫wait()方法時，執行緒並未獲取到鎖的話，則會丟擲IllegalMonitorStateException異常，這是個RuntimeException。被喚醒之後，如果再次獲取到鎖的話，當前執行緒才能從wait()方法處成功返回。
②notify
該方法也要在同步方法或同步塊中呼叫，即在呼叫前，執行緒也必須要獲得該物件的MONITOR，如果呼叫 notify()時沒有持有適當的鎖，也會丟擲 IllegalMonitorStateException。 該方法任意從在該MONITOR上的WAITTING狀態的執行緒中挑選一個進行通知，使得呼叫wait()方法的執行緒從等待佇列移入到同步佇列中，等待有機會再一次獲取到鎖，從而使得呼叫wait()方法的執行緒能夠從wait()方法處退出。呼叫notify後，當前執行緒不會馬上釋放該物件鎖，要等到程式退出同步塊後，當前執行緒才會釋放鎖。
③notifyAll
該方法與 notify ()方法的工作方式相同，重要的一點差異是： notifyAll 使所有原來在該物件的MONITOR上 wait 的執行緒統統退出WAITTING狀態，使得他們全部從等待佇列中移入到同步佇列中去，等待下一次能夠有機會獲取到物件監視器鎖。
(2)waitset概念
(3)notify早期通知
假設現在有兩個執行緒，threadWait、threadNotify，在測試程式中讓使得threadWait還沒開始 wait 的時候，threadNotify已經 notify 了。再這樣的前提下，儘管threadNotify進行了notify操作，但是這個通知是沒有任何響應的（threadWait還沒有進行wait），當 threadNotify退出 synchronized 程式碼塊後，threadWait再開始 wait，便會一直阻塞等待，直到被別的執行緒打斷。比如在下面的示例程式碼中，就模擬出notify早期通知帶來的問題：
public class EarlyNotify {
private static final Object object = new Object();
public static void main(String[] args) {
Thread threadWait = new Thread(new ThreadWait(object),"threadWait");
Thread threadNotify = new Thread(new ThreadNotify(object), "threadNotify");
threadNotify.start(); //喚醒執行緒線啟動
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
//wait執行緒後啟動
threadWait.start();
}
private static class ThreadWait implements Runnable {
private Object lock;
public ThreadWait(Object lock) {
this.lock = lock;
}
public void run() {
synchronized (lock) {
System.out.println(Thread.currentThread().getName() + "進入程式碼塊");
System.out.println(Thread.currentThread().getName() + "進入wait set");
try {
lock.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(Thread.currentThread().getName() + "退出wait set");
}
}
}
private static class ThreadNotify implements Runnable {
private Object lock;
public ThreadNotify(Object lock) {
this.lock = lock;
}
public void run() {
synchronized (lock) {
System.out.println(Thread.currentThread().getName() + "進入程式碼塊");
System.out.println(Thread.currentThread().getName() + "開始notify");
lock.notify();
System.out.println(Thread.currentThread().getName() + "notify結束");
}
}
}
}
執行之後使用jconsole檢視如下
(4)wait的條件發生變化
等待執行緒wait的條件發生變化，如果沒有再次檢查等待的條件就可能發生錯誤。看一下下面的程式例項以及執行結果。
//執行出現下面的結果
/*
* consumerThread1 list is empty
* producerThread begin produce
* consumerThread2 取出元素element0
* consumerThread1 end wait, begin consume
* Exception in thread "consumerThread1" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0
*/
public class WaitThreadUncheckCondition {
private static List<String> lock = new ArrayList<String>();
public static void main(String[] args) {
Thread producerThread = new Thread(new ProducerThread(lock),"producerThread");
Thread consumerThread1 = new Thread(new ConsumerThread(lock), "consumerThread1");
Thread consumerThread2 = new Thread(new ConsumerThread(lock), "consumerThread2");
consumerThread1.start();
consumerThread2.start();
producerThread.start();
}
private static class ProducerThread implements Runnable {
private List<String> lock;
private int i = 0;
public ProducerThread(List<String> lock) {
this.lock = lock;
}
public void run() {
synchronized (lock) {
System.out.println(Thread.currentThread().getName() + " begin produce");
lock.add("element" + i++);
lock.notifyAll();
}
}
}
private static class ConsumerThread implements Runnable {
private List<String> list;
public ConsumerThread(List<String> list) {
this.list = list;
}
public void run() {
synchronized (lock) {
try {
if (lock.isEmpty()) {
System.out.println(Thread.currentThread().getName() + " list is empty");
lock.wait();
System.out.println(Thread.currentThread().getName() + " end wait, begin consume");
}
System.out.println(Thread.currentThread().getName() + " 取出元素" + lock.remove(0));
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
} 在這個例子中一共開啟了3個執行緒，consumerThread1,consumerThread2以及producerThread。首先consumerThread1在獲取到lock之後判斷isEmpty為true，然後呼叫了wait方法進入阻塞等待狀態，並且將物件鎖釋放出來。然後，consumerThread2能夠獲取物件鎖，從而進入到同步代塊中，當執行到wait方法時判斷isEmpty也為true，同樣的也會釋放物件鎖。因此，producerThread能夠獲取到物件鎖，進入到同步程式碼塊中，向list中插入資料後，通過notifyAll方法通知處於WAITING狀態的consumerThread1和consumerThread2執行緒。consumerThread2得到物件鎖後，從wait方法出退出然後繼續向下執行，刪除了一個元素讓List為空，方法執行結束，退出同步塊，釋放掉物件鎖。這個時候consumerThread1獲取到物件鎖後，從wait方法退出，繼續往下執行，這個時候consumerThread1再執行lock.remove(0);就會出錯，因為List由於Consumer1刪除一個元素之後已經為空了。
(5)修改的生產者消費者模型
//多生產者消費者的情況，需要注意如果使用notify方法會導致程式“假死”，即所有的執行緒都wait住了，沒有能夠喚醒的執行緒執行：假設當前多個生產者執行緒會呼叫wait方法阻塞等待，當其中的生產者執行緒獲取到物件鎖之後使用notify通知處於WAITTING狀態的執行緒，如果喚醒的仍然是生產者執行緒，就會造成所有的生產者執行緒都處於等待狀態。這樣消費者得不到消費的訊號，就也是一直處於wait狀態
public class ProducersAndConsumersPattern {
private int sth = 0;
private volatile boolean isProduced = false;
private final Object object = new Object();
private void producer() {
synchronized (object) {
//如果不需要生產者生產
while (isProduced) {
//生產者進入等待狀態
try {
object.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//生產者進行生產
sth++;
System.out.println(Thread.currentThread().getName() + " producer: " + sth);
//喚醒消費者
object.notifyAll();
isProduced = true; //消費者可以進行消費
}
}
private void consumer() {
synchronized (object) {
//消費者不可以消費，進入等待阻塞狀態
while (!isProduced) {
try {
object.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
System.out.println(Thread.currentThread().getName() + " consumer: " + sth);
object.notifyAll();
isProduced = false;//生產者生產
}
}
public static void main(String[] args) {
final ProducersAndConsumersPattern producerAndConsumerPattern = new ProducersAndConsumersPattern();
for (int i = 0; i < 3; i++) {
new Thread("consumer thread-" + i) {
@Override
public void run() {
while (true) {
producerAndConsumerPattern.consumer();
}
}
}.start();
}
for (int i = 0; i < 3; i++) {
new Thread("producer thread-" + i) {
@Override
public void run() {
while (true) {
producerAndConsumerPattern.producer();
}
}
}.start();
}
}
}
8、wait和sleep的區別
(1)wait是Object的方法，而sleep是Thread的方法
(2)wait會釋放monitor，並將執行緒加入wait-set中，但是sleep不會釋放monitor
(3)使用wait必須要首先獲取monitor，但是sleep不需要
(4)呼叫sleep不需要被喚醒，但是wait呼叫者需要被喚醒
9、自定義顯示鎖
使用synchronized、wait、notify/notifyAll
(1)Lock介面
/**
* 自定義顯示鎖介面
*/
public interface Lock {
class TimeOutException extends Exception {
public TimeOutException(String msg) {
super(msg);
}
}
/**
* 獲取lock
* @throws InterruptedException
*/
void lock() throws InterruptedException;
/**
* 指定時間的lock
* @param millis
* @throws InterruptedException
* @throws TimeOutException
*/
void lock(long millis) throws InterruptedException,TimeOutException;
/**
* 釋放鎖
*/
void unlock();
/**
* 獲取阻塞在當前monitor上的執行緒
* @return
*/
Collection<Thread> getBlockedThreads();
/**
* 獲取當前阻塞佇列的大小
* @return
*/
int getBlockedSize();
}
(2)BooleanLock實現類
/**
* 實現Lock介面：通過一個boolean型別的變量表示同步狀態的獲取
*/
public class BooleanLock implements Lock {
//獲取鎖的標誌
//true：表示當前lock被獲取
//false：表示當前lock沒有被獲取
private boolean initValue;
//阻塞在lock monitor上的執行緒集合
private Collection<Thread> blockedThreads = new ArrayList<Thread>();
//當前獲取到鎖的執行緒:用於在迪呼叫unlock的時候進行check（只有是獲得lock monitor的執行緒才能unlock monitor）
private Thread currentGetLockThread;
public BooleanLock() {
this.initValue = false;
}
/**
* 原子性的獲取鎖
*
* @throws InterruptedException
*/
public synchronized void lock() throws InterruptedException {
//當initValue為true的時候，表示獲取失敗，就加入阻塞佇列中
while (initValue) {
blockedThreads.add(Thread.currentThread());
this.wait(); //加入阻塞佇列中，然後呼叫wait進入(這裡的monitor是當前類的instance)
}
//這裡表示可以獲取鎖，就置標誌位位true，並將自己移除阻塞佇列
blockedThreads.remove(Thread.currentThread());
this.initValue = true;
//設定當前獲得到lock monitor的執行緒
this.currentGetLockThread = Thread.currentThread();
}
/**
* 設定帶有超時的lock方法
* @param millis
* @throws InterruptedException
* @throws TimeOutException
*/
public synchronized void lock(long millis) throws InterruptedException, TimeOutException {
if (millis <= 0) {
lock();
}
long holdTime = millis;
long endTime = System.currentTimeMillis() + millis;
while (initValue) {
//如果已經超時
if (holdTime <= 0) {
throw new TimeOutException("TimeOut");
}
//已經被別的執行緒獲取到lock monitor,就將當前執行緒加入blockedThreads中
blockedThreads.add(Thread.currentThread());
this.wait(millis); //等待指定時間，如果還沒有獲取鎖，下一次迴圈判斷的時候就會丟擲TimeOutException
holdTime =endTime - System.currentTimeMillis();
System.out.println(holdTime);
}
this.initValue = true;
currentGetLockThread = Thread.currentThread();
}
/**
* 原子性的釋放鎖
*/
public synchronized void unlock() {
//檢查是否是當前獲取到lock monitor的執行緒釋放
if (Thread.currentThread() == currentGetLockThread) {
this.initValue = false;
System.out.println(Thread.currentThread().getName() + " release the lock monitor.");
this.notifyAll(); //當前執行緒釋放掉鎖之後，喚醒別的阻塞在這個monitor上的執行緒
}
}
public Collection<Thread> getBlockedThreads() {
//這裡的blockedThreads是read only的
return Collections.unmodifiableCollection(blockedThreads); //設定不允許修改
}
public int getBlockedSize() {
return blockedThreads.size();
}
}
10、自定義執行緒池實現
執行緒池的使用主要是解決兩個問題：①當執行大量非同步任務的時候執行緒池能夠提供更好的效能，在不使用執行緒池時候，每當需要執行非同步任務的時候直接new一個執行緒來執行的話，執行緒的建立和銷燬都是需要開銷的。而執行緒池中的執行緒是可複用的，不需要每次執行非同步任務的時候重新建立和銷燬執行緒；②執行緒池提供一種資源限制和管理的手段，比如可以限制執行緒的個數，動態的新增執行緒等等。
下面是Java中的執行緒池的處理流程，參考我總結的Java併發包中執行緒池ThreadPoolExecutor原理探究。
①corePoolSize：執行緒池核心執行緒個數；
②workQueue：用於儲存等待任務執行的任務的阻塞佇列（比如基於陣列的有界阻塞佇列ArrayBlockingQueue、基於連結串列的無界阻塞佇列LinkedBlockingQueue等等）；
③maximumPoolSize：執行緒池最大執行緒數量；
④ThreadFactory：建立執行緒的工廠；
⑤RejectedExecutionHandler：拒絕策略，表示當佇列已滿並且執行緒數量達到執行緒池最大執行緒數量的時候對新提交的任務所採取的策略，主要有四種策略：AbortPolicy（丟擲異常）、CallerRunsPolicy（只用呼叫者所線上程來執行該任務）、DiscardOldestPolicy（丟掉阻塞佇列中最近的一個任務來處理當前提交的任務）、DiscardPolicy（不做處理，直接丟棄掉）；
⑥keepAliveTime：存活時間，如果當前執行緒池中的數量比核心執行緒數量多，並且當前執行緒是閒置狀態，該變數就是這些執行緒的最大生存時間；
⑦TimeUnit：存活時間的時間單位。
下面是自定義一個簡單的執行緒池實現。
/**
* 自定義執行緒池實現
*/
public class SimpleThreadPool extends Thread {
//執行緒池執行緒數量
private int size;
//任務佇列大小
private int queueSize;
//任務佇列最大大小
private static final int MAX_TASK_QUEUE_SIZE = 200;
//執行緒池中執行緒name字首
private static final String THREAD_PREFIX = "SIMPLE-THREAD-POLL-WORKER";
//執行緒池中的執行緒池所在的THREADGROUP
private static final ThreadGroup THREAD_GROUP = new ThreadGroup("SIMPLE-THREAD-POLL-GROUP");
//任務佇列
private static final LinkedList<Runnable> TASK_QUEUE = new LinkedList<>();
//執行緒池中的工作執行緒集合
private static final List<Worker> WORKER_LIST = new ArrayList<>();
//執行緒池中的執行緒編號
private AtomicInteger threadSeq = new AtomicInteger();
//拒絕策略
private RejectionStrategy rejectionStrategy;
//預設的拒絕策略：丟擲異常
public final static RejectionStrategy DEFAULT_REJECTION_STRATEGY = () -> {
throw new RejectionException("Discard the task.");
};
//執行緒池當前狀態是否為shutdown
private boolean destroy = false;
//增加執行緒池自動維護功能的引數
//預設初始化的時候的執行緒池中的執行緒數量
private int minSize;
//預設初始化時候執行緒數量不夠時候，擴充到activeSize大小
private int activeSize;
//執行緒池中執行緒最大數量：當任務佇列中的數量大於activeSize的時候，會再進行擴充
//當執行完畢，會回收空閒執行緒，數量變為activeSize
private int maxSize;
/**
* 預設構造
*/
public SimpleThreadPool() {
this(2, 4, 8, MAX_TASK_QUEUE_SIZE, DEFAULT_REJECTION_STRATEGY);
}
/**
* 指定執行緒池初始化時候的執行緒個數
* @param minSize
* @param activeSize
* @param maxSize
* @param queueSize
* @param rejectionStrategy
*/
public SimpleThreadPool(int minSize, int activeSize, int maxSize, int queueSize, RejectionStrategy rejectionStrategy) {
if (queueSize <= MAX_TASK_QUEUE_SIZE) {
this.minSize = minSize;
this.activeSize = activeSize;
this.maxSize = maxSize;
this.queueSize = queueSize;
this.rejectionStrategy = rejectionStrategy;
init();
}
}
/**
* 執行緒池本身作為一個執行緒，run方法中維護執行緒中的執行緒個數
*/
@Override
public void run() {
while (!destroy) {
System.out.printf("Pool===#Min:%d; #Active:%d; #Max:%d; #Current:%d; QueueSize:%d\n",
this.minSize, this.activeSize, this.maxSize, this.size, this.TASK_QUEUE.size());
try {
TimeUnit.SECONDS.sleep(5);
//下面是擴充邏輯
//當前任務佇列中的任務數多於activeSize（擴充到activeSize個執行緒）
if (TASK_QUEUE.size() > activeSize && size < activeSize) {
for (int i = size; i < activeSize; i++) {
createWorkerTask();
}
size = activeSize;
System.out.println("Thread pool's capacity has increased to activeSize.");
}
if (TASK_QUEUE.size() > maxSize && size < maxSize) { //擴充到maxSize
for (int i = size; i < maxSize; i++) {
createWorkerTask();
}
size = maxSize;
System.out.println("Thread pool's capacity has increased to maxSize.");
}
//當任務佇列為空時候，需要將擴充的執行緒回收
//以activeSize為界限，大於activeSize並且執行緒池空閒的時候回收空閒的執行緒
//先獲取WORKER_LIST的monitor
//這樣就會在回收空閒執行緒的時候，防止提交任務
synchronized (WORKER_LIST) {
if (TASK_QUEUE.isEmpty() && size > activeSize) {
int releaseNum = size - activeSize;
for (Iterator<Worker> iterator = WORKER_LIST.iterator(); iterator.hasNext(); ) {
if (releaseNum <= 0) {
break;
}
Worker work = (Worker) iterator.next();
work.close();
work.interrupt();
iterator.remove();
System.out.println(Thread.currentThread().getName() + "===============Closed " + work.getName());
releaseNum--;
}
size = activeSize;
System.out.println("Thread pool's capacity has reduced to activeSize.");
}
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
/**
* 初始化執行緒池
*/
private void init() {
for (int i = 0; i < minSize; i++) {
createWorkerTask();
size++;
}
}
/**
* 執行緒池/執行緒狀態
*/
private enum TaskState {
FREE, RUNNING, BLOCKED, TERMINATED
}
/**
* 提交任務到執行緒池
* @param runnable
*/
public void execute(Runnable runnable) {
if (destroy == false) {
if (null != runnable) {
synchronized (TASK_QUEUE) {
if (TASK_QUEUE.size() >= queueSize) {
rejectionStrategy.throwExceptionDirectly();
}
TASK_QUEUE.addLast(runnable);
TASK_QUEUE.notify();//這裡使用notify方法而不是使用notifyAll方法，是因為能夠確定必然有工作者執行緒Worker被喚醒，所以比notifyAll會有更小的開銷（避免了將等待佇列中的工作執行緒全部移動到同步佇列中競爭任務）
}
}
} else {
throw new IllegalStateException("執行緒池已經關閉");
}
}
/**
* 向執行緒池中新增工作執行緒
*/
private void createWorkerTask() {
Worker worker = new Worker(THREAD_GROUP, THREAD_PREFIX + threadSeq.getAndIncrement()); //建立一個工作執行緒
worker.start(); //啟動工作者執行緒
WORKER_LIST.add(worker); //向執行緒池的工作執行緒集合中加入該執行緒
}
public void shutdown() {
//任務佇列中還有任務
while (!TASK_QUEUE.isEmpty()) {
try {
TimeUnit.SECONDS.sleep(20);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
synchronized (WORKER_LIST) {
int curNum = WORKER_LIST.size(); //獲取工作執行緒集合中的執行緒個數
while (curNum > 0) {
for (Worker work : WORKER_LIST) {
if (work.getTaskState() == TaskState.BLOCKED) {
work.interrupt();
work.close();
System.out.println(Thread.currentThread().getName() + "=============Closed.");
curNum--;
} else {
try {
TimeUnit.SECONDS.sleep(4);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
this.destroy = true;
System.out.println("Thread-Pool shutdown.");
}
/**
* 執行緒池的工作執行緒定義
*/
private static class Worker extends Thread {
//volatile型別的taskState，用於在關閉執行緒的時候呼叫
private volatile TaskState taskState = TaskState.FREE;
public Worker(ThreadGroup group, String name) {
super(group, name);
}
/**
* 返回執行緒狀態
*
* @return
*/
public TaskState getTaskState() {
return this.taskState;
}
@Override
public void run() {
BEGIN:
while (this.taskState != TaskState.TERMINATED) {
Runnable runnable = null;
//從任務佇列中取任務（首先獲取TASK_QUEUE的monitor）
synchronized (TASK_QUEUE) {
//判斷任務佇列是否為空
while (TASK_QUEUE.isEmpty()) {
try {
this.taskState = TaskState.BLOCKED;
TASK_QUEUE.wait(); //任務佇列為空，當前執行緒就進入等待狀態
} catch (InterruptedException e) {
System.out.println("Interrupted Close.");
break BEGIN; //被打斷之後，重新執行自己的run方法
}
}
//被喚醒之後，且判斷任務佇列不為空，就從任務佇列中獲取一個Runnable
runnable = TASK_QUEUE.removeFirst();
}
if (null != runnable) {
this.taskState = TaskState.RUNNING;//當前執行任務的執行緒為RUNNING
runnable.run();//執行run
this.taskState = TaskState.FREE; //任務執行完畢，狀態變為FREE
}
}
}
public void close() {
this.taskState = TaskState.TERMINATED;
}
}
public interface RejectionStrategy {
//直接丟擲異常
void throwExceptionDirectly() throws RejectionException;
}
//自定義異常類：拒絕策略——丟擲異常
public static class RejectionException extends RuntimeException {
public RejectionException(String msg) {
super(msg);
}
}
//測試編寫的SimpleThreadPool
public static void main(String[] args) {
SimpleThreadPool simpleThreadPool = new SimpleThreadPool();
for (int index = 0; index < 50; index++) {
simpleThreadPool.execute(new Runnable() {
@Override
public void run() {
System.out.println(Thread.currentThread().getName() + " start the task");
//do sth
int i = 0;
while (i++ < 50) { }
System.out.println(Thread.currentThread().getName() + " finish the task");
}
});
}
simpleThreadPool.start(); //執行緒池維護引數
simpleThreadPool.shutdown();
}
}11、單例模式的幾種實現
(1)立即載入/"餓漢模式"
//餓漢式是執行緒安全的，因為虛擬機器保證只會裝載一次，使用的時候是已經載入初始化好的instance。
public class SingletonPattern1 {
private static final SingletonPattern1 instance = new SingletonPattern1();
private SingletonPattern1() {
//do sth about init
}
public static SingletonPattern1 getInstance() {
return instance;
}
}(2)延遲載入/"懶漢模式"
①執行緒不安全的懶漢模式
//執行緒不安全的懶漢式
public class SingletonPattern2 {
private static SingletonPattern2 instance;
private SingletonPattern2() {
//do sth about init
}
public static SingletonPattern2 getInstance() {
//假設兩個執行緒執行thread1、thread2；
//(1)其中thread1執行到if判斷的時候instance為null，這時候恰好因為執行緒排程thread1失去執行權；
//(2)thread2也執行到此處，並進入if程式碼塊new一個instance，然後返回；
//(3)thread1再次從此處執行，因為判斷到的instance為null，也同樣new一個instance然後返回；
//(4)這樣thread1和thread2的instance實際上並不是相同的了
if(null == instance) {
instance = new SingletonPattern2();
}
return SingletonPattern2.instance;
}
}②使用synchronized解決
//使用synchronized的執行緒安全的懶漢式
public class SingletonPattern2 {
private static SingletonPattern2 instance;
private SingletonPattern2() {
//do sth about init
}
//缺點：實際上只有一個執行緒是寫的操作（獲得monitor之後new一個instance），後面的執行緒因為因為已經建立了instance，就是相當於讀的操作，但是read的操作還是加鎖同步了（序列化了），效率較低
public synchronized static SingletonPattern2 getInstance() {
if(null == instance) {
instance = new SingletonPattern2();
}
return SingletonPattern2.instance;
}
}③使用DCL機制(無volatile)
//double check的方式
public class SingletonPattern3 {
private static SingletonPattern3 instance;
private SingletonPattern3() {
//do sth about init
}
public static SingletonPattern3 getInstance() {
//假設兩個執行緒執行thread1、thread2，都判斷instance為null
if (null == instance) {
synchronized (SingletonPattern3.class) {
//(1)其中thread1進入同步程式碼塊之後，new了一個instance
//(2)在thread1退出同步程式碼塊之後，thread2進入同步程式碼塊，因為instance不為null，所以直接退出同步塊，返回建立好的instance
if(null == instance) {
instance = new SingletonPattern3();
}
}
}
//(3)現在已經建立好了instace，後面的執行緒在呼叫getInstance()方法的時候就會直接到此處，返回instance
return SingletonPattern3.instance;
}
}④使用volatile的DCL
//double check + volatile的方式
public class SingletonPattern3 {
//volatile禁止指令重排序(happens-before中有一條volatile變數規則：對一個volatile變數的寫的操作先行發生與對這個變數的讀操作)
private volatile static SingletonPattern3 instance;
//SingletonPattern3其他的一些引用型別的屬性PropertyXXX propertyxxx;
private SingletonPattern3() {
//do sth about init
}
public static SingletonPattern3 getInstance() {
if (null == instance) {
synchronized (SingletonPattern3.class) {
if(null == instance) {
//instance = new SingletonPattern3();這句，這裡看起來是一句話，但實際上它並不是一個原子操作，在被編譯後在JVM執行的對應彙編程式碼做了大致3件事情：
//(1)分配記憶體
//(2)呼叫構造器
//(3)將instance物件指向分配的記憶體空間首地址(這時候的instance不為null)
//但由於指令重排序(Java編譯器允許處理器亂序執行)的存在，上面三個步驟可能是1-2-3也可能是1-3-2，注意，如果是1-3-2的情況就可能出現問題，我們來分析一下可能出現的問題：
//I：假設現在兩個執行緒thread1、thread2，現在threa1獲取到monitor，然後按照上面的1-3-2執行，
//II：假設在3執行完畢、2未執行之前(或者說2只執行一部分，SingletonPattern3中的引用型別的屬性一部分還是null)，這個時候切換到thread2上，
//III：這時候instance因為已經在thread1內執行過了(3)，instance已經是非空了，所以thread2直接拿走instance，然後使用，但是實際上instance指向的記憶體地址並沒有呼叫構造器初始化的，這就可能會出現問題了。
instance = new SingletonPattern3();
}
}
}
return SingletonPattern3.instance;
}
}(3)Holder方式
其中關於主動引用參考總結的JVM之虛擬機器類載入機制中講到的主動引用和被動引用。
//Holder方式：延遲載入、不加鎖、執行緒安全
public class SingletonPattern4 {
private SingletonPattern4() {
//do sth about init
}
//在靜態內部類中，有SingletonPattern4的例項，並且直接被初始化
private static class Holder {
private static SingletonPattern4 instance = new SingletonPattern4();
}
//返回的是Holer的靜態成員instance
public static SingletonPattern4 getInstance() {
//在SingletonPattern4中沒有instance的靜態成員，而是將其放到了靜態內部類Holder之中，因此在SingletonPattern4類的初始化中並不會建立Singleton(延遲載入)的例項，Holder中定義了SingletonPattern4的靜態變數，並且直接進行了初始化。當Holder被主動引用的時候會建立SingletonPattern4的例項，SingletonPattern4例項的建立過程在Java程式編譯時候由同步方法<clinit>()方法執行，保證記憶體的可見性、JVM指令的順序性和原子性。
return Holder.instance;
}
}(4)列舉方式
//列舉方式實現
public class SingletonPattern5 {
//列舉型不允許被繼承、是執行緒安全的，只被例項化一次(同樣下面使用類似Holder的方式)
private enum EnumHolderSingleton {
INSTANCE;
private final SingletonPattern5 instance;
EnumHolderSingleton() {
this.instance = new SingletonPattern5();
}
private SingletonPattern5 getInstance() {
return instance;
}
}
public SingletonPattern5 getInstance() {
return EnumHolderSingleton.INSTANCE.getInstance();
}
}