数据结构-阻塞队列

阻塞队列存在的合理性

• 提供一个多线程环境下安全的添加和获取元素的容器
• 为了能够保证添加和获取元素同时避免不必要的cpu空转引起的资源消耗，当没有元素或者容量已满时，阻塞对应线程

基于双锁的阻塞队列要点如下

• 引入锁机制，保证多线程环境下临界区代码安全执行
• 使用双锁，offer 和 poll 分别使用不同的锁 tailLock 和 headLock 来控制，两个操作不会去竞争同一把锁，提高性能
• 判断 offer 或 poll 的条件时，通过 while 循环来避免线程虚假唤醒问题，类似的还有双检锁机制等
• 引入超时参数时，每次接收 awaitNanos() 的参数以便刷新下次等待的超时时间
• size 属性作为共享资源且在不同操作（不同锁）中都有被访问，存在线程安全问题，使用 AtomicInteger 代替
• 由于使用了不同的锁，当 offer 或者 poll 操作发生时，应当唤醒相对应操作中等待的线程以便其继续执行相关操作，例如当 offer 操作完成后，应当调用 headWaits.signal() 方法唤醒等待执行 poll 操作的线程继续执行
• 唤醒对应操作中的线程，应当在当前操作锁释放后执行，否则会发生死锁，例如当前持有 tailLock，想要唤醒 headWaits 中的线程，那么就需要先持有 headLock；同时假设另一边已经持有 headLock，想要唤醒 tailWaits 中的线程，那么就需要先持有 tailLock，由此造成死锁的情况
• 为了更好的优化性能（每次唤醒需要先获取锁的开销），在 offer 或 poll 操作中，不再无条件的执行唤醒操作；假设满足唤醒条件，也只唤醒一次，后续唤醒操作交给已经被唤醒的线程来执行（级联唤醒）
• offer 操作中执行 headWaits.signal() 的条件：size.getAndIncrement() == 0 即队列中第一个元素入队时开始唤醒等待的 poll 操作；与之对应的 poll 操作自身级联唤醒的条件：size.getAndDecrement() > 1 即只要队列中还存在元素，就唤醒还在等待的线程执行 poll 操作
• poll 操作中执行 tailWaits.signal() 的条件：size.getAndDecrement() == capacity 即队列中第一个元素出队时，代表有剩余空间，开始唤醒等待的 offer 操作；与之对应的 offer 操作自身级联唤醒的条件：size.getAndIncrement() < capacity - 1 即只要队列还有剩余容量，就唤醒还在等待的线程执行 offer 操作

附源码如下

public interface Collection {
    int size();

    boolean isEmpty();
}

public interface BlockingQueue<E> extends Collection {

    /**
     * 入队阻塞
     * @param e 待入队的元素
     */
    void offer(E e) throws InterruptedException;

    /**
     * 入队阻塞可超时
     * @param e 待入队的元素
     */
    void offer(E e, long timeout) throws InterruptedException;

    /**
     * 出队阻塞
     * @return 当前头部元素
     */
    E poll() throws InterruptedException;

    /**
     * 出队阻塞可超时
     * @return 当前头部元素
     */
    E poll(long timeout) throws InterruptedException;
}

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class ArrayBlockingQueue<E> implements BlockingQueue<E> {
    private final E[] data;
    private int head;
    private int tail;
    private final AtomicInteger size = new AtomicInteger();

    private final ReentrantLock headLock = new ReentrantLock();
    private final ReentrantLock tailLock = new ReentrantLock();

    private final Condition headWaits = headLock.newCondition();
    private final Condition tailWaits = tailLock.newCondition();

    @SuppressWarnings("unchecked")
    public ArrayBlockingQueue(int capacity) {
        this.data = (E[]) new Object[capacity];
    }

    private boolean fulled() {
        return size.get() == data.length;
    }

    @Override
    public void offer(E e) throws InterruptedException {
        int s;
        tailLock.lockInterruptibly();
        try {
            while (fulled()) {
                tailWaits.await();
            }
            data[tail] = e;
            if (++tail == data.length) {
                tail = 0;
            }
            s = size.getAndIncrement();
            if (s < data.length - 1) {
                tailWaits.signal();
            }
        } finally {
            tailLock.unlock();
        }
        if (s == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public void offer(E e, long timeout) throws InterruptedException {
        long t = TimeUnit.MILLISECONDS.toNanos(timeout);
        int s;
        tailLock.lockInterruptibly();
        try {
            while (fulled()) {
                t = tailWaits.awaitNanos(t);
            }
            data[tail] = e;
            // reset tail
            if (++tail == data.length) {
                tail = 0;
            }
            s = size.getAndIncrement();
            if (s < data.length - 1) {
                tailWaits.signal();
            }
        } finally {
            tailLock.unlock();
        }
        if (s == 0) {
            headLock.lock();
            try {
                headWaits.signal();
            } finally {
                headLock.unlock();
            }
        }
    }

    @Override
    public E poll() throws InterruptedException {
        E e;
        int s;
        headLock.lockInterruptibly();
        try {
            while (isEmpty()) {
                headWaits.await();
            }
            e = data[head];
            data[head] = null; // help GC
            // reset head
            if (++head == data.length) {
                head = 0;
            }
            s = size.getAndDecrement();
            if (s > 1) {
                headWaits.signal();
            }
        } finally {
            headLock.unlock();
        }

        if (s == data.length) {
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    @Override
    public E poll(long timeout) throws InterruptedException {
        E e;
        int s;
        long t = TimeUnit.MILLISECONDS.toNanos(timeout);
        headLock.lockInterruptibly();
        try {
            while (isEmpty()) {
                t = headWaits.awaitNanos(t);
            }
            e = data[head];
            data[head] = null; // help GC
            // reset head
            if (++head == data.length) {
                head = 0;
            }
            s = size.getAndDecrement();
            if (s > 1) {
                headWaits.signal();
            }
        } finally {
            headLock.unlock();
        }

        if (s == data.length) {
            tailLock.lock();
            try {
                tailWaits.signal();
            } finally {
                tailLock.unlock();
            }
        }
        return e;
    }

    @Override
    public int size() {
        return size.get();
    }

    @Override
    public boolean isEmpty() {
        return size.get() == 0;
    }

    @Override
    public String toString() {
        return Arrays.toString(data);
    }
}