Java并发之LinkedBlockingQueue

        上一篇我们已经学习过了 ArrayBlockingQueue的知识及相关方法的使用，这一篇我们就来再学习一下ArrayBlockingQueue的亲戚 LinkedBlockingQueue。在集合类中 ArrayList与 LinkedList会常常拿来比较，ArrayList内部实现是基于数组的，而 LinkedList内部实现是基于链表，所以他们之间会有很多不同，但是本文不会去重点讨论，感兴趣的朋友可以参考我之前发过的几篇文章，那么有请本节的主角 LinkedBlockingQueue！

        LinkedBlockingQueue

        LinkedBlockingQueue是一个一个基于已链接节点的、范围任意（相对而论）的 blocking queue。此队列按 FIFO（先进先出）排序元素。队列的头部 是在队列中时间最长的元素。队列的尾部 是在队列中时间最短的元素。新元素插入到队列的尾部，并且队列获取操作会获得位于队列头部的元素。链接队列的吞吐量通常要高于基于数组的队列，但是在大多数并发应用程序中，其可预知的性能要低。 

        可选的容量范围构造方法参数作为防止队列过度扩展的一种方法。如果未指定容量，则它等于 Integer.MAX_VALUE。除非插入节点会使队列超出容量，否则每次插入后会动态地创建链接节点。 

        LinkedBlockingQueue及其迭代器实现了 Collection 和 Iterator 接口的所有可选 方法。 

        我们已经学习过了 ArrayBlockingQueue，所以学习 LinkedBlockingQueue就自然比较轻松，所以本文对于已经明确的相关概念就不做过多介绍了，而是重点放在两者的区别之上。

 

        1.成员变量

        与ArrayBlockingQueue不同 LinkedBlockingQueue的成员变量有些变化，以下是 LinkedBlockingQueue的成员变量：

/** 容量范围,默认值为 Integer.MAX_VALUE */
private final int capacity;

/** 当前队列中元素数量 */
private final AtomicInteger count = new AtomicInteger(0);

/** 头节点 */
private transient Node<E> head;

/** 尾节点 */
private transient Node<E> last;

/** take, poll等方法的锁 */
private final ReentrantLock takeLock = new ReentrantLock();

/** 获取队列的 Condition（条件）实例 */
private final Condition notEmpty = takeLock.newCondition();

/** put, offer等方法的锁 */
private final ReentrantLock putLock = new ReentrantLock();

/** 插入队列的 Condition（条件）实例 */
private final Condition notFull = putLock.newCondition();

        1）首先 LinkedBlockingQueue明确了容量变量，当为指定容量时，默认容量为Int的最大值Integer.MAX_VALUE。

        2）队列元素数量变量 count采用的是 AtomicInteger ，而不是普通的Int型。CAS相关可参考http://286.iteye.com/blog/2295165

        3）LinkedBlockingQueue内部队列实现使用的是 Node节点类，这与 LinkedList类似。

        4）最后也是最重要的一点，那就是获取与插入操作分成了两个锁：takeLock与 putLock来处理，这点下面还会重点分析。

 

        2.构造方法

        有三个构造方法，分别为默认，指定容量，指定容量和初始元素。

/**
 * 创建一个容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue
 */
public LinkedBlockingQueue() {
	this(Integer.MAX_VALUE);
}

/**
 * 创建一个具有给定（固定）容量的 LinkedBlockingQueue
 */
public LinkedBlockingQueue(int capacity) {
	if (capacity <= 0)
		throw new IllegalArgumentException();
	this.capacity = capacity;
	last = head = new Node<E>(null);
}

/**
 * 创建一个容量为 Integer.MAX_VALUE 的 LinkedBlockingQueue，
 * 最初包含给定 collection 的元素，元素按该 collection 迭代器的遍历顺序添加。
 */
public LinkedBlockingQueue(Collection<? extends E> c) {
	this(Integer.MAX_VALUE);
	for (E e : c)
		add(e);
}

        默认构造方法创建一个容量为 Integer.MAX_VALUE的 LinkedBlockingQueue实例。

        第二种构造方法，指定了队列容量，首先判断指定容量是否大于零，否则抛出异常。然后为 capacity 赋值，最后创建空节点，并指向 head与 last，两者的 item与 next此时均为 null。

 

 

        最后一种，利用循环向队列中添加指定集合中的元素。

 

        3.Node类

        LinkedBlockingQueue内部列表实现是使用的 Node内部类，Node类也并不复杂，以下是其源代码：

