java1.7集合源码阅读：ArrayBlockingQueue
2024-08-29 01:30:28
ArrayBlockingQueue是一个先进先出线程安全的队列，队列头部是进入队列时间最长的元素，队尾是进入队列时间最短的元素，同时队列的最大容量是固定的。
先看类定义：
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
......
}
ArrayBlockingQueue实现了BlockingQueue接口，是一个阻塞队列实现，内部通过数组进行数据存储：
/** 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;
对元素的增加操作，主要提供了三种使用场景：添加(队列已满则返回false)、限时等待添加，无限等待添加；
/**
* Inserts the specified element at the tail of this queue if it is
* possible to do so immediately without exceeding the queue's capacity,
* returning {@code true} upon success and throwing an
* {@code IllegalStateException} if this queue is full.
*
* @param e the element to add
* @return {@code true} (as specified by {@link Collection#add})
* @throws IllegalStateException if this queue is full
* @throws NullPointerException if the specified element is null
*/
public boolean add(E e) { //add 和offer(e) 是一样的
return super.add(e);
} /**
* Inserts the specified element at the tail of this queue if it is
* possible to do so immediately without exceeding the queue's capacity,
* returning {@code true} upon success and {@code false} if this queue
* is full. This method is generally preferable to method {@link #add},
* which can fail to insert an element only by throwing an exception.
*
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)
return false;
else {
insert(e);
return true;
}
} finally {
lock.unlock();
}
} /**
* Inserts the specified element at the tail of this queue, waiting
* for space to become available if the queue is full.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
insert(e);
} finally {
lock.unlock();
}
} /**
* Inserts the specified element at the tail of this queue, waiting
* up to the specified wait time for space to become available if
* the queue is full.
*
* @throws InterruptedException {@inheritDoc}
* @throws NullPointerException {@inheritDoc}
*/
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);
}
insert(e);
return true;
} finally {
lock.unlock();
}
}
对元素的获取同样提供了三种使用场景：添加(队列空则返回null)、限时等待获取，无限等待获取；
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return (count == 0) ? null : extract();
} finally {
lock.unlock();
}
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return extract();
} 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 extract();
} finally {
lock.unlock();
}
}
获取队列元素大小也是强一致性的：
/**
* Returns the number of elements in this queue.
*
* @return the number of elements in this queue
*/
public int size() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return count;
} finally {
lock.unlock();
}
}
元素的删除也是同步的,元素存在则删除返回true，如果元素不存在则返回false：
/**
* Removes a single instance of the specified element from this queue,
* if it is present. More formally, removes an element {@code e} such
* that {@code o.equals(e)}, if this queue contains one or more such
* elements.
* Returns {@code true} if this queue contained the specified element
* (or equivalently, if this queue changed as a result of the call).
*
* <p>Removal of interior elements in circular array based queues
* is an intrinsically slow and disruptive operation, so should
* be undertaken only in exceptional circumstances, ideally
* only when the queue is known not to be accessible by other
* threads.
*
* @param o element to be removed from this queue, if present
* @return {@code true} if this queue changed as a result of the call
*/
public boolean remove(Object o) {
if (o == null) return false;
final Object[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lock();
try {
for (int i = takeIndex, k = count; k > 0; i = inc(i), k--) {
if (o.equals(items[i])) {
removeAt(i);
return true;
}
}
return false;
} finally {
lock.unlock();
}
}
ArrayBlockingQueue的所有操作都是通过加锁进行同步的，代码也比较简单。
最新文章
- 基础总结之Activity
- indows server 2008 多用户远程桌面连接设置（验证有效
- C# 中的as和is小结
- MongoDB学习笔记—Linux下搭建MongoDB环境
- C#保存CookieContainer到文件
- UIScrollView循环滚动1
- Java集合之Map接口
- quartz简单实现
- ExtJs + Struts2 + JSON
- (转)syslog日志等级
- .进程&线程（&java.lang.Thread）详解
- 在C#中实现串口通信的方法
- 安装docker17.06.0版本报错和解决方法
- jq closet的使用，找到距离最近的一个父元素；
- A-论文一些好的句子
- load data妙用
- [转]Bing Maps Tile System 学习
- request-2高级用法
- Unix环境高级编程(十一)线程
- C#后台获取ajax传来的xml格式数据值