一、wait、notify、notifyAll综述
wait/notify/notifyAll是JAVA提供通过对锁的监视的方式进行线程间的协作。正因为通过对锁的监视，因此wait()/notify()/notifyAll()方法的调用必须先获得锁,再调用锁对象的wait()/notify()/notifyAll()方法。因此一般情况下配合synchronized进行使用。
二、交互过程
锁的监视器模式
在并发编程中，我们设计的对象有可能会被多线程进行调用，而这种调用通常会改变对象的状态，那么在并发编程的时候通常为了保证这种对象状态改变的原子性，一般情况下会通过锁的进行处理。那问题来了，如果多个线程在等待获取锁的情况下，JAVA是怎样处理的呢？JAVA对于锁的释放和协作通知是通过监视器模式来进行处理。如下图为一个锁的释放通知，线程A获取到锁后，线程B、C、D分别去取获取锁，发现锁已经被线程A获取。线程B、C、D进入阻塞状态，等待监视器的通知。当线程A释放锁后，锁监视器通知请求获取锁的队列中的线程。
image.png
wait、notify、notifyAll都是属于Object的方法，而在wait、notify、notifyAll的设计上只有一个Condition队列，而在重入锁ReentrantLock中则有多个Condition队列。多个Condition队列可以隔离每个Condition的等待和唤醒。ReentrantLock属于编程式的锁机制，而wait、notify、notifyAll通过synchronized进行使用，而synchronized属于关键字，因此由JAVA内部实现。
交互时序图
交互过程.png
使用wait\notify\notify的目的是通过线程协作使并发线程串行化执行。
-
Thread(wait)线程通过锁对象获取到锁进入到同步方法中执行。然后通过锁对象的wait方法，通知锁监视器，当前线程挂起并且释放锁。
-
Thread(notify\noityall)线程获取到锁后执行同步方法，然后调用锁对象的notify\notifyall方法，告诉锁监视器，然后释放锁。锁监视器通知Thread(wait)线程。
-
Thread(wait)线程收到锁监视器的通知后，恢复线程并且再次获得锁，然后执行代码。
三、wait、notify、notifyAll标准范式
- 等待方
1.获取锁
2.循环判断是否符合条件，若不符合等待，若符合执行业务逻辑 - 通知方
1.获取锁
2.执行业务逻辑
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的区别
notify.png
notifyall.png
通过上面图的分析我们应该使用notifyall而非notify。
五、实现等待超时
1、wait(long timeout)
JAVA为我们提供了wait的方法还给我们提供了等待超时方法wait(long timeout)。而我们可以通过wait(long timeout)来实现等待超时。我们通过标准范式的修改实现等待超时。
2、实现超时等待的标准范式
- 等待方：
(1) 获取锁
(2)根据传入的超时时间确定结束时间
long overTime = System.currentTimeMillis() +timeMillis;
(3)进行等待超时并且循环判断条件和当前时间是否已经超时。
long remain = timeMillis;
while(pool.isEmpty()&&remain>0) {
pool.wait(timeMillis);
remain = overTime - System.currentTimeMillis();
}
(4)超时后，继续判断条件是否符合，符合执行业务逻辑。
Connection connection= null;
if(!pool.isEmpty()) {
connection = pool.removeFirst();
}
- 通知方
(1)获取锁
(2)执行业务逻辑
(3)通知所有等待方。
3、获取连接池超时等待
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
其他方法默认实现即可
public class MysqlConnection implements Connection{
/*拿一个数据库连接*/
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() + "等待超时");
}
} 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("拿到连接的次数： " + got);
System.out.println("没能连接的次数： " + notGot);
}
}
网友评论