《java.util.concurrent 包源码阅读》07 LinkedBlockingQueue

这篇文章来说说稍微复杂一些的LinkedBlockingQueue。LinkedBlockingQueue使用一个链表来实现,会有一个head和tail分别指向队列的开始和队列的结尾。因此LinkedBlockingQueue会有两把锁,分别控制这两个元素,这样在添加元素和拿走元素的时候就不会有锁的冲突,因此取走元素操作的是head,而添加元素操作的是tail。

老规矩先看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;
        Node<E> node = new Node(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }

可以看到offer方法在添加元素时候仅仅涉及到putLock,但是还是会需要takeLock,看看signalNotEmpty代码就知道。而poll方法拿走元素的时候涉及到takeLock,也是会需要putLock。参见signalNotFull()。关于signalNotEmpty会在后面讲阻塞的时候讲到。

    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.lock();
        try {
            if (count.get() > 0) {
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

这里顺便说说队列长度的count,因为有两把锁存在,所以如果还是像ArrayBlockingQueue一样使用基本类型的count的话会同时用到两把锁,这样就会很复杂,因此直接使用原子数据类型AtomicInteger来操作count。

接下来谈谈阻塞的问题,一个BlockingQueue会有两个Condition:notFull和notEmpty,LinkedBlockingQueue会有两把锁,因此这两个Condition肯定是由这两个锁分别创建的,takeLock创建notEmpty,putLock创建notFull。

    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

接下来看看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;
        Node<E> node = new Node(e);
        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.
             */
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

其实大体逻辑和ArrayBlockingQueue差不多,也会需要通知notEmpty条件,因为notEmpty条件属于takeLock,而调用signal方法需要获取Lock,因此put方法也是用到了另外一个锁:takeLock。这里有一点会不同,按照道理来说put方法是不需要通知notFull条件的,是由由拿走元素的操作来通知的,但是notFull条件属于putLock,而拿走元素时,是用了takeLock,因此这里put方法在拥有putLock的情况通知notFull条件,会让其他添加元素的方法避免过长时间的等待。同理对于take方法来说也通知notEmpty条件。

 

    public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

 

最后说说remove和contains方法,因为需要操作整个链表,因此需要同时拥有锁才能操作。

 

《java.util.concurrent 包源码阅读》07 LinkedBlockingQueue,布布扣,bubuko.com

《java.util.concurrent 包源码阅读》07 LinkedBlockingQueue

上一篇:Mysql 存储过程使用游标


下一篇:mysql5.5 主从配置