我们再来看ArrayBlockingQueue的源码，它的底层是基于Object[]数组的（数组是内存中一段连续的空间，所以插入数据会比较方便），所以肯定是有界队列。使用了一个ReentrantLock来保证线程安全（粒度较粗，会带来更多锁的竞争，不利于高并发）。
内部成员变量
items是object类型的数组。
putIndex是压入元素的指针。
takeIndex是取出元素的指针。
cout是queue里面元素的总和。
后面queue是不是满了，通过count和items.length二者相比较来判断的。
如果一直put元素，没有任何take元素的操作，那么++putIndex=item.length的时候putIndex会被重置为0，此时count==item.length，所以put操作会被阻塞。
同理如果put满了，take操作开始拿空了queue，那么++takeIndex=item.length的时候，count=0，takeIndex会被重置为0。
/** 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;
add
add方法是queue满的时候插入不进去会抛出IllegalStateException("Queue full")的异常，且add方法其实是调用的offer()方法，offer返回true表示add成功，否则抛出异常。
public boolean add(E e) {
return super.add(e);
}
public boolean add(E e) {
//add的本质是调用offer方法
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");
}
offer
offer方法是返回boolean值的
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
//如果count等于items数组的长度，表示queue已经满了，返回false
if (count == items.length)
return false;
else {
//调用enqueue()方法添加元素到queue
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;
//将元素放到putIndex的位置，然后putIndex往后移一位
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
//当有元素插入，则唤醒notEmpty等待队列
notEmpty.signal();
}
put
put在queue满的情况下，再添加元素会调用notFull.await()进入阻塞，等待take()方法取出元素后notFull.signal()。
且put使用的是lock.lockInterruptibly()是会响应中断的，即抛出InterruptedException
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
//lock过程中响应中断
lock.lockInterruptibly();
try {
//是while循环，避免过早唤醒以及虚假唤醒
while (count == items.length)
//如果queue满了，就阻塞
notFull.await();
//唤醒之后再将元素加入到queue里
enqueue(e);
} finally {
lock.unlock();
}
}
offer(time)
offer(time)是带返回值的，但内部调用的是awaitNanos(time)，所以会阻塞一定时间，即等待一段时间后自己醒来，如果queue还是满的，就返回false；如果不是满的，就将元素加入queue，并返回true。
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) {
//如果被唤醒之后发现queue还是满的，且等待时间为0，就直接return false
if (nanos <= 0)
return false;
nanos = notFull.awaitNanos(nanos);
}
//否则就将新元素加入queue，并返回true。
enqueue(e);
return true;
} finally {
lock.unlock();
}
}
remove
remove和add是一对的，所以实现也是类似，调用的是poll()方法，如果poll的返回值不是null，则返回该值；如果是null，则抛出NoSuchElementException。
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();
}
poll
poll方法有值就返回值（从queue里移除），没有值就返回null。
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : dequeue();
} finally {
lock.unlock();
}
}
//dequeue主要还是对items数组进行操作，最后唤醒notFull的等待队列
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;
}
peek
peek操作是返回queue里的头元素，即items数组里takeIndex位置的元素。
public E peek() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return itemAt(takeIndex); // null when queue is empty
} finally {
lock.unlock();
}
}
//返回items数组里takeIndex的值
final E itemAt(int i) {
return (E) items[i];
}
take
take对应的是put操作，take在queue为空的时候，会阻塞，也是会响应中断的。
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
//在循环中判断count是不是为0，如果是0，表示queue为空，notEmpty.await进入等待
notEmpty.await();
//唤醒之后就返回当前takeindex的值
return dequeue();
} finally {
lock.unlock();
}
}
poll(time)
poll(time)与offer(time)相对应，poll一段时间，如果queue还是为空就返回null值，否则返回takeIndex对应的数组里的值。
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();
}
}
网友评论