阻塞队列

阻塞队列出现原因

1、很多场景要求分离生产者、消费者两个角色，题目得由不同的线程来担当，而之前的实现根本没有考虑多线程的情况
2、队列为空，那么在之前的实现里面会返回null，如果就是硬要拿到一个元素呢？原来的方式只能不断循环尝试寻找
3、队列为满，那么再之前的实现里面会返回false，如果就是硬要塞入一个元素呢？原来的方式只能不断循环尝试寻找

解决方法：
1、用锁保证线程安全
2、用条件变量让poll或offer线程进入等待状态，而不是不断循环尝试，让CPU空转

核心问题：
多线程导致添加队列时线程错乱，可能会覆盖值或者和预期的不对

单锁实现阻塞队列

public class BlockingQueue1<E> implements BlockingQueue<E> {
    
    private final E[] array;
    private int head;
    private int tail;
    private int size;

    @SuppressWarnings("all")
    public BlockingQueue1(int capacity) {
        array = (E[]) new Object[capacity];
    }

    //加锁对象
    private ReentrantLock lock = new ReentrantLock();
    private Condition headWaits = lock.newCondition();
    private Condition tailWaits = lock.newCondition();

    private boolean isEmpty() {
        return size == 0;
    }

    private boolean isFull() {
        return size == array.length;
    }

    @Override
    public void offer(E e) throws InterruptedException {
        lock.lockInterruptibly();
        try{
            while(isFull()) {
                //线程等待
                tailWaits.await();
            }
            //出循环就表示可以添加元素了
            array[tail] = e;
            if(++tail == array.length) {
                tail = 0;
            }
            size++;
            //添加元素成功队列非空，这个时候poll方法就不用等待，让它恢复执行
            headWaits.signal();
        } finally {
            lock.unlock();
        }

    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        lock.lockInterruptibly();
        try{
            //毫秒转纳秒
            long nanos = TimeUnit.MILLISECONDS.toNanos(timeout);
            while(isFull()) {
                //线程等待,且有时间上限的等待
                if(nanos <= 0) {
                    return false;
                }
                nanos = tailWaits.awaitNanos(nanos);// 最多等待多少纳秒 返回值:代表剩余时间
            }
            //出循环就表示可以添加元素了
            array[tail] = e;
            if(++tail == array.length) {
                tail = 0;
            }
            size++;
            //添加元素成功队列非空，这个时候poll方法就不用等待，让它恢复执行
            headWaits.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while(isEmpty()){
                //线程等待
                headWaits.await();
            }
            //出循环表示队列不空了，添加元素
            E e = array[head];
            array[head] = null; // 垃圾回收
            if(++head == array.length) {
                head = 0;
            }
            size--;
            //取走元素了队列不满，这个时候offer方法可以被唤醒
            tailWaits.signal();
            return e;
        } finally {
            lock.unlock();
        }
    }
}

双锁实现阻塞队列(完整版)

public class BlockingQueue2<E> implements BlockingQueue<E> {

    private final E[] array;
    private int head;
    private int tail;
    //改成原子型整数防止offer和poll方法多锁情况下的size交错情况
    private AtomicInteger size;

    @SuppressWarnings("all")
    public BlockingQueue2(int capacity) {
        array = (E[]) new Object[capacity];
    }

    //双锁，一把在offer上一把在poll用
    private ReentrantLock tailLock = new ReentrantLock();
    private Condition tailWaits = tailLock.newCondition();

    private ReentrantLock headLock = new ReentrantLock();
    private Condition headWaits = tailLock.newCondition();

    private boolean isEmpty() {
        return size.get() == 0;
    }

    private boolean isFull() {
        return size.get() == array.length;
    }

    @Override
    public void offer(E e) throws InterruptedException {
        tailLock.lockInterruptibly();
        int c; // 新增元素前的元素个数
        try {
            //1、队列满则等待
            while(isFull()) {
                tailWaits.await();
            }

            //2、不满则入队
            array[tail] = e;
            if(++tail == array.length) {
                tail = 0;
            }
            //3、修改size
            //原子整型自增
            c = size.getAndIncrement();// size++
            if(c + 1 < array.length) {
                //自己唤醒
                tailWaits.signal();
            }
            /*
                1、读取成员变量size的值
                2、自增
                3、结果写回成员变量size
             */

        } finally {
            tailLock.unlock();
        }
        //阻塞要和锁配对使用，但这里不配对不能直接使用
        //唤醒等待非空的poll线程
        if(c == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }

    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        return true;
    }

    @Override
    public E poll() throws InterruptedException {
        E e;
        int c;// 取走前的元素个数
        headLock.lockInterruptibly();
        try {
            //1、队列空则等待
            while(isEmpty()) {
                headWaits.await();
            }
            //出循环表示队列不空了，添加元素
            e = array[head];
            array[head] = null; // 垃圾回收
            if(++head == array.length) {
                head = 0;
            }
            c = size.getAndDecrement();// size--
            /*
                1、读取成员变量size的值
                2、自减
                3、结果写回成员变量size
             */
            if(c > 1) {
                //自己唤醒
                headWaits.signal();
            }

        }finally {
            headLock.unlock();
        }
        //阻塞要和锁配对使用，但这里不配对不能使用
        //不产生死锁：平级而非嵌套结构
        //队列从满 -> 不满时，由poll唤醒等待不满的offer线程
        if(c == array.length) {
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }
}