ArrayBlockingQueue 数组阻塞队列源码阅读

1. 概述

数组阻塞队列

有界的阻塞数组, 容量一旦创建, 无法修改
阻塞队列, 队列满的时候, 往队列put数据会被阻塞, 队列空, 取数据也会被阻塞
并发安全

2. 数据结构

/** 存储队列元素的数组 */
/** 存储队列元素的数组 */
final Object[] items;

/** 队首位置，下一次 take, poll, peek, remove 方法在 items 中的位置  */
int takeIndex;

/** 队末位置，下一次 put, offer, add 方法在 items 中的位置 */
int putIndex;

/** 队列中元素的数量 */
int count;

// 锁 保证队列的并发安全性
final ReentrantLock lock;

/** 队列不为空的监视器 */
private final Condition notEmpty;

/** 不满（队列元素数量小于items.size()）的监视器 */
private final Condition notFull;

3. 初始化构造函数

    public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        // 数组
        this.items = new Object[capacity];
        // 队列公平不公平指的是用的ReentrantLock
        lock = new ReentrantLock(fair);
        // 创建队列不为空，不满的监视器
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    public ArrayBlockingQueue(int capacity, boolean fair,
                              Collection<? extends E> c) {
        this(capacity, fair);

        final ReentrantLock lock = this.lock;

        // 加锁
        lock.lock(); // Lock only for visibility, not mutual exclusion
        try {
            int i = 0;
            try {
                // 集合元素一个个入队
                for (E e : c) {
                    checkNotNull(e);
                    items[i++] = e;
                }
            } catch (ArrayIndexOutOfBoundsException ex) {
                throw new IllegalArgumentException();
            }
            count = i;
            putIndex = (i == capacity) ? 0 : i;
        } finally {
            lock.unlock();
        }
    }

4. 方法

4.1 核心方法

核心方法入队，出队，移除元素，调用这些方法都加锁的。

    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        // 数组满了，putIndex重新从0开始
        if (++putIndex == items.length)
            putIndex = 0;
        // 添加元素count+1
        count++;
        // 唤醒队列不为空的wait
        notEmpty.signal();
    }

    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        // 添加元素count-1
        count--;
        // 如果迭代器不为空维护一下
        if (itrs != null)
            itrs.elementDequeued();
        // 唤醒队列不满的wait
        notFull.signal();
        return x;
    }

   void removeAt(final int removeIndex) {
        // assert lock.getHoldCount() == 1;
        // assert items[removeIndex] != null;
        // assert removeIndex >= 0 && removeIndex < items.length;
        final Object[] items = this.items;
        if (removeIndex == takeIndex) {
            // 如果要移除的是就是要出队的元素
            // removing front item; just advance
            // 元素设置为null
            items[takeIndex] = null;
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;
            if (itrs != null)
                itrs.elementDequeued();
        } else {
            // an "interior" remove
            // slide over all others up through putIndex.
            // 用removeIndex后面的元素，向前移动一位
            final int putIndex = this.putIndex;
            for (int i = removeIndex;;) {
                int next = i + 1;
                if (next == items.length)
                    next = 0;
                if (next != putIndex) {
                    items[i] = items[next];
                    i = next;
                } else {
                    items[i] = null;
                    this.putIndex = i;
                    break;
                }
            }
            count--;
            if (itrs != null)
                itrs.removedAt(removeIndex);
        }
        notFull.signal();
    }

4.2 常用外部调用的方法

offer()，入队，队列满了直接返回 false，不会阻塞，入队成功返回true

   public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }

add，其实调用了 offer

    public boolean add(E e) {
        return super.add(e);
    }

    // 父类
    public boolean add(E e) {
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }

put lock可以被中断，队列满了会阻塞， notFull.await()，等队列元素出队notFull.signal() 唤醒

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
        notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

在put基础上，等待timeout个unit时间

public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException {

    checkNotNull(e);
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length) {
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}

取出元素，列表为空直接返回null，不阻塞

public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}

取元素，队列空阻塞等待

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
        notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}

取元素，队列空等待参数指定的时间

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0) {
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}

查看将要取出的元素，元素不出队

public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return itemAt(takeIndex); // null when queue is empty
    } finally {
        lock.unlock();
    }
}

final E itemAt(int i) {
    return (E) items[i];
}