/**
 * 节点类
 */
static class Node<E> {
	/** volatile保障读写分离 */
	volatile E item;
	Node<E> next;

	Node(E x) {
		item = x;
	}
}

        item用于表示元素对象，next指向链表的下一个节点。

 

        LinkedBlockingQueue的大部分方法其实是与  ArrayBlockingQueue类似的，所以本文就只介绍不同于ArrayBlockingQueue的相关方法。

 

        4.添加元素

        1）add方法

        add方法相同就不介绍了，同样调用的是offer方法。

        2）offer方法

        将指定元素插入到此队列的尾部（如果立即可行且不会超出此队列的容量），在成功时返回 true，如果此队列已满，则返回 false。当使用有容量限制的队列时，此方法通常要优于 add 方法，后者可能无法插入元素，而只是抛出一个异常。 

        与ArrayBlockingQueue不同，LinkedBlockingQueue多了一些容量方面的判断。

/**
 * 将指定元素插入到此队列的尾部（如果立即可行且不会超出此队列的容量）
 * 在成功时返回 true，如果此队列已满，则返回 false。
 * 当使用有容量限制的队列时，此方法通常要优于 add 方法，
 * 后者可能无法插入元素，而只是抛出一个异常。 
 */
public boolean offer(E e) {
	//判断添加元素是否为null
	if (e == null)
		throw new NullPointerException();
	//第一点不同，使用原子类操作count，因为有两个锁
	final AtomicInteger count = this.count;
	//判断容量，队列是否已满
	if (count.get() == capacity)
		return false;
	int c = -1;
	final ReentrantLock putLock = this.putLock;
	//获取添加锁
	putLock.lock();
	try {
		//再次判断，如果队列未满
		if (count.get() < capacity) {
			//插入元素
			insert(e);
			//增加元素数count
			c = count.getAndIncrement();
			if (c + 1 < capacity)
				//未满则唤醒添加线程
				notFull.signal();
		}
	} finally {
		//释放锁
		putLock.unlock();
	}
	//c等于0说明添加成功
	if (c == 0)
		//唤醒读取线程
		signalNotEmpty();
	return c >= 0;
}

        可以看到offer方法的关键在于 insert方法。

        3）insert方法

         insert方法非常简单，但是却不要小看。

/**
 * 再队尾添加元素
 */
private void insert(E x) {
	last = last.next = new Node<E>(x);
}

        首先，根据指定参数x创建一个Node实例。

        然后，将原尾节点的next指向此节点。

        最后，将尾节点设置尾此节点。

        这样新添加的节点就成为了新的尾节点。

 

        当向链表中添加第一个节点时，因为在初始化时

last = head = new Node<E>(null);

        所以此时 head与 last指向的是同一个对象new Node<E>(null)。

        之后将last.next指向x。

last.next = new Node<E>(x);

        因为此时 head与 last是同一个对象，所以 head.next也指向x。

        最后将 last指向x。

last =  new Node<E>(x);

        这样 head的next就指向了 last。此时head中的 item仍为 null。

 

        4）put方法

        将指定元素插入到此队列的尾部，如有必要，则等待空间变得可用。

/**
 * 将指定元素插入到此队列的尾部，如有必要，则等待空间变得可用
 */
public void put(E e) throws InterruptedException {
	//判断添加元素是否为null
	if (e == null)
		throw new NullPointerException();
	int c = -1;
	final ReentrantLock putLock = this.putLock;
	final AtomicInteger count = this.count;
	//获取插入的可中断锁
	putLock.lockInterruptibly();
	try {
		try {
			//判断队列是否已满
			while (count.get() == capacity)
				//如果已满则阻塞添加线程
				notFull.await();
		} catch (InterruptedException ie) {
			//失败就唤醒添加线程
			notFull.signal(); 
			throw ie;
		}
		//添加元素
		insert(e);
		//修改c值
		c = count.getAndIncrement();
		//根据c值判断队列是否已满
		if (c + 1 < capacity)
			//未满则唤醒添加线程
			notFull.signal();
	} finally {
		//释放锁
		putLock.unlock();
	}
	//c等于0代表添加成功
	if (c == 0)
		signalNotEmpty();
}

 

        5.获取元素

        1）peek方法

        peek方法获取但不移除此队列的头；如果此队列为空，则返回 null。

/**
 * 获取但不移除此队列的头；如果此队列为空，则返回 null
 */
