8. ArrayBlockingQueue

ArrayBlockingQueue类实现了BlockingQueue接口。阅读BlockingQueue文本以获取有关的更多信息。

ArrayBlockingQueue是一个有界的阻塞队列，它将元素存储在数组内部。有界意味着它无法存储无限量的元素，它可以同时存储的元素数量有一个上限。你需要在实例化时设置上限，之后无法更改，所以它和ArrayList有些区别，不要因为它们的名称相似而将它们的功能混杂。

ArrayBlockingQueue内部是以FIFO（先入先出）次序来存储元素的。队列的头部是在队列中存活时间最长的元素，而队列的尾部是在队列中存活时间最短的元素。

以下是实例化和使用ArrayBlockingQueue的例子：

BlockingQueue queue = new ArrayBlockingQueue(1024);

queue.put("1");

Object object = queue.take();

这是一个使用Java 泛型的BlockingQueue例子：

BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1024);

queue.put("1");

String string = queue.take();

源码

ArrayBlockingQueue中使用了这几个成员变量来保证操作，其实内部使用了一个循环数组，其中takeIndex和putIndex其实相当于队列的头部和尾部。

    /** 使用数组保存元素 */
    final Object[] items;

    /** 下一个take，poll，peek或remove方法调用时访问此下标的元素 */
    int takeIndex;

    /** 下一个put, offer, 或add方法调用时访问此下标的元素 */
    int putIndex;

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

    /** 保护所有操作的主锁 */
    final ReentrantLock lock;

    /** 获取元素的等待条件 */
    private final Condition notEmpty;

    /** 放置元素的等待条件 */
    private final Condition notFull;

构造函数如下：

/**
 * 使用一个固定的数值和默认的访问规则创建，默认是使用非公平锁
 *
 * @param capacity the capacity of this queue
 * @throws IllegalArgumentException if {@code capacity < 1}
 */
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}

/**
 * 使用一个固定的数值和指定的访问规则创建
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 *        on insertion or removal, are processed in FIFO order;
 *        if {@code false} the access order is unspecified.
 * @throws IllegalArgumentException if {@code capacity < 1}
 */
public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

/**
 *使用一个固定的数值和指定的访问规则创建，并将给定集合中的元素
 * 增加到队列中，增加的顺序是指定的集合迭代器的遍历顺序
 *
 * @param capacity the capacity of this queue
 * @param fair if {@code true} then queue accesses for threads blocked
 *        on insertion or removal, are processed in FIFO order;
 *        if {@code false} the access order is unspecified.
 * @param c the collection of elements to initially contain
 * @throws IllegalArgumentException if {@code capacity} is less than
 *         {@code c.size()}, or less than 1.
 * @throws NullPointerException if the specified collection or any
 *         of its elements are null
 */
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 {
        final Object[] items = this.items;
        int i = 0;
        try {
            for (E e : c)
                items[i++] = Objects.requireNonNull(e);
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;
    } finally {
        lock.unlock();
    }
}

增加操作

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

public boolean add(E e) {
    // 内部重用offer方法
    if (offer(e))
        return true;
    // 如果增加失败，抛出异常指示队列已满
    else
        throw new IllegalStateException("Queue full");
}

-------------------------------------------------------------------------

public boolean offer(E e) {
    // 检查是否是否为null，如果是抛出NPE异常
    Objects.requireNonNull(e);
    // 加锁。  此处使用final的原因是将成员变量赋值为局部变量，
    // 然后使用此变量就不需要经过两次访问，即先访问this，再
    // 访问lock，轻微提升程序性能，后面此种方法的使用也是一样。
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 如果队列满了，返回false
        if (count == items.length)
            return false;
        // 否则，加入队列
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

public static <T> T requireNonNull(T obj) {
    if (obj == null)
        throw new NullPointerException();
    return obj;
}

private void enqueue(E e) {
    // assert lock.isHeldByCurrentThread();
    // assert lock.getHoldCount() == 1;
    // assert items[putIndex] == null;

    final Object[] items = this.items;
    // 插入元素
    items[putIndex] = e;
    if (++putIndex == items.length) putIndex = 0;
    count++;
    // 随机通知一个等待的线程
    notEmpty.signal();
}

-------------------------------------------------------------------------

// 阻塞方法
public void put(E e) throws InterruptedException {
    Objects.requireNonNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        // 如果队列已经，在notFull上阻塞自己等待通知
        // 关于等待-通知机制已经说过很多次，此处不再多说
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

-------------------------------------------------------------------------

// 超时方法
public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {

    Objects.requireNonNull(e);
    // 计算超时时间，转换为纳秒
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        // 如果队列已满，超时等待，如果时间用完，返回false
        while (count == items.length) {
            if (nanos <= 0L)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}

删除操作

// 删除指定元素
public boolean remove(Object o) {
    if (o == null) return false;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 如果队列中存在元素
        if (count > 0) {
            final Object[] items = this.items;
            // 注意此处精彩的循环使用，因为内部是一个循环数组
            for (int i = takeIndex, end = putIndex,
                     to = (i < end) ? end : items.length;
                 ; i = 0, to = end) {
                for (; i < to; i++)
                    if (o.equals(items[i])) {
                        removeAt(i);
                        return true;
                    }
                if (to == end) break;
            }
        }
        return false;
    } finally {
        lock.unlock();
    }
}

void removeAt(final int removeIndex) {
    // assert lock.isHeldByCurrentThread();
    // 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
        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.
        for (int i = removeIndex, putIndex = this.putIndex;;) {
            int pred = i;
            if (++i == items.length) i = 0;
            // 如果已经移到了最后一个元素，跳出循环
            if (i == putIndex) {
                items[pred] = null;
                this.putIndex = pred;
                break;
            }
            // 将元素前移一位
            items[pred] = items[i];
        }
        count--;
        if (itrs != null)
            itrs.removedAt(removeIndex);
    }
    notFull.signal();
}

-------------------------------------------------------------------------

public E remove() {
    // 重用poll方法，如果队列为空，抛出异常
    E x = poll();
    if (x != null)
        return x;
    else
        throw new NoSuchElementException();
}

-------------------------------------------------------------------------

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

private E dequeue() {
    // assert lock.isHeldByCurrentThread();
    // assert lock.getHoldCount() == 1;
    // assert items[takeIndex] != null;

    final Object[] items = this.items;
    // 获取头元素，因为使用Object[]保存，所以要进行类型转换
    // 因为只能增加指定类型的元素，所以可以确保类型转换一定
    // 会成功，抑制此非受检警告
    @SuppressWarnings("unchecked")
    E e = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length) takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    notFull.signal();
    return e;
}

