深入理解LinkedBlockingQueue

╰+攻爆jí腚メ 2022-05-28 22:39 91阅读 0赞

# 说明 #

通过阅读源码，了解LinkedBlockingQueue的特性。本文基于JDK1.7源码

# 正文 #

通过查询API对LinkedBlockingQueue特点进行简单的了解：

*  LinkedBlockingQueue是一个**基于已链接节点**的，**范围任意**的blocking queue
 *  此队列按**FIFO（先进先出）排序**元素
 *  新元素插入到队列的尾部，并且队列获取操作会获得位于队列头部的元素
 *  **链接队列的吞吐量通常要高于基于数组的对列**（ArrayBlockingQueue）,但是在大多数并发应用程序中，其可预知的性能要低
 *  可选的容量范围构造方法参数作为防止队列过度扩展的一种方法，如果**未指定容量，则等于Integer.MAX\_VALUE**，除非插入节点会使队列超出容量，否则每次**插入后会动态地创建链接节点**

## LinkedBlockingQueue ##

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
            implements BlockingQueue<E>, java.io.Serializable {

LinkedBlockingQueue继承自AbstractQueue,实现了BlockingQueue,Serializable接口  
关于BlockingQueue接口在[《深入理解ArrayBlockingQueue》][ArrayBlockingQueue]一文中已有介绍，在此不再赘述

## 节点内部类Node ##

static class Node<E> {
            E item;
    
            /**
             * One of:
             * - the real successor Node
             * - this Node, meaning the successor is head.next
             * - null, meaning there is no successor (this is the last node)
             */
            Node<E> next; //指向下个节点
    
            Node(E x) { item = x; }
        }

## LinkedBlockingQueue的属性 ##

private final int capacity; // 队列容量，如果构造时未指定则为Integer.MAX_VALUE
    
    	//使用AtomicInteger来统计队列中元素数量
        private final AtomicInteger count = new AtomicInteger(0);
    
        /**
         * Head of linked list.
         * Invariant: head.item == null
         */
        private transient Node<E> head; // 队列的头元素 值为null 
    
        /**
         * Tail of linked list.
         * Invariant: last.next == null
         */
        private transient Node<E> last; // 队列的尾元素 它的下一个节点为null
    
        /** Lock held by take, poll, etc */
        private final ReentrantLock takeLock = new ReentrantLock(); // 出队锁
    
        /** Wait queue for waiting takes */
        private final Condition notEmpty = takeLock.newCondition(); // 取出线程condition
    
        /** Lock held by put, offer, etc */
        private final ReentrantLock putLock = new ReentrantLock(); // 入队锁
    
        /** Wait queue for waiting puts */
        private final Condition notFull = putLock.newCondition(); // 添加线程condition

## LinkedBlockingQueue的构造函数 ##

public LinkedBlockingQueue() {
            this(Integer.MAX_VALUE); //当未指定容量时，为Integer.MAX_VALUE
        }

public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null); // 构造队列时创建值为null的节点
        }

## LinkedBlockingQueue的添加方法 ##

### put(e) ###

该方法没有返回值，当队列已满时，会阻塞当前线程

public void put(E e) throws InterruptedException {
            if (e == null) throw new NullPointerException();//先检查添加值是否为null
            // 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; //使用putLock来保证线程安全
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
               
                while (count.get() == capacity) {//当队列已满，添加线程阻塞
                    notFull.await();
                }
                enqueue(node); // 调用enqueue方法添加到队尾
                c = count.getAndIncrement(); //调用AtomicInteger的getAndIncrement()是数量加1
                if (c + 1 < capacity)//添加成功后判断是否可以继续添加，队列未满
                    notFull.signal(); //唤醒添加线程
            } finally {
                putLock.unlock();
            }
            if (c == 0) // 添加后如果队列中只有一个元素，唤醒一个取出线程，使用取出锁
                signalNotEmpty();
        }

### offer(e,timeout,unit) ###

该方法返回true或false，当队列已满时，会阻塞给定时间，添加操作成功返回true，否则返回false

