前言
ArrayBlockingQueue以数组的方式实现了BlockingQueue接口。其是线程安全的，提供了offer, put,plll,take,offer(E,long,TimeUnit),poll(E,long,TimeUnit)方法。
源码分析
/** The queued items */
final Object[] items;
/** items index for next take, poll, peek or remove */
int takeIndex;
/** items index for next put, offer, or add */
int putIndex;
/** Number of elements in the queue */
int count;
/*
* Concurrency control uses the classic two-condition algorithm
* found in any textbook.
*/
/** Main lock guarding all access */
final ReentrantLock lock;
/** Condition for waiting takes */
private final Condition notEmpty;
/** Condition for waiting puts */
private final Condition notFull;
可以看到成员变量有一个数组Object[] ，ReentrantLock lock,Condition notEmpty, Condition notFull。
成员变量的作用
- lock ，因为ArrayBlockingQueue是线程安全的，需要一把锁。
- notEmpty 维护竞争锁的条件为ArrayBlockingQueue为非空的线程集合，当线程运行的条件为ArrayBlockingQueue为非空，但是此时条件为空时，线程会被挂起，并线程会被添加到notEmpty中的队列中。当条件变为非空时，可以通知里面线程队列，让其开始竞争锁。
- notFull 维护竞争锁的条件为ArrayBlockingQueue为有剩余空间的线程集合，当线程运行的条件为ArrayBlockingQueue为，但是此时条件为位有剩余空间时，线程会被挂起，并线程会被添加到未有剩余空间中的队列中。当条件变为有剩余空间时，可以通知里面线程队列，让其开始竞争锁。
方法实现分析
- offer(E e)
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();
}
}
将元素添加到队列中，并且整个过程时线程安全的。添加之后将通知到notEmpty队列。
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)
putIndex = 0;
count++;
notEmpty.signal();
}
- poll()方法
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
从队列中取出元素，线程安全，取出之后通知需要满足有存储空间的队列。
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;
}
- 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);
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(long timeout, TimeUnit unit)
尝试取出一个元素，直到指定的时间返回。
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();
}
}
总结
利用两个队列（Condition）挂在条件不同的线程，不同的条件唤醒，比同一个队列效率更高，减少了不必要的自旋锁。
