认识ArrayBlockingQueue、LinkedBlockingQueue、ConcurrentLinkedQueue

本文的主要内容是对jdk并发包中ArrayBlockingQueue、LinkedBlockingQueue、ConcurrentLinkedQueue类的代码解析，主要对比他们的不同。

 

前面两个类都是生产者消费者模型的实现，性能都不错，这毕竟都是大师的杰作。先来认识下ArrayBlockingQueue，这个类的内部数据结构是一个Object数组，通过ReentrantLock控制同步操作，由这个锁派生出两个条件对象，如下代码

public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = (E[]) new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

 

这是非常经典的条件操作的应用。下面看下put方法的实现

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 {
            try {
                while (count == items.length)
                    notFull.await();
            } catch (InterruptedException ie) {
                notFull.signal(); // propagate to non-interrupted thread
                throw ie;
            }
            insert(e);
        } finally {
            lock.unlock();
        }
    }


 private void insert(E x) {
        items[putIndex] = x;
        putIndex = inc(putIndex);
        ++count;
        notEmpty.signal();
    }

它的思路是：如果队列满，等待，否则入队操作，并发出队列不空的信号。

      以上的思路似乎很合理，但是仔细想想还可以改进：只要在队列为空的时候，入队成功才需发信号说队列不空了；还有一点就是，入队成功后，如果队列依然不满，可以唤醒正在等待的入队操作。这两点都是可以考虑的。

 

以上提到的在LinkedBlockingQueue中得到了实现，它的内部数据结构是一个单向链表，通过两个ReentrantLock来控制同步操作，下面是put方法的实现

public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset
        // local var holding count  negative to indicate failure unless set.
        int c = -1;
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from
             * capacity. Similarly for all other uses of count in
             * other wait guards.
             */
            try {
                while (count.get() == capacity)
                    notFull.await();
            } catch (InterruptedException ie) {
                notFull.signal(); // propagate to a non-interrupted thread
                throw ie;
            }
            insert(e);
            c = count.getAndIncrement();
            if (c + 1 < capacity) // 标记1
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)  // 标记2
            signalNotEmpty();
    }

 private void insert(E x) {
        last = last.next = new Node<E>(x);
    }

上述代码中的两个标记处就实现了前面提到的两个不足。

take()方法的对比类似。

 

下面再来说说ConcurrentLinkedQueue，这个类和上面两个有本质的不同，它的内部采用的是非阻塞算法实现，性能比前两个都要好(书上说的，用机会测试下)，这个类中采用的原子类是域更新器，看下代码更清楚

 private static final
        AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
        tailUpdater =
        AtomicReferenceFieldUpdater.newUpdater
        (ConcurrentLinkedQueue.class, Node.class, "tail");
    private static final
        AtomicReferenceFieldUpdater<ConcurrentLinkedQueue, Node>
        headUpdater =
        AtomicReferenceFieldUpdater.newUpdater
        (ConcurrentLinkedQueue.class,  Node.class, "head");

Node节点中还有两个

 private static class Node<E> {
        private volatile E item;
        private volatile Node<E> next;

        private static final
            AtomicReferenceFieldUpdater<Node, Node>
            nextUpdater =
            AtomicReferenceFieldUpdater.newUpdater
            (Node.class, Node.class, "next");
        private static final
            AtomicReferenceFieldUpdater<Node, Object>
            itemUpdater =
            AtomicReferenceFieldUpdater.newUpdater
            (Node.class, Object.class, "item");

        Node(E x) { item = x; }

        Node(E x, Node<E> n) { item = x; next = n; }

分别是用来更新队列的头结点、尾节点以及每个节点的item和next。

下面看下它的代码

public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        Node<E> n = new Node<E>(e, null);
        for (;;) {
            Node<E> t = tail;
            Node<E> s = t.getNext();
            if (t == tail) {
                if (s == null) {
                    if (t.casNext(s, n)) {
                        casTail(t, n);
                        return true;
                    }
                } else {
                    casTail(t, s);// 标记1
                }
            }
        }
    }

    public E poll() {
        for (;;) {
            Node<E> h = head;
            Node<E> t = tail;
            Node<E> first = h.getNext();
            if (h == head) {
                if (h == t) {
                    if (first == null)
                        return null;
                    else
                        casTail(t, first);// 标记2
                } else if (casHead(h, first)) {
                    E item = first.getItem();
                    if (item != null) {
                        first.setItem(null);
                        return item;
                    }
                    // else skip over deleted item, continue loop,
                }
            }
        }
    }

代码中的关键问题就是操作的原子性：单个原子操作当然是原子性的，但两个原子操作合起来就不是原子的了。比如在入队操作中要加入节点n，在更新tail节点的next成功后，tail节点指向n不一定会成功；这就需要在其他的操作中弥补这一“过错”，这在offer和poll中都有考虑的，如上述代码中的标记1和标记2。