public E peek() {
	//判断元素数是否为0
	if (count.get() == 0)
		return null;
	final ReentrantLock takeLock = this.takeLock;
	//获取获取锁
	takeLock.lock();
	try {
		//头节点的 next节点即为添加的第一个节点
		Node<E> first = head.next;
		//如果不为空则返回该节点
		if (first == null)
			return null;
		else
			return first.item;
	} finally {
		//释放锁
		takeLock.unlock();
	}
}

        peek方法从头节点直接就可以获取到第一个添加的元素，所以效率是比较高的。如果不存在则返回null。

        2）poll方法

        poll方法获取并移除此队列的头，如果此队列为空，则返回 null。

/**
 * 获取并移除此队列的头，如果此队列为空，则返回 null
 */
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) {
			//调用extract方法获取第一个元素
			x = extract();
			//c=count++
			c = count.getAndDecrement();
			//如果队列中含有元素
			if (c > 1)
				//唤醒读取线程
				notEmpty.signal();
		}
	} finally {
		//释放锁
		takeLock.unlock();
	}
	//如果队列已满
	if (c == capacity)
		//唤醒等待中的添加线程
		signalNotFull();
	return x;
}

        poll与 peek方法不同在于poll获取完元素后移除这个元素，获取与移除是通过 extract()方法实现的。

        注意：其中需要注意的是最后部分代码：

//如果队列已满
if (c == capacity)
	//唤醒等待中的添加线程
	signalNotFull();

        肯定会有朋友有以下疑问：

        1）队列都已经满了，还需要唤醒添加线程干什么？

        2）线程满了就不应该再向里面添加元素了啊？

        3）signalNotFull方法是干什么的？

    signalNotFull方法的作用是唤醒等待中的put线程，signalNotFull只能被 take/poll方法调用，以下是 signalNotFull方法的源代码：

/**
 * 唤醒等待中的put线程，只能被 take/poll方法调用
 */
private void signalNotFull() {
    final ReentrantLock putLock = this.putLock;
    //获取锁
    putLock.lock();
    try {
    	//唤醒添加线程
        notFull.signal();
    } finally {
    	//释放锁
        putLock.unlock();
    }
}

      前两点问题其实转换一下角度就能很好的理解了，虽然队列已经满了，但是此时本线程已经完成了添加，但是其他线程还在等待获取条件进行添加，如果不去主动唤醒的话，那么这些添加操作就只能无限期的等待下去，所以这些等待的添加操作就会失效。所以此时需要唤醒已经排队的添加线程，虽然他们已经无法添加元素至队列。

 

        3）extract方法

        extract方法用于获取并移除头节点。

/**
 * 获取并移除头节点
 */
private E extract() {
	//获取第一个节点，即 head的下一个元素
	Node<E> first = head.next;
	//将head指向此元素
	head = first;
	//获取元素值
	E x = first.item;
	//清除first的item元素为空，即head元素的item为空
	first.item = null;
	//返回
	return x;
}

        这里需要注意的是这里指的头节点并不是 head，而是 head的 next所指 Node的 item元素。因为 head的 item永远为 null。last的 next永远为 null。

        4）take方法

        获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）。

/**
 * 获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）
 */
public E take() throws InterruptedException {
	E x;
	int c = -1;
	final AtomicInteger count = this.count;
	final ReentrantLock takeLock = this.takeLock;
	//获取可中断锁
	takeLock.lockInterruptibly();
	try {
		try {
			//判断队列是否含有元素
			while (count.get() == 0)
				//没有元素就阻塞获取线程，因为没有元素所以获取线程也就没有必要运行
				notEmpty.await();
		} catch (InterruptedException ie) {
			//失败就唤醒获取线程
			notEmpty.signal(); 
			throw ie;
		}
		//调用 extract方法获取元素
		x = extract();
		//计数c的新值
		c = count.getAndDecrement();
		//如果元素数大于1
		if (c > 1)
			//唤醒获取线程
			notEmpty.signal();
	} finally {
		//释放锁
		takeLock.unlock();
	}
	//如果队列已满
	if (c == capacity)
		//唤醒还在等待的put线程
		signalNotFull();
	return x;
}

        与 poll方法类似，只是take方法采用阻塞的方式来获取元素。

 

        7.其他方法

        1）remainingCapacity方法

/**
 * 返回理想情况下（没有内存和资源约束）此队列可接受并且不会被阻塞的附加元素数量
 */