移除某个元素，移除找到的第一个这个元素，并返回true

    public boolean remove(Object o) {
        if (o == null) return false;
        final Object[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count > 0) {
                final int putIndex = this.putIndex;
                // 遍历
                int i = takeIndex;
                do {
                    if (o.equals(items[i])) {
                        // 找到了移除对应元素
                        removeAt(i);
                        return true;
                    } 
                    if (++i == items.length)
                        i = 0;
                } while (i != putIndex);
            }
            return false;
        } finally {
            lock.unlock();
        }
    }

  public boolean contains(Object o) {
        if (o == null) return false;
        final Object[] items = this.items;
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count > 0) {
                final int putIndex = this.putIndex;
                int i = takeIndex;
                do {
                    if (o.equals(items[i]))
                        // 找到了返回true
                        return true;
                    if (++i == items.length)
                        i = 0;
                } while (i != putIndex);
            }
            return false;
        } finally {
            lock.unlock();
        }
    }

清理队列

public void clear() {
    final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int k = count;
        if (k > 0) {
            final int putIndex = this.putIndex;
            int i = takeIndex;
            do {
                items[i] = null;
                if (++i == items.length)
                    i = 0;
            } while (i != putIndex);
            takeIndex = putIndex;
            count = 0;
            if (itrs != null)
                itrs.queueIsEmpty();
            for (; k > 0 && lock.hasWaiters(notFull); k--)
            notFull.signal();
        }
    } finally {
        lock.unlock();
    }
}

将队列中的元素移动到c集合中

public int drainTo(Collection<? super E> c) {
    return drainTo(c, Integer.MAX_VALUE);
}

/**
 * @throws UnsupportedOperationException {@inheritDoc}
 * @throws ClassCastException            {@inheritDoc}
 * @throws NullPointerException          {@inheritDoc}
 * @throws IllegalArgumentException      {@inheritDoc}
 */
public int drainTo(Collection<? super E> c, int maxElements) {
    checkNotNull(c);
    if (c == this)
        throw new IllegalArgumentException();
    if (maxElements <= 0)
        return 0;
    final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // n取 maxElements 和 集合元素的最小值
        int n = Math.min(maxElements, count);
        int take = takeIndex;
        int i = 0;
        try {
            while (i < n) {
                @SuppressWarnings("unchecked")
                E x = (E) items[take];
                c.add(x);
                items[take] = null;
                if (++take == items.length)
                    take = 0;
                i++;
            }
            return n;
        } finally {
            // Restore invariants even if c.add() threw
            if (i > 0) {
                count -= i;
                // 修改队首位置
                takeIndex = take;
                if (itrs != null) {
                    if (count == 0)
                        itrs.queueIsEmpty();
                    else if (i > take)
                        itrs.takeIndexWrapped();
                }
                for (; i > 0 && lock.hasWaiters(notFull); i--)
                notFull.signal();
            }
        }
    } finally {
        lock.unlock();
    }
}

5. 单元测试


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;

public class ArrayBlockingQueueStudy {
    public static void main(String[] args) throws InterruptedException {
        ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
        queue.put("1");
        queue.put("2");

        // true
        System.out.println(queue.offer("3"));
        // false
        System.out.println(queue.offer("4"));

        // 抛出异常
        // Exception in thread "main" java.lang.IllegalStateException: Queue full
        // at java.util.AbstractQueue.add(AbstractQueue.java:98)
        // at java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:312)
        //  at arrayBlockingQueue.ArrayBlockingQueueStudy.main(ArrayBlockingQueueStudy.java:18)
        // System.out.println(queue.add("4"));

        // 阻塞
        // queue.put("4");

        // 等待1s 返回false
        System.out.println(queue.offer("4", 1, TimeUnit.SECONDS));

        System.out.println(queue.poll());
        System.out.println(queue.poll());
        System.out.println(queue.poll());
        // null
        System.out.println(queue.poll());

        // 阻塞
        // System.out.println(queue.take());

        // 等待1s 返回null
        System.out.println(queue.poll(1, TimeUnit.SECONDS));

        queue.put("1");
        // 1
        System.out.println(queue.peek());
        // 1
        System.out.println(queue.peek());

        queue.put("1");
        System.out.println(queue.remove("1"));
        // 1
        System.out.println(queue.peek());
        System.out.println(queue.remove("1"));
        // null
        System.out.println(queue.peek());

        queue.put("1");
        queue.put("1");
        queue.clear();
        // null
        System.out.println(queue.peek());

        List<String> c = new ArrayList<>();
        queue.put("1");
        queue.put("1");
        queue.drainTo(c);
        // 2 ["1","1"]
        System.out.println(c.size());
        // null
        System.out.println(queue.peek());
    }
}