posts - 241,  comments - 116,  trackbacks - 0
今天就聊聊这两种Queue,本文分为以下两个部分,用分割线分开:
  • BlockingQueue
  • ConcurrentLinkedQueue,非阻塞算法


首先来看看BlockingQueue
Queue是什么就不需要多说了吧,一句话:队列是先进先出。相对的,栈是后进先出。如果不熟悉的话先找本基础的数据结构的书看看吧。

BlockingQueue,顾名思义,“阻塞队列”:可以提供阻塞功能的队列。
首先,看看BlockingQueue提供的常用方法:
        Throws exceptionSpecial valueBlocks       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 墙头草 阅读(4349) 评论(0)  编辑  收藏

只有注册用户登录后才能发表评论。


网站导航:
 
人人游戏网 软件开发网 货运专家