简介
队列是一种先进先出的数据结构，在排队、削峰、缓存等多种场景下都会用到。今天学习下JUC中提供的并发队列-ConcurrentLinkedQueue
可以看他的继承和实现接口非常简单，继承了AbstractQueue类，实现了Queue接口。
ConcurrentLinkedQueue
add
在队列尾部新添加一个元素
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never throw
* {@link IllegalStateException} or return {@code false}.
*
* @return {@code true} (as specified by {@link Collection#add})
* @throws NullPointerException if the specified element is null
*/
//在队列尾部加入新的元素，元素不能为null。这里直接调用看offer方法。
public boolean add(E e) {
return offer(e);
}offer
在队列尾部新添加一个元素
/**
* Inserts the specified element at the tail of this queue.
* As the queue is unbounded, this method will never return {@code false}.
*
* @return {@code true} (as specified by {@link Queue#offer})
* @throws NullPointerException if the specified element is null
*/
public boolean offer(E e) {
//创建一个新的节点，新的节点元素不能为null。
final Node<E> newNode = new Node<E>(Objects.requireNonNull(e));
//从tail开始遍历队列，直到队尾
for (Node<E> t = tail, p = t;;) {
Node<E> q = p.next;
//如果p是最后一个节点
if (q == null) {
// p is last node
//通过CAS的方式把newNode加到下一个节点
if (NEXT.compareAndSet(p, null, newNode)) {
// Successful CAS is the linearization point
// for e to become an element of this queue,
// and for newNode to become "live".
//插入成功后，再次判断，tail是否已经改变，如果p != tail，则尝试将newNode设置为Tail，设置失败也没关系，因为有其它线程设置了
if (p != t) // hop two nodes at a time; failure is OK
TAIL.weakCompareAndSet(this, t, newNode);
return true;
}
// Lost CAS race to another thread; re-read next
}
else if (p == q)
// We have fallen off list. If tail is unchanged, it
// will also be off-list, in which case we need to
// jump to head, from which all live nodes are always
// reachable. Else the new tail is a better bet.
//当p == q时如并发时节点被删除等。需要重新设置p
p = (t != (t = tail)) ? t : head;
else
// Check for tail updates after two hops.
//如果p在上次赋值后，if处理前，节点有新增，则走到这一步
p = (p != t && t != (t = tail)) ? t : q;
}
}poll
取出队列头部的一个元素删除
public E poll() {
restartFromHead: for (;;) {
for (Node<E> h = head, p = h, q;; p = q) {
final E item;
//从队列首部head取出元素，并使用cas将头部设为null。
if ((item = p.item) != null && p.casItem(item, null)) {
// Successful CAS is the linearization point
// for item to be removed from this queue.
if (p != h) // hop two nodes at a time
updateHead(h, ((q = p.next) != null) ? q : p);
return item;
}
//如果队列空了，返回null
else if ((q = p.next) == null) {
updateHead(h, p);
return null;
}
//如果p已经被删除 返回头部重新开始获取
else if (p == q)
continue restartFromHead;
}
}
}peek
取出队列头的元素，但不删除
public E peek() {
restartFromHead: for (;;) {
for (Node<E> h = head, p = h, q;; p = q) {
final E item;
//返回head的元素，如果队列为空，返回null
if ((item = p.item) != null
|| (q = p.next) == null) {
updateHead(h, p);
return item;
}
//如果head被删除掉，则返回头部继续获取
else if (p == q)
continue restartFromHead;
}
}
}remove
//取出队列头的元素并在队列中删除，和poll的区别是，如果队列为空，remove会抛出异常。而poll会返回null
/**
* Retrieves and removes the head of this queue. This method differs
* from {@link #poll poll} only in that it throws an exception if this
* queue is empty.
*
* <p>This implementation returns the result of {@code poll}
* unless the queue is empty.
*
* @return the head of this queue
* @throws NoSuchElementException if this queue is empty
*/
public E remove() {
E x = poll();
//调用poll方法，获取队列头部的元素并删除，如果队列为空，抛出异常
if (x != null)
return x;
else
throw new NoSuchElementException();
}
//移除队列中的某个元素
/**
* Removes a single instance of the specified element from this queue,
* if it is present. More formally, removes an element {@code e} such
* that {@code o.equals(e)}, if this queue contains one or more such
* elements.
* Returns {@code true} if this queue contained the specified element
* (or equivalently, if this queue changed as a result of the call).
*
* @param o element to be removed from this queue, if present
* @return {@code true} if this queue changed as a result of the call
*/
public boolean remove(Object o) {
if (o == null) return false;
restartFromHead: for (;;) {
//从队列头部开始查找
for (Node<E> p = head, pred = null; p != null; ) {
Node<E> q = p.next;
final E item;
//如果找到等于o的元素，则移除并返回true
if ((item = p.item) != null) {
if (o.equals(item) && p.casItem(item, null)) {
skipDeadNodes(pred, p, p, q);
return true;
}
pred = p; p = q; continue;
}
//如果是队列尾部，判断是否新插入了数据。如果插入了，则返回头部重新查找
for (Node<E> c = p;; q = p.next) {
if (q == null || q.item != null) {
pred = skipDeadNodes(pred, c, p, q); p = q; break;
}
if (p == (p = q)) continue restartFromHead;
}
}
//如果没找到 返回false
return false;
}
}element
获取队列头的元素，但不会从队列中删除，和peek的区别是，如果队列为空，则element会抛出异常，peek是返回null
/**
* Retrieves, but does not remove, the head of this queue. This method
* differs from {@link #peek peek} only in that it throws an exception if
* this queue is empty.
*
* <p>This implementation returns the result of {@code peek}
* unless the queue is empty.
*
* @return the head of this queue
* @throws NoSuchElementException if this queue is empty
*/
public E element() {
E x = peek();
//调用peek方法返回队列头部数据但不删除，如果队列为空返回null，则抛出异常
if (x != null)
return x;
else
throw new NoSuchElementException();
}isEmpty
判断队列是否为空
/**
* Returns {@code true} if this queue contains no elements.
*
* @return {@code true} if this queue contains no elements
*/
public boolean isEmpty() {
//如果第一个节点时null 则是空的
return first() == null;
}
/**
* Returns the first live (non-deleted) node on list, or null if none.
* This is yet another variant of poll/peek; here returning the
* first node, not element. We could make peek() a wrapper around
* first(), but that would cost an extra volatile read of item,
* and the need to add a retry loop to deal with the possibility
* of losing a race to a concurrent poll().
*/
//会返回第一个Node，和peek的区别是，peek是返回的第一个Node中的元素
Node<E> first() {
restartFromHead: for (;;) {
for (Node<E> h = head, p = h, q;; p = q) {
boolean hasItem = (p.item != null);
if (hasItem || (q = p.next) == null) {
updateHead(h, p);
return hasItem ? p : null;
}
else if (p == q)
continue restartFromHead;
}
}
}size
获取队列的大小，会遍历队列计算。但是因为此方法无锁，获取的数据可能会不准确。
/**
* Returns the number of elements in this queue. If this queue
* contains more than {@code Integer.MAX_VALUE} elements, returns
* {@code Integer.MAX_VALUE}.
*
* <p>Beware that, unlike in most collections, this method is
* <em>NOT</em> a constant-time operation. Because of the
* asynchronous nature of these queues, determining the current
* number of elements requires an O(n) traversal.
* Additionally, if elements are added or removed during execution
* of this method, the returned result may be inaccurate. Thus,
* this method is typically not very useful in concurrent
* applications.
*
* @return the number of elements in this queue
*/
public int size() {
restartFromHead: for (;;) {
int count = 0;
//会遍历队列中的节点。count值每次加一。因此，size方法是比较耗时的，且在队列中的节点删除或添加时，这个值可能不准。因此建议还是用isEmpty来判断队列是否为空。尽量别使用此方法。
for (Node<E> p = first(); p != null;) {
if (p.item != null)
if (++count == Integer.MAX_VALUE)
break; // @see Collection.size()
if (p == (p = p.next))
continue restartFromHead;
}
return count;
}
}总结
通过对源码的学习，我们了解到以下几个重点：
- ConcurrentLinkedQueue是基于CAS来进行并发控制的，因此同步的代价较小，并发性能比较好。是一个非阻塞队列
- ConcurrentLinkedQueue是无界的，队列中的元素无个数限制
- size方法需要遍历队列的节点时间复杂度为O(n)，性能会随着队列长度降低，且因为无锁，队列可能被增删，不一定能获取到正确的结果。
更多文章
见我的博客：https://nc2era.com
written by AloofJr，转载请注明出处