public boolean offer(E e, long timeout, TimeUnit unit)
            throws InterruptedException {
    
            if (e == null) throw new NullPointerException();
            long nanos = unit.toNanos(timeout);
            int c = -1;
            final ReentrantLock putLock = this.putLock; //使用putLock
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
                while (count.get() == capacity) { //当队列已满阻塞给定时间
                    if (nanos <= 0) //当时间消耗完全，操作未成功 返回false
                        return false; 
                    nanos = notFull.awaitNanos(nanos);
                }
                enqueue(new Node<E>(e)); // 调用enqueue方法添加一个新的节点
                c = count.getAndIncrement(); //同样调用AtomicInteger的方法
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return true; // 操作成功返回true
        }

### offer(e) ###

该方法返回true或false，不会阻塞，直接返回

public boolean offer(E e) {
            if (e == null) throw new NullPointerException();
            final AtomicInteger count = this.count;
            if (count.get() == capacity)
                return false; //当队列已满，直接返回false
            int c = -1;
            Node<E> node = new Node(e); // 先创建新的节点
            final ReentrantLock putLock = this.putLock;//使用putLock
            putLock.lock();
            try {
                if (count.get() < capacity) { // 加锁后再次判断队列是否已满
                    enqueue(node); //调用enqueue方法将节点添加到队尾
                    c = count.getAndIncrement();
                    if (c + 1 < capacity)
                        notFull.signal();
                }
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return c >= 0; // 比较c的大小，判断是否成功，当c大于-1时则添加操作成功
        }

通过以上三种添加方法我们发现：

*  添加时，使用**putLock这个可重入锁**保证线程安全
 *  共同调用了\*\*enqueue()\*\*方法实现节点的添加
 *  改变队列元素数量时，调用了\*\*AtomicInteger的getAndIncrement()\*\*方法，保证原子性

### enqueue(e) ###

/**
         * Links node at end of queue.
         *
         * @param node the node
         */
        private void enqueue(Node<E> node) {
            // assert putLock.isHeldByCurrentThread();
            // assert last.next == null;
            last = last.next = node; //将新的节点添加到队尾，并变成新的尾节点
        }

### getAndIncrement() ###

此方法内部调用了compareAndSet()方法

public final int getAndIncrement() {
            for (;;) { // 采用循环的方式，将值加1
                int current = get();
                int next = current + 1;
                if (compareAndSet(current, next))
                    return current; //加1成功后返回原值
            }
        }

**compareAndSet( expect, update)**  
此方法内部调用了unsafe类的compareAndSwapInt()方法，该方法共有四个参数，第一个参数为需要修改的对象，第二个为偏移量（内存的旧值），第三个参数为期待的值，第四个为更新后的值；**若果valueOffset的值和expect的值相等，则将valueOffset值修改为update返回true，否则不做操作，返回false**

public final boolean compareAndSet(int expect, int update) {
            return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
        }

## LinkedBlockingQueue的取出方法 ##

### take() ###

public E take() throws InterruptedException {
            E x;
            int c = -1;
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;//使用takeLock保证线程安全
            takeLock.lockInterruptibly();
            try {
                while (count.get() == 0) {//当队列为空，取出线程阻塞
                    notEmpty.await();
                }
                x = dequeue(); //掉用dequeue方法从队头取出元素
                c = count.getAndDecrement(); //调用AtomicInteger的getAndDecrement()将count值减1
                if (c > 1)//判断如果当前队列之前元素的数量大于1，唤醒取出线程
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)//之前队列元素数量为容量值，取出一个，只能唤醒一个添加线程
                signalNotFull();
            return x;
        }

### poll(timeout,unit) ###

该方法取出元素时，如果队列为空，则阻塞给定的时间

public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            E x = null;
            int c = -1;
            long nanos = unit.toNanos(timeout);
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;//使用takeLock保证线程安全        
            takeLock.lockInterruptibly();
            try {
                while (count.get() == 0) {//当队列为空则阻塞给定时间
                    if (nanos <= 0)//时间消耗完全后，如果操作未成功则返回null
                        return null;
                    nanos = notEmpty.awaitNanos(nanos);
                }
                x = dequeue();//调用dequeue方法返回节点值
                c = count.getAndDecrement();//将count值减1
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }

### poll() ###

该方法取出元素时，如果队列为空，则直接返回null

public E poll() {
            final AtomicInteger count = this.count;
            if (count.get() == 0)// 如果队列为空，直接返回null
                return null;
            E x = null;
            int c = -1;
            final ReentrantLock takeLock = this.takeLock;//使用takeLock保证线程安全
            takeLock.lock();
            try {
                if (count.get() > 0) {
                    x = dequeue();//调用dequeue方法取出队头节点元素的值
                    c = count.getAndDecrement();//count减1
                    if (c > 1)//如果取出元素不是唯一的，唤醒取出线程
                        notEmpty.signal();
                }
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)//如果从已满队列取出的，则唤醒一个添加线程
                signalNotFull();
            return x;
        }

通过以上三种取出方法，发现：

*  所有的取出方法都是使用**takeLock这个可重入锁**保证线程安全
 *  都调用了\*\*dequeue()\*\*方法从队头取出元素
 *  调用了AtomicInteger的**getAndDecrement()方法将count值减1**，保证原子性

### dequeue() ###

/**
         * Removes a node from head of queue.
         *
         * @return the node
         */
        private E dequeue() {
            // assert takeLock.isHeldByCurrentThread();
            // assert head.item == null;
            Node<E> h = head; //队列的头结点是值为null的节点
            Node<E> first = h.next; //返回头节点之后的第一个节点
            h.next = h; // help GC 
            //因为创建节点时创建了一个新的对象，所以需要GC，即需要将头节点的后继节点指向自身，帮助GC
            head = first;//将新的头节点置为将要删除的第一个节点
            E x = first.item; //将节点的值赋给x
            first.item = null;//将节点值置为null，变为新的头节点
            return x;//返回取出的值
        }

### getAndDecrement() ###

此方法与getAndIncrement()方法相似，底层都是调用了unsafe的compareAndSwapInt方法保证操作的原子性

### peek() ###

该方法只返回队头元素的值，并不能将节点从队列中删除

public E peek() {
            if (count.get() == 0)
                return null;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();
            try {
                Node<E> first = head.next;
                if (first == null)//如果队列为空，则直接返回null
                    return null;
                else
                    return first.item;
            } finally {
                takeLock.unlock();
            }
        }

### remove(o) ###

从队列中删除指定元素值的节点

public boolean remove(Object o) {
            if (o == null) return false;
            fullyLock(); //此时将入队锁和出队锁全部锁住来保证线程安全
            try {
                for (Node<E> trail = head, p = trail.next;
                     p != null;
                     trail = p, p = p.next) {// 循环遍历查找值相等的元素
                    if (o.equals(p.item)) {
                        unlink(p, trail);//调用unlink删除此节点
                        return true;//操作成功返回true
                    }
                }
                return false;
            } finally {
                fullyUnlock();
            }
        }

**fullyLock()**

void fullyLock() {
            putLock.lock();
            takeLock.lock();
        }

**unlink(p, trail)**

void unlink(Node<E> p, Node<E> trail) {//p为要删除节点，trail为删除节点的前一个节点
            // assert isFullyLocked();
            // p.next is not changed, to allow iterators that are
            // traversing p to maintain their weak-consistency guarantee.
            p.item = null;
            trail.next = p.next; // 改变指针将前一节点的后继节点指向删除节点的后一个节点
            if (last == p)
                last = trail;
            if (count.getAndDecrement() == capacity)
                notFull.signal();
        }

## 获取队列当前大小及剩余容量 ##

### size() ###

public int size() {
            return count.get();
        }

### remainingCapacity() ###

public int remainingCapacity() {
            return capacity - count.get();
        }

这两个方法都没有使用锁来保证线程安全，是因为count自身为AtomicInteger对象，保证了操作的原子性

## 参考资料 ##

[java Unsafe类中compareAndSwap相关介绍][java Unsafe_compareAndSwap]  
[源码剖析sun.misc.Unsafe && Compare And Swap(CAS)操作][sun.misc.Unsafe _ Compare And Swap_CAS]  
[Java8中CAS的增强][Java8_CAS]

[ArrayBlockingQueue]: http://blog.csdn.net/sinat_36553913/article/details/79529197
[java Unsafe_compareAndSwap]: http://blog.csdn.net/sherld/article/details/42492259
[sun.misc.Unsafe _ Compare And Swap_CAS]: http://blog.csdn.net/liupeng900605/article/details/7825286
[Java8_CAS]: http://ifeve.com/enhanced-cas-in-jdk8/