java線程間的協(xié)作

?本次內(nèi)容主要講等待/通知機(jī)制以及用等待/通知機(jī)制手寫一個(gè)數(shù)據(jù)庫連接池。

?

1、為什么線程之間需要協(xié)作

　
　線程之間相互配合，完成某項(xiàng)工作，比如：一個(gè)線程修改了一個(gè)對(duì)象的值，而另一個(gè)線程感知到了變化，然后進(jìn)行相應(yīng)的操作，整個(gè)過程開始于一個(gè)線程，而最終執(zhí)行又是另一個(gè)線程。前者是生產(chǎn)者，后者就是消費(fèi)者，這種模式隔離了“做什么”（What）和“怎么做”（How）。簡(jiǎn)單的辦法是讓消費(fèi)者線程不斷地循環(huán)檢查變量是否符合預(yù)期，在while循環(huán)中設(shè)置不滿足的條件，如果條件滿足則退出while循環(huán)，從而完成消費(fèi)者的工作。這樣進(jìn)行線程之間的協(xié)作卻存在如下2個(gè)問題：

（1）難以確保及時(shí)性。

（2）難以降低開銷。如果降低睡眠的時(shí)間，比如休眠1毫秒，這樣消費(fèi)者能更加迅速地發(fā)現(xiàn)條件變化，但是卻可能消耗更多的處理器資源，造成了無端的浪費(fèi)。

那么有沒有什么辦法可以解決以上2個(gè)問題呢？此時(shí)等待/通知機(jī)制毫不客氣的站出來說，都讓開，交給我，我能行！　　

2、等待/通知機(jī)制

2.1 等待/通知機(jī)制介紹

等待/通知機(jī)制是指一個(gè)線程A調(diào)用了對(duì)象obejct的wait()方法進(jìn)入等待狀態(tài)，而另一個(gè)線程B調(diào)用了對(duì)象obejct的notify()或者
notifyAll()方法，線程A收到通知后從對(duì)象obejct的wait()方法返回，進(jìn)而執(zhí)行后續(xù)操作。上述兩個(gè)線程通過對(duì)象obejct來完成交互，而對(duì)象上的
wait()和notify/notifyAll()的關(guān)系就如同開關(guān)信號(hào)一樣，用來完成等待方和通知方之間的交互工作。

notify()：

通知一個(gè)在對(duì)象上等待的線程，使其從wait()方法返回，而返回的前提是該線程獲取到了對(duì)象的鎖。哪個(gè)線程能得到通知是隨機(jī)的，不能指定。

notifyAll()：

通知所有等待在該對(duì)象上的線程，這些線程會(huì)去競(jìng)爭(zhēng)對(duì)象鎖，得到鎖的某一個(gè)線程可以繼續(xù)執(zhí)行wait()后的邏輯。

wait():

調(diào)用該方法的線程進(jìn)入 WAITING狀態(tài)，只有等待另外線程的通知或被中斷才會(huì)返回。需要注意，調(diào)用wait()方法后，會(huì)釋放對(duì)象的鎖。

wait(long):

超時(shí)等待一段時(shí)間，這里的參數(shù)時(shí)間是毫秒，也就是等待長(zhǎng)達(dá)n毫秒，如果沒有通知就超時(shí)返回。

wait (long,int):

對(duì)于超時(shí)時(shí)間更細(xì)粒度的控制，可以達(dá)到納秒。

2.1 等待/通知機(jī)制使用的標(biāo)準(zhǔn)范式

等待方遵循如下原則：

（1）獲取對(duì)象的鎖。

（2）如果條件不滿足，那么調(diào)用對(duì)象的wait()方法，被通知后仍要檢查條件。

（3）條件滿足則執(zhí)行對(duì)應(yīng)的邏輯。

用一段偽代碼表示：
synchronized (對(duì)象) { while (條件不滿足) { 對(duì)象.wait(); } 對(duì)應(yīng)邏輯處理 }
通知方遵循如下原則：

