一、wait、notify谆棺、notifyAll綜述
wait/notify/notifyAll是JAVA提供通過對(duì)鎖的監(jiān)視的方式進(jìn)行線程間的協(xié)作。正因?yàn)橥ㄟ^對(duì)鎖的監(jiān)視鹃骂，因此wait()/notify()/notifyAll()方法的調(diào)用必須先獲得鎖,再調(diào)用鎖對(duì)象的wait()/notify()/notifyAll()方法。因此一般情況下配合synchronized進(jìn)行使用罢绽。
二畏线、交互過程
鎖的監(jiān)視器模式
在并發(fā)編程中，我們?cè)O(shè)計(jì)的對(duì)象有可能會(huì)被多線程進(jìn)行調(diào)用良价，而這種調(diào)用通常會(huì)改變對(duì)象的狀態(tài)寝殴，那么在并發(fā)編程的時(shí)候通常為了保證這種對(duì)象狀態(tài)改變的原子性，一般情況下會(huì)通過鎖的進(jìn)行處理明垢。那問題來了蚣常，如果多個(gè)線程在等待獲取鎖的情況下，JAVA是怎樣處理的呢袖外？JAVA對(duì)于鎖的釋放和協(xié)作通知是通過監(jiān)視器模式來進(jìn)行處理史隆。如下圖為一個(gè)鎖的釋放通知魂务，線程A獲取到鎖后曼验，線程B、C粘姜、D分別去取獲取鎖鬓照，發(fā)現(xiàn)鎖已經(jīng)被線程A獲取。線程B孤紧、C豺裆、D進(jìn)入阻塞狀態(tài)，等待監(jiān)視器的通知。當(dāng)線程A釋放鎖后臭猜，鎖監(jiān)視器通知請(qǐng)求獲取鎖的隊(duì)列中的線程躺酒。
wait、notify蔑歌、notifyAll都是屬于Object的方法羹应，而在wait、notify次屠、notifyAll的設(shè)計(jì)上只有一個(gè)Condition隊(duì)列园匹，而在重入鎖ReentrantLock中則有多個(gè)Condition隊(duì)列。多個(gè)Condition隊(duì)列可以隔離每個(gè)Condition的等待和喚醒劫灶。ReentrantLock屬于編程式的鎖機(jī)制裸违，而wait、notify本昏、notifyAll通過synchronized進(jìn)行使用供汛，而synchronized屬于關(guān)鍵字，因此由JAVA內(nèi)部實(shí)現(xiàn)涌穆。
交互時(shí)序圖
使用wait\notify\notify的目的是通過線程協(xié)作使并發(fā)線程串行化執(zhí)行紊馏。
Thread(wait)線程通過鎖對(duì)象獲取到鎖進(jìn)入到同步方法中執(zhí)行。然后通過鎖對(duì)象的wait方法蒲犬，通知鎖監(jiān)視器朱监，當(dāng)前線程掛起并且釋放鎖。
Thread(notify\noityall)線程獲取到鎖后執(zhí)行同步方法原叮，然后調(diào)用鎖對(duì)象的notify\notifyall方法赫编，告訴鎖監(jiān)視器，然后釋放鎖奋隶。鎖監(jiān)視器通知Thread(wait)線程擂送。
Thread(wait)線程收到鎖監(jiān)視器的通知后，恢復(fù)線程并且再次獲得鎖唯欣，然后執(zhí)行代碼嘹吨。
三、wait境氢、notify蟀拷、notifyAll標(biāo)準(zhǔn)范式
- 等待方
1.獲取鎖
2.循環(huán)判斷是否符合條件，若不符合等待萍聊，若符合執(zhí)行業(yè)務(wù)邏輯 - 通知方
1.獲取鎖
2.執(zhí)行業(yè)務(wù)邏輯
3.通知所有等待方问芬。
public static class BookARoomNotification {
public boolean isNotification = false;
public void waitNotification() {
synchronized (this) {
while (!isNotification) {
try {
wait();
System.out.println("book a room");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public void notification() {
synchronized (this) {
this.isNotification = true;
System.out.println("Booking Time");
notifyAll();
}
}
}
四、notify寿桨、notifyAll的區(qū)別
通過上面圖的分析我們應(yīng)該使用notifyall而非notify此衅。
五、實(shí)現(xiàn)等待超時(shí)
1、wait(long timeout)
JAVA為我們提供了wait的方法還給我們提供了等待超時(shí)方法wait(long timeout)挡鞍。而我們可以通過wait(long timeout)來實(shí)現(xiàn)等待超時(shí)骑歹。我們通過標(biāo)準(zhǔn)范式的修改實(shí)現(xiàn)等待超時(shí)。
2墨微、實(shí)現(xiàn)超時(shí)等待的標(biāo)準(zhǔn)范式
- 等待方：
(1) 獲取鎖
(2)根據(jù)傳入的超時(shí)時(shí)間確定結(jié)束時(shí)間
long overTime = System.currentTimeMillis() +timeMillis;
(3)進(jìn)行等待超時(shí)并且循環(huán)判斷條件和當(dāng)前時(shí)間是否已經(jīng)超時(shí)陵刹。
long remain = timeMillis;
while(pool.isEmpty()&&remain>0) {
pool.wait(timeMillis);
remain = overTime - System.currentTimeMillis();
}
(4)超時(shí)后，繼續(xù)判斷條件是否符合欢嘿，符合執(zhí)行業(yè)務(wù)邏輯衰琐。
Connection connection= null;
if(!pool.isEmpty()) {
connection = pool.removeFirst();
}
- 通知方
(1)獲取鎖
(2)執(zhí)行業(yè)務(wù)邏輯
(3)通知所有等待方。
3炼蹦、獲取連接池超時(shí)等待
DBPool.java
public class DBPool {
private LinkedList<Connection> pool = new LinkedList<>();
public DBPool(int initalSize) {
super();
for (int i = 0; i < initalSize; i++) {
pool.addLast(MysqlConnection.fetchConnection());
}
}
public int getPoolCount() {
return pool.size();
}
public Connection fetchConnection(long timeMillis) throws InterruptedException {
synchronized (pool) {
if (timeMillis <= 0) {
while (pool.isEmpty()) {
pool.wait();
}
return pool.removeFirst();
} else {
long overTime = System.currentTimeMillis() +timeMillis;
long remain = timeMillis;
while(pool.isEmpty()&&remain>0) {
pool.wait(timeMillis);
remain = overTime - System.currentTimeMillis();
}
Connection connection= null;
if(!pool.isEmpty()) {
connection = pool.removeFirst();
}
return connection;
}
}
}
public void closeConnection(Connection conn) {
if(conn!=null) {
synchronized (pool) {
pool.addLast(conn);
pool.notifyAll();
}
}
}
}
MysqlConnection.java
其他方法默認(rèn)實(shí)現(xiàn)即可
public class MysqlConnection implements Connection{
/*拿一個(gè)數(shù)據(jù)庫連接*/
public static final Connection fetchConnection(){
return new MysqlConnection();
}
@Override
public void commit() throws SQLException {
try {
Thread.sleep(70);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public Statement createStatement() throws SQLException {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
return null;
}
DBpoolTest.java
public class DBpoolTest {
static DBPool pool = new DBPool(10);
static CountDownLatch countDownLatch;
static int threadCount = 50;
static int getConnCount = 20;
static class GetConnectionThread implements Runnable {
int count;
AtomicLong got;
AtomicLong notGot;
public GetConnectionThread(int count, AtomicLong got, AtomicLong notGot) {
super();
this.count = count;
this.got = got;
this.notGot = notGot;
}
@Override
public void run() {
while (count > 0) {
try {
Connection conn = pool.fetchConnection(1000);
if (conn != null){
try {
conn.createStatement();
conn.commit();
} catch (SQLException e) {
e.printStackTrace();
} finally {
pool.closeConnection(conn);
got.incrementAndGet();
}
} else {
notGot.incrementAndGet();
System.out.println(Thread.currentThread().getName() + "等待超時(shí)");
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
count--;
}
}
countDownLatch.countDown();
}
}
public static void main(String[] args) throws InterruptedException {
countDownLatch = new CountDownLatch(threadCount);
AtomicLong got = new AtomicLong();
AtomicLong notGot = new AtomicLong();
for (int i = 0; i < threadCount; i++) {
Thread t = new Thread(new GetConnectionThread(getConnCount, got, notGot), "DBFactory" + i);
t.start();
}
countDownLatch.await();
System.out.println("總共嘗試了: " + (threadCount * getConnCount));
System.out.println("拿到連接的次數(shù)： " + got);
System.out.println("沒能連接的次數(shù)： " + notGot);
}
}
