Java高并发系列: 使用wait - notify实现高效异步方法

1. 背景

在项目开发中, 通常会有异步执行操作, 例如: 提交一个异步清空一系列数据库中ID = ${_id} 的记录, 这个时候通常的做法是主线程将任务添加到一个异步队列中, 后台维护一个线程不断地循环扫描这个队列, 如果有需要执行的任务, 则执行相应的逻辑. 如下图所示:

2. 一个简单的异步执行方法

代码实现如下所示:

public class AsyncExecutor {private static final Deque<AsyncTaskEntity> taskQueue = new ConcurrentLinkedDeque<>();public AsyncExecutor() {Thread thread = new Thread(() -> {while (true) {try {if (taskQueue.isEmpty()) {// 休眠50毫秒ThreadUtil.sleep(50);continue;}AsyncTaskEntity entity = taskQueue.pollFirst();execute(entity);} catch (Exception e) {LOGGER.error("异步执行任务出现异常!", e);}}});thread.setName("异步任务执行器");thread.start();System.out.println("analysis异步队列任务启动完成！");}public static <T> void asyncExecute(AsyncTaskEntity<T> entity) {taskQueue.push(entity);}
}/*** 队列中任务对象封装*/
@Data
public class AsyncTaskEntity <T>{// 消费的参数private T param;public AsyncTaskEntity(T param){this.param = param;}
}

有了上面的异步执行器之后, 这里我们写一个main方法, 在main方法中通过异步的方式执行一些任务:

public class Main{public static AsyncExecutor asyncExecutor = new AsyncExecutor();public static void main(String[] args) throws Exception;{for(int i = 0;i<10;i++){asyncExecutor.asyncExecute(new AsyncTaskEntity<Integer>(i));}Thread.sleep(10_000);}
}

到此为止一个简单清晰的异步调用逻辑就已经写完了. 但是现在不得不考虑一个事情, 异步线程中while(true)会一直空转, 即使没有任务。因此下面我们使用wait - notify进行优化

3. 优化版本1 - 使用wait - notify

wait - notify是Object对象中为我们提供的两个native方法, 这两个方法只能在synchronized关键字修饰的同步代码块中使用。Thread.sleep()方法不会释放锁，wait()方法会释放锁，直到被其他线程notify之后，才会重新获得锁。我们对上述异步队列进行改造：

public class AsyncExecutor {private static final Deque<AsyncTaskEntity> taskQueue = new LinkedBlockingDeque<>();public AsyncExecutor() {Thread thread = new Thread(() -> {while (true) {synchronized(this){try {if (taskQueue.isEmpty()) {this.wait();}AsyncTaskEntity entity = taskQueue.pollFirst();execute(entity);} catch (Exception e) {LOGGER.error("异步执行任务出现异常!", e);}}}});thread.setName("异步任务执行器");thread.start();System.out.println("analysis异步队列任务启动完成！");}public synchronized <T> void asyncExecute(AsyncTaskEntity<T> entity) {taskQueue.push(entity);this.notify();}
}

经过上面改造之后，当后台队列中任务为空时，轮训扫描线程就会进入到this.wait()逻辑，此时会释放synchronized获取到的this锁。因此调用asyncExecute()方法会正常的获取到this锁。当push数据之后，执行了notify，便会唤醒一个当前this上正在wait()的线程。这种方式就避免了占用资源始终空转的问题。

其实结合线程的三种核心状态可以更好的理解，当调用wait()方法时，该线程会放弃CPU执行权，进入到阻塞状态，直到被其他线程唤醒(notify())。