今天就聊聊这两种Queue,本文分为以下两个部分,用分割线分开:
- BlockingQueue
- ConcurrentLinkedQueue,非阻塞算法
首先来看看
BlockingQueue:
Queue是什么就不需要多说了吧,一句话:队列是先进先出。相对的,栈是后进先出。如果不熟悉的话先找本基础的数据结构的书看看吧。
BlockingQueue,顾名思义,“阻塞队列”:可以提供阻塞功能的队列。
首先,看看BlockingQueue提供的常用方法:
| Throws exception | Special value | Blocks | Times out |
Insert | add(e) | offer(e) | put(e) | offer(e, timeout, unit) |
Remove | remove() | poll() | take() | poll(timeout, unit) |
Examine | element() | peek() | not applicable | not applicable |
从上表可以很明显看出每个方法的作用,这个不用多说。我想说的是:
- add(e) remove() element() 方法不会阻塞线程。当不满足约束条件时,会抛出IllegalStateException 异常。例如:当队列被元素填满后,再调用add(e),则会抛出异常。
- offer(e) poll() peek() 方法即不会阻塞线程,也不会抛出异常。例如:当队列被元素填满后,再调用offer(e),则不会插入元素,函数返回false。
- 要想要实现阻塞功能,需要调用put(e) take() 方法。当不满足约束条件时,会阻塞线程。
好,上点源码你就更明白了。以
ArrayBlockingQueue类为例:
对于第一类方法,很明显如果操作不成功就抛异常。而且可以看到其实调用的是第二类的方法,为什么?因为第二类方法返回boolean啊。
public boolean add(E e) {
if (offer(e))
return true;
else
throw new IllegalStateException("Queue full");//队列已满,抛异常
}
public E remove() {
E x = poll();
if (x != null)
return x;
else
throw new NoSuchElementException();//队列为空,抛异常
}
注:先不看阻塞与否,这ReentrantLock的使用方式就能说明这个类是线程安全类。
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == items.length)//队列已满,返回false
return false;
else {
insert(e);//insert方法中发出了notEmpty.signal();
return true;
}
} finally {
lock.unlock();
}
}
public E poll() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
if (count == 0)//队列为空,返回false
return null;
E x = extract();//extract方法中发出了notFull.signal();
return x;
} finally {
lock.unlock();
}
}
对于第三类方法,这里面涉及到Condition类,简要提一下,
await方法指:造成当前线程在接到信号或被中断之前一直处于等待状态。
signal方法指:唤醒一个等待线程。
public void put(E e) throws InterruptedException {
if (e == null) throw new NullPointerException();
final E[] items = this.items;
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
flex对应java数据类型 try {
while (count == items.length)//如果队列已满,等待notFull这个条件,这时当前线程被阻塞
notFull.await();
} catch (InterruptedException ie) {
notFull.signal(); //唤醒受notFull阻塞的当前线程
throw ie;
}
insert(e);
} finally {
lock.unlock();
}
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
try {
while (count == 0)//如果队列为空,等待notEmpty这个条件,这时当前线程被阻塞
notEmpty.await();
} catch (InterruptedException ie) {
notEmpty.signal();//唤醒受notEmpty阻塞的当前线程
throw ie;
}
E x = extract();
return x;
} finally {
lock.unlock();
}
}
第四类方法就是指在有必要时等待指定时间,就不详细说了。
再来看看BlockingQueue接口的具体实现类吧:
- ArrayBlockingQueue,其构造函数必须带一个int参数来指明其大小
- LinkedBlockingQueue,若其构造函数带一个规定大小的参数,生成的BlockingQueue有大小限制,若不带大小参数,所生成的BlockingQueue的大小由Integer.MAX_VALUE来决定
- PriorityBlockingQueue,其所含对象的排序不是FIFO,而是依据对象的自然排序顺序或者是构造函数的Comparator决定的顺序
上面是用ArrayBlockingQueue举得例子,下面看看
LinkedBlockingQueue:
首先,既然是链表,就应该有Node节点,它是一个内部静态类:
static class Node<E> {
/** The item, volatile to ensure barrier separating write and read */
volatile E item;
Node<E> next;
Node(E x) { item = x; }
}
然后,对于链表来说,肯定需要两个变量来标示头和尾:
/** 头指针 */
private transient Node<E> head; //head.next是队列的头元素
/** 尾指针 */
private transient Node<E> last; //last.next是null
么,对于入队和出队就很自然能理解了:
private void enqueue(E x) {
last = last.next = new Node<E>(x); //入队是为last再找个下家
}
private E dequeue() {
Node<E> first = head.next; //出队是把head.next取出来,然后将head向后移一位
head = first;
E x = first.item;
first.item = null;
return x;
}
另外,LinkedBlockingQueue相对于ArrayBlockingQueue还有不同是,有两个ReentrantLock,且队列现有元素的大小由一个AtomicInteger对象标示。
注:AtomicInteger类是以原子的方式操作整型变量。
private final AtomicInteger count = new AtomicInteger(0);
/** 用于读取的独占锁*/
private final ReentrantLock takeLock = new ReentrantLock();
/** 队列是否为空的条件 */
private final Condition notEmpty = takeLock.newCondition();
/** 用于写入的独占锁 */
private final ReentrantLock putLock = new ReentrantLock();
/** 队列是否已满的条件 */
private final Condition notFull = putLock.newCondition();
有两个Condition很好理解,在ArrayBlockingQueue也是这样做的。但是为什么需要两个ReentrantLock呢?下面会慢慢道来。
让我们来看看offer和poll方法的代码:
public boolean offer(E e) {
if (e == null) throw new NullPointerException();
final AtomicInteger count = this.count;
if (count.get() == capacity)
return false;
int c = -1;
final ReentrantLock putLock = this.putLock;//入队当然用putLock
putLock.lock();
try {
if (count.get() < capacity) {
enqueue(e); //入队
c = count.getAndIncrement(); //队长度+1
if (c + 1 < capacity)
notFull.signal(); //队列没满,当然可以解锁了
}
} finally {
putLock.unlock();
}
if (c == 0)
signalNotEmpty();//这个方法里发出了notEmpty.signal();
return c >= 0;
}
public E poll() {
final AtomicInteger count = this.count;
if (count.get() == 0)
return null;
E x = null;
int c = -1;
final ReentrantLock takeLock = this.takeLock;出队当然用takeLock
takeLock.lock();
try {
if (count.get() > 0) {
x = dequeue();//出队
c = count.getAndDecrement();//队长度-1
if (c > 1)
notEmpty.signal();//队列没空,解锁
}
} finally {
takeLock.unlock();
}
if (c == capacity)
signalNotFull();//这个方法里发出了notFull.signal();
return x;
}
看看源代码发现和上面ArrayBlockingQueue的很类似,关键的问题在于:为什么要用两个ReentrantLockputLock和takeLock?
我们仔细想一下,入队操作其实操作的只有队尾引用last,并且没有牵涉到head。而出队操作其实只针对head,和last没有关系。那么就 是说入队和出队的操作完全不需要公用一把锁,所以就设计了两个锁,这样就实现了多个不同任务的线程入队的同时可以进行出队的操作,另一方面由于两个操作所 共同使用的count是AtomicInteger类型的,所以完全不用考虑计数器递增递减的问题。
另外,还有一点需要说明一下:await()和singal()这两个方法执行时都会检查当前线程是否是独占锁的当前线程,如果不是则抛出 java.lang.IllegalMonitorStateException异常。所以可以看到在源码中这两个方法都出现在Lock的保护块中。
posted on 2011-07-18 09:51
墙头草 阅读(4355)
评论(0) 编辑 收藏