ArrayBlockingQueue

ArrayBlockingQueue是一个有界的阻塞队列,其内部实现是将对象放到一个数组里。 有界也就意味着,它不能够存储无限多数量的元素。它有一个同一时间能够存储元素数量的上限。 你可以在对其初始化的时候设定这个上限,但之后就无法对这个上限进行修改了 (因为它是基于数组实现的,也就具有数组的特性:一旦初始化,大小就无法修改)。

源码分析

初始化必须指定容量 可选是否为公平锁或者通过一个现有集合(Collection)初始化

  • ArrayBlockingQueue(int capacity)
  • ArrayBlockingQueue(int capacity, boolean fair)
  • ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)

属性

// 存储队列元素的数组,是个循环数组
final Object[] items;

// 拿数据的索引,用于take,poll,peek,remove方法
int takeIndex;

// 放数据的索引,用于put,offer,add方法
int putIndex;

// 元素个数
int count;

// 可重入锁
final ReentrantLock lock;
// notEmpty条件对象,由lock创建
private final Condition notEmpty;
// notFull条件对象,由lock创建
private final Condition notFull;

方法

添加元素

  • boolean offer(E e)
    public boolean offer(E e) {
      checkNotNull(e); // null元素抛出NullPointerException
      final ReentrantLock lock = this.lock;
      lock.lock(); // 加锁
      try {
          if (count == items.length)
              return false; // 队列满了返回false
          else {
              enqueue(e); // 入队
              return true;
          }
      } finally {
          lock.unlock(); // 释放锁
      }
    }
    private void enqueue(E x) {
      // assert lock.getHoldCount() == 1;
      // assert items[putIndex] == null;
      final Object[] items = this.items;
      items[putIndex] = x;
      if (++putIndex == items.length) // 如果put索引满了 移动到0
          putIndex = 0;
      count++; // 队列元素个数增加
      // 使用条件对象notEmpty通知,比如使用take方法的时候队列里没有数据,被阻塞。
      // 这个时候队列insert了一条数据,需要调用signal进行通知
      notEmpty.signal(); 
    }
    
  • boolean offer(E e, long timeout, TimeUnit unit)
    public boolean offer(E e, long timeout, TimeUnit unit)
      throws InterruptedException {
      checkNotNull(e);
      long nanos = unit.toNanos(timeout); // 将TimeUnit转为纳秒
      final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();
      try {
          while (count == items.length) { // 若果队列满了
              if (nanos <= 0) // 等待时间已经完成直接返回false
                  return false;
              // notFull.awaitNanos(nanos)返回未等待纳秒时间   
              // cpu时间片可能没有timeout这么长
              nanos = notFull.awaitNanos(nanos);
          }
          enqueue(e);
          return true;
      } finally {
          lock.unlock();
      }
    }
    
  • boolean add(E e)
    public boolean add(E e) {
      if (offer(e)) // 队列未满 直接添加元素 见上offer方法
          return true;
      else // 否则抛出队列满的异常
          throw new IllegalStateException("Queue full");
    }
    
  • void put(E e)
    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();
      }
    }
    

移除元素

  • E poll()
    public E poll() {
      final ReentrantLock lock = this.lock;
      lock.lock();
      try {
          // 队列空返回null 否者进行出队操作
          return (count == 0) ? null : dequeue();
      } finally {
          lock.unlock();
      }
    }
    private E dequeue() {
      // assert lock.getHoldCount() == 1;
      // assert items[takeIndex] != null;
      final Object[] items = this.items;
      // 删除take索引位置的元素
      E x = (E) items[takeIndex];
      items[takeIndex] = null;
      // take索引后移 并判断是否需要移动到起始位置
      if (++takeIndex == items.length)
          takeIndex = 0;
      count--;
      // 重置迭代器如果take索引回到起始位置 
      // 如果队列为空 将迭代器置为null
      if (itrs != null) 
          itrs.elementDequeued();
      // 使用条件对象notFull通知,比如使用put方法放数据的时候队列已满,被阻塞。
      // 这个时候消费了一条数据,队列没满了,就需要调用signal进行通知
      notFull.signal();
      return x;
    }
    
  • E poll(long timeout, TimeUnit unit)
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
      long nanos = unit.toNanos(timeout);
      final ReentrantLock lock = this.lock;
      // 如果当前线程被中断 抛出InterruptedException
      lock.lockInterruptibly();
      try {
          while (count == 0) {
              if (nanos <= 0)
                  return null;
              // notEmpty.awaitNanos(nanos)返回未等待纳秒时间   
              // cpu时间片可能没有timeout这么长
              nanos = notEmpty.awaitNanos(nanos);
          }
          return dequeue();
      } finally {
          lock.unlock();
      }
    }
    
  • E remove()
     public E remove() {
      E x = poll(); // 见poll方法
      if (x != null)
          return x;
      else
          throw new NoSuchElementException();
    }
    
  • boolean remove(Object o)
    // 删除一个指定元素
    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) {
              // put索引
              final int putIndex = this.putIndex;
              // take索引
              int i = takeIndex;
              // 从take索引遍历到put索引 如果找到删除指定索引位置元素
              // 否则 返回false
              do {
                  if (o.equals(items[i])) {
                      removeAt(i);
                      return true;
                  }
                  if (++i == items.length)
                      i = 0;
              } while (i != putIndex);
          }
          return false;
      } finally {
          lock.unlock();
      }
    }
    void removeAt(final int removeIndex) {
      // assert lock.getHoldCount() == 1;
      // assert items[removeIndex] != null;
      // assert removeIndex >= 0 && removeIndex < items.length;
      final Object[] items = this.items;
      // 若果删除索引跟take索引相同 则直接删除元素
      // 并更新take索引
      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.
          final int putIndex = this.putIndex;
          // 将删除索引后的元素向前移动1移除最后一个元素更新put索引
          for (int i = removeIndex;;) {
              int next = i + 1;
              if (next == items.length) // 循环数组
                  next = 0;
              if (next != putIndex) { // 索引递增并向前移动元素
                  items[i] = items[next];
                  i = next; 
              } else { // 删除当前put索引的元素 并更新put索引-1
                  items[i] = null;
                  this.putIndex = i;
                  break;
              }
          }
          count--;
          if (itrs != null)
              itrs.removedAt(removeIndex);
      }
      notFull.signal();
    }
    
  • E take()
    public E take() throws InterruptedException {
      final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();
      try {
          while (count == 0)
              notEmpty.await(); // 如果队列为空 阻塞到队列非空
          return dequeue(); // 出队操作 见E poll()
      } finally {
          lock.unlock();
      }
    }
    

    总结

    ArrayBlockingQueue底层是一个循环的数组,使用一个重入锁和这个锁生成的两个条件对象进行并发控制。 其他方法都会加锁比如contains、toArray、drainTo、clear、toString。