【从入门到放弃-Java】并发编程-JUC-ConcurrentLinkedQueue

简介

队列是一种先进先出的数据结构,在排队、削峰、缓存等多种场景下都会用到。今天学习下JUC中提供的并发队列-ConcurrentLinkedQueue

可以看他的继承和实现接口非常简单,继承了AbstractQueue类,实现了Queue接口。

ConcurrentLinkedQueue

add

在队列尾部新添加一个元素

1
2
3
4
5
6
7
8
9
10
11
12
/**
* 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

在队列尾部新添加一个元素

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
/**
* 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

取出队列头部的一个元素删除

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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

取出队列头的元素,但不删除

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
//取出队列头的元素并在队列中删除,和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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* 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

判断队列是否为空

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
/**
* 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

获取队列的大小,会遍历队列计算。但是因为此方法无锁,获取的数据可能会不准确。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
/**
* 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),性能会随着队列长度降低,且因为无锁,队列可能被增删,不一定能获取到正确的结果。