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;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

有了上面的异步执行器之后, 这里我们写一个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);
	}
}
1
2
3
4
5
6
7
8
9
10

到此为止一个简单清晰的异步调用逻辑就已经写完了. 但是现在不得不考虑一个事情, 异步线程中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();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

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

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