手写一个属于自己的阻塞队列

Posted 2021-08-16 醉酒的小男人

咱们使用Condition和Lock实现一个自己的阻塞队列,容器使用List集合,Condition的await/signal、synchronized的await/notify/notifyall、LockSupport的park/unpark都是实现线程之间通信的,只是应用场景不一样。

什么是阻塞队列

队列是一种只允许在一端进行删除操作,在另一端进行插入操作的线性表，允许插入的一端称为队尾、允许删除的一端称为队头。

编码思路

主要有两个方法：

put:支持阻塞插入,队列满了的情况下,会阻塞继续往队列中添加数据的线程,直到队列元素被释放

take:支持阻塞移除,队列为空的情况下,会阻塞从队列中获取元素的线程,直到队列添加了新的元素

package com;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionBlockedQueueTest {
    //表示堵塞队列中的容器
    private List<String> items;
    //集合中已添加元素个数
    private volatile int size;
    //集合的容量
    private volatile int count;
    private Lock lock = new ReentrantLock();
    //让take取值方法堵塞
    private final Condition notEmpty = lock.newCondition();
    //让put添加方法堵塞
    private final Condition notFull = lock.newCondition();

    public ConditionBlockedQueueTest(int count) {
        this.count = count;
        items = new ArrayList<>(count);
    }

    /**
     * 阻塞方式添加元素
     */
    public void put(String e) throws InterruptedException {
        lock.lock();
        try {
            if(items.size() >= count){
                System.out.println("堵塞队列满了,需要等待");
                notFull.await();
            }
            ++size;//添加元素个数
            items.add(e);
            notEmpty.signal();
        }finally {
            lock.unlock();
        }
    }

    /**
     * 阻塞方法移除元素
     * @return
     * @throws InterruptedException
     */
    public String take() throws InterruptedException {
        lock.lock();
        try {
            if(items.isEmpty()){
                System.out.println("堵塞队列空了,需要等待添加元素");
                notEmpty.await();
            }
            --size;//减少元素个数
            String item = items.remove(0);
            notFull.signal();
            return item;
        }finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) throws Exception{
        ConditionBlockedQueueTest conditionBlockedQueueTest = new ConditionBlockedQueueTest(10);
        //生产者线程
        Thread t1 = new Thread(()->{
            for (int i=0; i< 100;i++) {
                try {
                    String val = "item-"+i;
                    conditionBlockedQueueTest.put(val);
                    System.out.println("生产一个元素:"+val);
                    Thread.sleep(new Random().nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t1.start();
        Thread.sleep(1000);
        //消费者线程
        Thread t2 = new Thread(()->{
            for (;;) {
                try {
                    String result = conditionBlockedQueueTest.take();
                    System.out.println("消费者线程消费元素:"+result);
                    Thread.sleep(new Random().nextInt(100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        t2.start();
    }
}

结果