介绍

是一个用数组实现的有界阻塞队列，此队列按照先进先出（FIFO）的原则对元素进行排序。支持公平锁和非公平锁。【注：每一个线程在获取锁的时候可能都会排队等待，如果在等待时间上，先获取锁的线程的请求一定先被满足，那么这个锁就是公平的。反之，这个锁就是不公平的。公平的获取锁，也就是当前等待时间最长的线程先获取锁】。

锁控制参考：ReentrantLock源码

数据结构

/**队列元素 */
final Object[] items;/** items index for next take, poll, peek or remove */
int takeIndex;/** items index for next put, offer, or add */
int putIndex;/** 以入队数量 */
int count;/** 锁控制 */
final ReentrantLock lock;/** Condition for waiting takes */
private final Condition notEmpty;/** Condition for waiting puts */
private final Condition notFull;/* iterator */transient Itrs itrs = null;

方法实现

offer,poll,peek

    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();}}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();}}public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {return (count == 0) ? null : 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();}}

put,take

    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();}}public E take() throws InterruptedException {final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (count == 0)notEmpty.await();return dequeue();} finally {lock.unlock();}}

enqueue，dequeue

    private void enqueue(E x) {// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;final Object[] items = this.items;//在put index 放置元素items[putIndex] = x;//如果putIndex +1 == items.length，则设置为0，循环数组。if (++putIndex == items.length)putIndex = 0;count++;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--;if (itrs != null)itrs.elementDequeued();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 advanceitems[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;//i位置值设为null，后续节点全部前移，直到putIndex，并且把putIndex -1;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();}

注意

数组是个循环数组，当下标为item.length时则为0。

参考

ReentrantLock源码
Java并发包--阻塞队列（BlockingQueue）

ArrayBlockingQueue源码相关推荐

面试官系统精讲Java源码及大厂真题 - 22 ArrayBlockingQueue 源码解析
22 ArrayBlockingQueue 源码解析耐心和恒心总会得到报酬的. 引导语本小节我们来介绍本章最后一个队列:ArrayBlockingQueue.按照字面翻译,中文叫做数组阻塞队列,从 ...
Android多线程之ArrayBlockingQueue源码解析
阻塞队列系列 Android多线程之LinkedBlockingQueue源码解析 Android多线程之SynchronousQueue源码解析 Andorid多线程之DelayQueue源码分析 ...
阻塞队列和ArrayBlockingQueue源码解析
什么是阻塞队列当队列中为空时,从队列总获取元素的操作将被阻塞,当队列满时,向队列中添加元素的操作将被阻塞.试图从空的阻塞队列中获取元素的线程将会被阻塞,知道其它的线程往队列中插入新的元素.同样,试图 ...
Java8 ArrayBlockingQueue 源码阅读
一.什么是 ArrayBlockingQueue ArrayBlockingQueue 是 GUC(java.util.concurrent) 包下的一个线程安全的阻塞队列,底层使用数组实现. 除了线 ...
java queue源码_Java高并发系列之ArrayBlockingQueue源码解析
JUC包下定义了一个接口:BlockingQueue.其实现类有ArrayBlockingQueue等.本文先来介绍一下ArrayBlockingQueue.从字面可以看出,ArrayBlocking ...
ArrayBlockingQueue源码解析(JDK1.8)
ArrayBlockingQueue是基于数组的先进先出的有界循环队列同样我们还是先上类的关系图从类的关系图中可以看出ArrayBlockingQueue继承一个抽象类和实现了两个接口,然后分别简 ...
Java8 ArrayBlockingQueue 源码解析
目录 1.定义 2.构造方法 3.add / offer / put 4.poll / take / peek 5.remove / clear /drainTo 6.iterator / Itr / ...
【JUC】JDK1.8源码分析之ArrayBlockingQueue（三）
一.前言在完成Map下的并发集合后,现在来分析ArrayBlockingQueue,ArrayBlockingQueue可以用作一个阻塞型队列,支持多任务并发操作,有了之前看源码的积累,再看Arra ...
阻塞队列（ArrayBlockingQueue）迭代器源码分析
文章目录为什么 ArrayBlockingQueue 迭代器复杂呢? 提出几个 ArrayBlockingQueue 迭代器的问题用于下面代码分析时进行思考 Itrs Itr 1.重要变量 2. 构 ...

ArrayBlockingQueue源码

介绍