-------------------------------------------------------------------------

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 <= 0L)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}

阻塞方法以及超时方法和增加操作一样，此处不多做讲解。

访问操作

// element()方法在AbstractQueue<E>类中，ArrayBlockingQueue继承自此类
public E element() {
    // 重用peek方法，如果队列为空抛出异常
    E x = peek();
    if (x != null)
        return x;
    else
        throw new NoSuchElementException();
}

-------------------------------------------------------------------------

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];
}

辅助方法

部分方法逻辑简单，有兴趣自己查看即可。

public void clear() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int k;
        // 如果队列中存在元素，清空队列
        if ((k = count) > 0) {
            circularClear(items, takeIndex, putIndex);
            takeIndex = putIndex;
            count = 0;
            // 使迭代器保持一致
            if (itrs != null)
                itrs.queueIsEmpty();
            // 如果有线程等待插入元素，唤醒
            for (; k > 0 && lock.hasWaiters(notFull); k--)
                notFull.signal();
        }
    } finally {
        lock.unlock();
    }
}

// 将存在的元素全部置为null即可，等待 gc回收它们，此时等于清空了队列。
private static void circularClear(Object[] items, int i, int end) {
    // assert 0 <= i && i < items.length;
    // assert 0 <= end && end < items.length;

    for (int to = (i < end) ? end : items.length;
         ; i = 0, to = end) {
        for (; i < to; i++) items[i] = null;
        if (to == end) break;
    }
}

-------------------------------------------------------------------------

public int drainTo(Collection<? super E> c) {
    // 重用drainTo(Collection<? super E> c, int maxElements)方法
    return drainTo(c, Integer.MAX_VALUE);
}

public int drainTo(Collection<? super E> c, int maxElements) {
    Objects.requireNonNull(c);
    // 如果指定的集合是自己，抛出异常，符合BlockingQueue接口文档中的定义
    if (c == this)
        throw new IllegalArgumentException();
    if (maxElements <= 0)
        return 0;
    final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 获取需要转移的元素数量
        int n = Math.min(maxElements, count);
        int take = takeIndex;
        int i = 0;
        try {
            // 通过直接访问数组，比重复调用poll()方法再增加性能会高很多
            while (i < n) {
                @SuppressWarnings("unchecked")
                E e = (E) items[take];
                c.add(e);
                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();
    }
}

核心要点

  1. 内部使用了一个循环数组
  2. 是一个有界数组，提供了容量后无法被更改
  3. 可以指定锁的公平性
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 216,744评论 6赞 502
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 92,505评论 3赞 392
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 163,105评论 0赞 353
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,242评论 1赞 292
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,269评论 6赞 389
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,215评论 1赞 299
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,096评论 3赞 418
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 38,939评论 0赞 274
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,354评论 1赞 311
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,573评论 2赞 333
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 39,745评论 1赞 348
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,448评论 5赞 344
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,048评论 3赞 327
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 31,683评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,838评论 1赞 269
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 47,776评论 2赞 369
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 44,652评论 2赞 354

推荐阅读更多精彩内容

  • 我想要的人生
    很多时候，人们都会想象自己想要过的人生，可是每个人的人生都是什么呢？
    三寸小时光阅读 128评论 0赞 0
  • 2017-06-09  安住当下
    小裴：艺术 你的价值就在你身后的那个金色的圆盘之上，就在那炼金的火炉里，此刻你正拿着火把和圣杯，勾兑出硫酸来溶解死...
    bluespirit_0阅读 259评论 0赞 0
  • 石碑
    我们不需要石碑的时候，我们就视而不见。 我不相信，石碑也不需要我们。 因为，此刻我看见他了，我感动了，我落泪了。
    愚悦_2814阅读 134评论 0赞 0