public int remainingCapacity() {
    return capacity - count.get();
}

        也就是返回可以立即添加元素的数量。

 

        2）iterator方法

        iterator方法返回在队列中的元素上按适当顺序进行迭代的迭代器。返回的 Iterator 是一个“弱一致”的迭代器，从不抛出 ConcurrentModificationException，并且确保可遍历迭代器构造后所存在的所有元素，并且可能（但并不保证）反映构造后的所有修改。 

/**
 * 返回Itr实例
 */
public Iterator<E> iterator() {
    return new Itr();
}

        iterator方法返回的是一个Itr内部类的实例，通过这个实例可以遍历整个队列。以下是Itr内部类的源代码：

private class Itr implements Iterator<E> {
	//当前节点
	private Node<E> current;
	private Node<E> lastRet;
	//当前元素
	private E currentElement;

	Itr() {
		final ReentrantLock putLock = LinkedBlockingQueue.this.putLock;
		final ReentrantLock takeLock = LinkedBlockingQueue.this.takeLock;
		//获取获取与添加锁
		putLock.lock();
		takeLock.lock();
		try {
			current = head.next;
			if (current != null)
				currentElement = current.item;
		} finally {
			takeLock.unlock();
			putLock.unlock();
		}
	}

	public boolean hasNext() {
		return current != null;
	}

	public E next() {
		final ReentrantLock putLock = LinkedBlockingQueue.this.putLock;
		final ReentrantLock takeLock = LinkedBlockingQueue.this.takeLock;
		putLock.lock();
		takeLock.lock();
		try {
			if (current == null)
				throw new NoSuchElementException();
			E x = currentElement;
			lastRet = current;
			current = current.next;
			if (current != null)
				currentElement = current.item;
			return x;
		} finally {
			takeLock.unlock();
			putLock.unlock();
		}
	}

	public void remove() {
		if (lastRet == null)
			throw new IllegalStateException();
		final ReentrantLock putLock = LinkedBlockingQueue.this.putLock;
		final ReentrantLock takeLock = LinkedBlockingQueue.this.takeLock;
		putLock.lock();
		takeLock.lock();
		try {
			Node<E> node = lastRet;
			lastRet = null;
			Node<E> trail = head;
			Node<E> p = head.next;
			while (p != null && p != node) {
				trail = p;
				p = p.next;
			}
			if (p == node) {
				p.item = null;
				trail.next = p.next;
				if (last == p)
					last = trail;
				int c = count.getAndDecrement();
				if (c == capacity)
					notFull.signalAll();
			}
		} finally {
			takeLock.unlock();
			putLock.unlock();
		}
	}
}

         Itr类不复杂，我就不详细解释了。

        3）清除方法

        clear，drainTo等方法与 ArrayBlockingQueue类似，这里就不说了。

 

        8,.LinkedBlockingQueue与 ArrayBlockingQueue

        1）内部实现不同

        ArrayBlockingQueue内部队列存储使用的是数组：

private final E[] items;

        而 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; }
}

 

        2）队列中锁的实现不同

/** LinkedBlockingQueue的获取锁 */
private final ReentrantLock takeLock = new ReentrantLock();

/** LinkedBlockingQueue的添加锁 */
private final ReentrantLock putLock = new ReentrantLock();


/** ArrayBlockingQueue的唯一锁 */
private final ReentrantLock lock;

        从源代码就可以看出 ArrayBlockingQueue实现的队列中的锁是没有分离的，即添加与获取使用的是同一个锁；而 LinkedBlockingQueue实现的队列中的锁是分离的，即添加用的是 putLock，获取是 takeLock。

        3）初始化条件不同

        ArrayBlockingQueue实现的队列中必须指定队列的大小。

        LinkedBlockingQueue实现的队列中可以不指定队列的大小，默认容量为Integer.MAX_VALUE。

        4）操作不同

        ArrayBlockingQueue无论是添加还是获取使用的是同一个锁，所以添加的同时就不能读取，读取的同时就不能添加，所以锁方面性能不如 LinkedBlockingQueue。

        LinkedBlockingQueue读取与添加操作使用不同的锁，因为其内部实现的特殊性，添加的时候只需要修改 last即可，而不会影响 head节点。而获取时也只需要修改 head节点即可，同样不会影响 last节点。所以在添加获取方面理论上性能会高于 ArrayBlockingQueue。

        所以 LinkedBlockingQueue更适合实现生产者-消费者队列。