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

news2024/10/5 14:33:43

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;
    @Override
    public void run() {
        while (true) {
            try {
                Test.queue.addT(cnt ++);
                Thread.sleep(800);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

/**
 * 消费者
 */
class ConsumerThread implements Runnable{
    @Override
    public void run() {
        while (true) {
            try {
                Integer i = Test.queue.getT();
                Thread.sleep(500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}