（1）獲得對(duì)象的鎖。

（2）改變條件。

（3）通知所有等待在對(duì)象上的線程。

（4）通知方法放在同步代碼塊的最后一行。

用一段偽代碼表示：
synchronized (對(duì)象) { 改變條件 對(duì)象.notifyAll(); }
　　在調(diào)用wait()、notify()和notifyAll()方法之前，線程必須要獲得該對(duì)象的對(duì)象鎖，即只能在同步方法或同步塊中調(diào)用wait()方法、
notify()和notifyAll()方法。調(diào)用wait()方法后，當(dāng)前線程釋放鎖，
執(zhí)行notify()和notifyAll()方法的線程退出synchronized代碼塊的時(shí)候，假設(shè)是執(zhí)行的notifyAll()，會(huì)喚醒所有處于等待的線程，這些線程會(huì)去競(jìng)爭(zhēng)對(duì)象鎖。如果其中一個(gè)線程A獲得了該對(duì)象鎖，線程A就會(huì)繼續(xù)往下執(zhí)行，其余被喚醒的線程處于阻塞狀態(tài)。在線程A退出synchronized代碼塊釋放鎖后，其余已經(jīng)被喚醒的處于阻塞狀態(tài)的線程將會(huì)繼續(xù)競(jìng)爭(zhēng)該鎖，一直進(jìn)行下去，直到所有被喚醒的線程都執(zhí)行完畢。

2.2?notify()和notifyAll()應(yīng)該用誰

　　盡可能用notifyAll()，謹(jǐn)慎使用notify()，因?yàn)閚otify()只會(huì)喚醒一個(gè)線程，我們無法確保被喚醒的這個(gè)線程一定就是我們需要喚醒的線程。

2.3 手寫一個(gè)數(shù)據(jù)庫連接池


　　我們?cè)谑褂脭?shù)據(jù)庫連接池的時(shí)候，當(dāng)某一個(gè)線程超過配置的最大等待時(shí)長(zhǎng)還沒有拿到連接時(shí)，就會(huì)報(bào)出異常。我們使用等待/通知機(jī)制來模擬一個(gè)數(shù)據(jù)庫連接池。分別定義連接類、連接池實(shí)現(xiàn)類和測(cè)試類。

