使用Java的等待/通知机制实现一个简单的阻塞队列

Java的等待/通知机制

Java的等待通知机制是多线程间进行通信的一种方式。

有三个重要的方法：wait()，notify() 和以及notifyAll()

• wait()：该方法用于让当前线程（即调用该方法的线程）进入等待状态并且释放掉该对象上的锁。而这个线程会在以下三种情况下从 wait() 返回到执行状态：
  • 其他线程调用了同一个对象的 notify() 方法。
  • 其他线程调用了同一个对象的 notifyAll() 方法。
  • 其他线程调用了该线程的 interrupt() 方法，线程收到中断信号。
• notify()：唤醒在此对象监视器上等待的单个线程，选择其唤醒的线程是任意的，并且是随机的。唤醒后，等待线程会尝试重新获取锁并继续执行。
• notifyAll()：唤醒在此对象监视器上等待的所有线程。

实现阻塞队列

我们就是使用这几个方法来实现了一个简单的阻塞队列，借助生产者-消费者模型来构建更方便理解。而且是较为简单的单个生产者和单个消费者。
我们有一个容量是n的仓库，起初仓库是空的，此时消费者来消费是不可以的，因此就被阻塞，而当生产者生产了一个物品之后，就会调用notify，唤醒在此对象监视器上等待的单个线程（消费者）。当仓库是满的时候，此时生产者再生产也是不行的，也会被阻塞，此时当消费消费之后，也会调用notify唤醒在此对象监视器上等待的单个线程（生产者）。

import java.util.*;public class BlockQ<T> {// 队列的最大容量static final int MAX_CAPACITY = 10;// 队列的默认容量static final int DEFAULT_CAPACITY = 5;// 队列的最小容量static final int MIN_CAPACITY = 1;// 队列（仓库，仓库不满生产者才能生产，仓库不空消费者才能消费）private Queue<T> q = new LinkedList<>();// 队列的容量private int capacity;public BlockQ(){this.capacity = DEFAULT_CAPACITY;}public BlockQ(int capacity){this.capacity = Math.min(MAX_CAPACITY, Math.max(MIN_CAPACITY, capacity));}public void addT(T record) throws InterruptedException {synchronized (q){while(q.size() == this.capacity){System.out.println("size:" + q.size() + ",records:" + Arrays.toString(q.toArray()));// 该线程等待，并释放q上的锁q.wait();}System.out.println("生产者生产的数字: " + record);q.offer(record);// 唤醒一个在q上等待的线程q.notify();}}public T getT() throws InterruptedException {synchronized (q){while(q.size() == 0){System.out.println("size:" + q.size() + ",records:" + Arrays.toString(q.toArray()));q.wait();}T res = q.poll();System.out.println("消费者获取到数字: " + res);q.notify();return res;}}
}

测试

我们创建了两个线程，一个生产者线程，一个消费者线程。在主线程中设置Thread.sleep(7000)，是为了让生产者先生产，然后我们也能提前看见生产者被阻塞。而后面也会出现消费者被阻塞的情况，这些都是系统设置的时间片，我们无法改变。但是我们可以设置生产者生产和消费者消费的速率，也即修改各自线程中的沉睡时间，这样我们就能看见生产者被阻塞，或者消费者被阻塞。

public class Test {static BlockQ<Integer> queue = new BlockQ<>(4);public static void main(String[] args) throws InterruptedException {Thread t1 = new Thread(new ProducerThread());Thread t2 = new Thread(new ConsumerThread());t1.start();Thread.sleep(7000);t2.start();}
}/*** 生产者*/
class ProducerThread implements Runnable{private int cnt = 0;@Overridepublic void run() {while (true) {try {Test.queue.addT(cnt ++);Thread.sleep(800);} catch (InterruptedException e) {e.printStackTrace();}}}
}/*** 消费者*/
class ConsumerThread implements Runnable{@Overridepublic void run() {while (true) {try {Integer i = Test.queue.getT();Thread.sleep(500);} catch (InterruptedException e) {e.printStackTrace();}}}
}

运行结果

---

运行的时候，这里等待了几秒才出现以下内容，跟自己在主线程设置的的沉睡时间有关

---