java并发面试常识之ArrayBlockingQueue

>>强大,10k+点赞的 SpringBoot 后台管理系统竟然出了详细教程!

ArrayBlockingQueue是常用的线程集合,在线程池中也常常被当做任务队列来使用。使用频率特别高。他是维护的是一个循环队列(基于数组实现),循环结构在数据结构中比较常见,但是在源码实现中还是比较少见的。

线程安全的实现

线程安全队列,基本是离不开锁的。ArrayBlockingQueue使用的是ReentrantLock,配合两种Condition,实现了集合的线程安全操作。这里稍微说一个好习惯,下面是成员变量的声明。

    private static final long serialVersionUID = -817911632652898426L;
   final Object[] items;
   int takeIndex;
   int putIndex;
   int count;
   final ReentrantLock lock;
   private final Condition notEmpty;
   private final Condition notFull;
   transient Itrs itrs = null;

赋值的操作基本都是在构造函数里做的。这样有个好处,代码执行可控。成员变量的初始化也是会合并在构造方法里执行的,但是在执行顺序上需要好好斟酌,如果写在构造方法里初始化,则没有相关问题。

阻塞队列的常用场所就是生产者消费者。一般都是生产者放入,消费者从头取数据。下面重点说这两个操作。
这两个操作都是依靠锁来保证线程安全的。

生产操作

    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等放入操作,首先是获取锁,如果发现数据满了,就通过notFull的condition,来阻塞线程。这里的条件判定一定是用while而不是if,多线程情况下,可以被唤醒后发现又满了。

    private void enqueue(E x) {
       final Object[] items = this.items;
       items[putIndex] = x;
       if (++putIndex == items.length)
           putIndex = 0;
       count++;
       notEmpty.signal();
   }

这个是入队列的操作。首先获取维护的数组。putindex就是放入操作的标志。这个操作会一直加。达到预定的长度后就变成0从头开始计数。这样插入的操作就是一个循环的操作了,count就是用来做计数的,作为能否插入数据的一个标准,插入数据后就通过notEmpty的condition发出一个信号唤醒消费线程。

消费操作

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

消费的方法也是这样。先获取锁,然后进行条件判断,如果没有数据,则阻塞线程。注意点和put一样。

 private E dequeue() {
       final Object[] items = this.items;
       @SuppressWarnings("unchecked")
       E x = (E) items[takeIndex];
       items[takeIndex] = null;
       if (++takeIndex == items.length)
           takeIndex = 0;
       count--;
       if (itrs != null)
           itrs.elementDequeued();
       notFull.signal();
       return x;
   }

取数据的时候,也依靠takeIndex,这是一个标志,这个数值也会一直增加,表示取的第一个数据的位置。如果这个标志走到最后,然后变成0,从头再来。这样保证取出的数据都是fifo的顺序。删除的时候如果发现迭代中,则会修改迭代器的遍历。然后通过notFull的condition来唤醒生产线程。

移除操作

    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();
       }
   }

对于remove操作就比较麻烦了,首先获取锁之后,把两个标志位本地化,然后找到要删除的元素的位置。调用removeAt,这里删除需要对标志位做改变。

    void removeAt(final int removeIndex) {
       final Object[] items = this.items;
       if (removeIndex == takeIndex) {
           items[takeIndex] = null;
           if (++takeIndex == items.length)
               takeIndex = 0;
           count--;
           if (itrs != null)
               itrs.elementDequeued();
       } else {
           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();
   }

如果删除的元素是位置和takeindex一样。那就可以直接删除,然后让删除标志位向后移动。如果不是,则从删除的位置开始,进行后面向前面的数据覆盖的操作。直到遇到putindex的前一个位置。然后把那个位置的数据设置为null。并且把putindex的位置往前移动一格,正在迭代的时候要删除数据并且唤醒生产线程。


原文始发于微信公众号(一次编译多处运行):java并发面试常识之ArrayBlockingQueue