連接類：
import java.sql.*; import java.util.Map; import java.util.Properties; import
java.util.concurrent.Executor;public class MySqlConnection implements
Connection {public static final Connection createConnection(){ return new
MySqlConnection(); }//todo 其余接口使用默認(rèn)實(shí)現(xiàn)，這里就不一一給出。 }
連接池實(shí)現(xiàn)類：
import java.sql.Connection; import java.util.LinkedList; public class
MyConnectionPool {/**裝連接的容器*/ private static LinkedList<Connection> pool = new
LinkedList<>(); /** * 初始化連接池 * @param poolSize */ public MyConnectionPool(int
poolSize){if(poolSize > 0){ for (int i = 0; i < poolSize; i++) {
pool.add(MySqlConnection.createConnection()); } } }/** * 釋放一個(gè)連接 * @param
connection*/ public void releaseConnection(Connection connection){ if
(connection !=null){ synchronized (pool){ pool.add(connection);
pool.notifyAll(); } } }/** * 獲取一個(gè)連接 * @param millions 超時(shí)時(shí)間 * @return */ public
Connection getConnection(long millions){ synchronized (pool){ if(millions <= 0){
while (pool.isEmpty()){ try { pool.wait(); }catch (InterruptedException e){
e.printStackTrace(); } }return pool.removeFirst(); } else { //計(jì)算超時(shí)時(shí)刻 long
overTime = System.currentTimeMillis() + millions; //剩余等待時(shí)長(zhǎng) long remaining =
millions;//當(dāng)剩余等待時(shí)間大于0并且連接池為空，就等待 while (remaining > 0 && pool.isEmpty()){ try {
pool.wait(); }catch (InterruptedException e){ e.printStackTrace(); } //
被喚醒后重新計(jì)算剩余等待時(shí)長(zhǎng) remaining = overTime - System.currentTimeMillis(); } Connection
result= null; if(!pool.isEmpty()){ result = pool.removeFirst(); } return
result; } } } }
測(cè)試類：
import java.sql.Connection; import java.sql.Statement; import
java.util.concurrent.CountDownLatch;import
java.util.concurrent.atomic.AtomicInteger;public class MyConnectionPoolTest {
/**初始化10個(gè)數(shù)據(jù)庫連接*/ static MyConnectionPool pool = new MyConnectionPool(10); /**
等待時(shí)長(zhǎng)為1秒*/ static long maxWait = 1000; /**獲取連接的線程數(shù)量*/ static int threadNumber =
500; /**所有獲取連接線程執(zhí)行完成后再執(zhí)行main線程*/ static CountDownLatch mainCountDownLatch = new
CountDownLatch(threadNumber);/**保證所有獲取連接線程同時(shí)執(zhí)行*/ static CountDownLatch
workCountDownLatch =new CountDownLatch(threadNumber); static class Worker
implements Runnable{ AtomicInteger success; AtomicInteger fail; public
Worker(AtomicInteger success, AtomicInteger fail){this.success = success; this
.fail = fail; } @Override public void run() { try { workCountDownLatch.await();
Connection connection= pool.getConnection(maxWait); if(connection != null){ try
{ Statement statement= connection.createStatement(); Thread.sleep(30);//
休眠30毫秒模擬實(shí)際業(yè)務(wù) connection.commit(); }finally {
pool.releaseConnection(connection); success.getAndAdd(1); } } else {
fail.getAndAdd(1); System.out.println(Thread.currentThread().getName() +
"等待超時(shí)，沒有拿到連接！"); } }catch (Exception e){ e.printStackTrace(); }
mainCountDownLatch.countDown(); } }public static void main(String[] args) {
AtomicInteger success= new AtomicInteger(0);//記錄成功次數(shù) AtomicInteger fail = new
AtomicInteger(0);//記錄失敗次數(shù) for (int i = 0; i < threadNumber; i++) { new Thread(
new Worker(success, fail)).start(); workCountDownLatch.countDown(); } try {
mainCountDownLatch.await(); }catch (InterruptedException e){
e.printStackTrace(); } System.out.println("總共嘗試獲取連接次數(shù)：" + threadNumber);
System.out.println("成功次數(shù)：" + success.get()); System.out.println("失敗次數(shù)：" +
fail.get()); } }
運(yùn)行程序，從輸出結(jié)果可以看出，通過通知/等待超時(shí)模式成功的實(shí)現(xiàn)了一個(gè)簡(jiǎn)易的數(shù)據(jù)庫連接池。



?2.4 常見面試題

調(diào)用yield() 、sleep()、wait()、notify()等方法對(duì)鎖有何影響？

答：yield() 、sleep()被調(diào)用后，都不會(huì)釋放當(dāng)前線程所持有的鎖。

調(diào)用wait()方法后，會(huì)釋放當(dāng)前線程持有的鎖，而且當(dāng)前線程被喚醒后，會(huì)重新去競(jìng)爭(zhēng)鎖，得到鎖到后才會(huì)執(zhí)行wait()方法后面的代碼。


調(diào)用notify()系列方法后，對(duì)鎖無影響，線程只有在synchronized同步代碼執(zhí)行完后才會(huì)自然而然的釋放鎖，所以notify()系列方法一般都是synchronized同步代碼的最后一行。

?

線程的并發(fā)工具類在下一篇文章中介紹，在閱讀過程中如發(fā)現(xiàn)描述有誤，請(qǐng)指出，謝謝。

?

?

?

?

?

?

?

?

?

• ? 上一篇：復(fù)盤|怎樣做好一次復(fù)盤
• ? 下一篇：軟件測(cè)試如何獲得高薪？