4. ConcurrentLinkedQueue non-blocking unbounded linked list queue

ConcurrentLinkedQueue is a thread-safe queue, implemented based on the linked list structure, and is an unbounded queue. Theoretically, the length of the queue can be expanded infinitely.

Like other queues, ConcurrentLinkedQueue also uses the first-in-first-out (FIFO) enqueuing rule to sort elements. (Recommended study: java interview questions)

When we add elements to the queue, the newly inserted element will be inserted into the end of the queue; and when we get an element, it will Removed from the head of the queue.

Because ConcurrentLinkedQueue is a linked list structure, when entering the queue, the inserted elements are extended backwards in order to form a linked list; when dequeuing, they are obtained starting from the first element of the linked list and increasing in sequence;

It is worth noting that when using ConcurrentLinkedQueue, if it involves judging whether the queue is empty, remember not to use size()==0, because in the size() method, the entire linked list is traversed. In practice, when there are many queue elements, the size() method consumes a lot of performance and time. You can simply use isEmpty() to determine if the queue is empty.

public class ConcurrentLinkedQueueTest {<br/>    public static int threadCount = 10;<br/>    public static ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();<br/>    static class Offer implements Runnable {<br/>        public void run() {<br/>            //不建議使用 queue.size()==0，影響效率。可以使用!queue.isEmpty()<br/>            if (queue.size() == 0) {<br/>                String ele = new Random().nextInt(Integer.MAX_VALUE) + "";<br/>                queue.offer(ele);<br/>                System.out.println("入隊(duì)元素為" + ele);<br/>            }<br/>        }<br/>    }<br/>    static class Poll implements Runnable {<br/>        public void run() {<br/>            if (!queue.isEmpty()) {<br/>                String ele = queue.poll();<br/>                System.out.println("出隊(duì)元素為" + ele);<br/>            }<br/>        }<br/>    }<br/>    public static void main(String[] agrs) {<br/>        ExecutorService executorService = Executors.newFixedThreadPool(4);<br/>        for (int x = 0; x < threadCount; x++) {<br/>            executorService.submit(new Offer());<br/>            executorService.submit(new Poll());<br/>        }<br/>        executorService.shutdown();<br/>    }<br/>}<br/>

One output:

入隊(duì)元素為313732926<br/>出隊(duì)元素為313732926<br/>入隊(duì)元素為812655435<br/>出隊(duì)元素為812655435<br/>入隊(duì)元素為1893079357<br/>出隊(duì)元素為1893079357<br/>入隊(duì)元素為1137820958<br/>出隊(duì)元素為1137820958<br/>入隊(duì)元素為1965962048<br/>出隊(duì)元素為1965962048<br/>出隊(duì)元素為685567162<br/>入隊(duì)元素為685567162<br/>出隊(duì)元素為1441081163<br/>入隊(duì)元素為1441081163<br/>出隊(duì)元素為1627184732<br/>入隊(duì)元素為1627184732<br/>

ConcurrentLinkedQuere class diagram

As shown in the figure There are two volatile Node nodes in ConcurrentLinkedQueue, which are used to store the first and last nodes of the list. The head node stores the node whose first item in the linked list is null, and the tail does not always point to the last node.

The Node node maintains a variable item internally to store the value of the node, and next is used to store the next node, thus linking it to a one-way unbounded list.

public ConcurrentLinkedQueue(){<br/>    head=tail=new Node<E>(null);<br/>}<br/>

When the above code is initialized, an empty node with item NULL will be constructed as the head and tail nodes of the linked list.

Offer operation The offer operation is to add an element at the end of the linked list.

Let’s take a look at the implementation principle.

public boolean offer(E e) {<br/>    //e 為 null 則拋出空指針異常<br/>    checkNotNull(e);<br/>    //構(gòu)造 Node 節(jié)點(diǎn)構(gòu)造函數(shù)內(nèi)部調(diào)用 unsafe.putObject，后面統(tǒng)一講<br/>    final Node<E> newNode = new Node<E>(e);<br/>    //從尾節(jié)點(diǎn)插入<br/>    for (Node<E> t = tail, p = t; ; ) {<br/>        Node<E> q = p.next;<br/>        //如果 q=null 說(shuō)明 p 是尾節(jié)點(diǎn)則插入<br/>        if (q == null) {<br/>            //cas 插入（1）<br/>            if (p.casNext(null, newNode)) {<br/>                //cas 成功說(shuō)明新增節(jié)點(diǎn)已經(jīng)被放入鏈表，然后設(shè)置當(dāng)前尾節(jié)點(diǎn)（包含 head，1，3，5.。。個(gè)節(jié)點(diǎn)為尾節(jié)點(diǎn)）<br/>                if (p != t)// hop two nodes at a time<br/>                    casTail(t, newNode); // Failure is OK. return true;<br/>            }<br/>            // Lost CAS race to another thread; re-read next<br/>        } else if (p == q)//(2)<br/>            //多線(xiàn)程操作時(shí)候，由于 poll 時(shí)候會(huì)把老的 head 變?yōu)樽砸?，然?head 的 next 變?yōu)樾?head，所以這里需要<br/>            //重新找新的 head，因?yàn)樾碌?head 后面的節(jié)點(diǎn)才是激活的節(jié)點(diǎn)<br/>            p = (t != (t = tail)) ? t : head;<br/>        else<br/>            // 尋找尾節(jié)點(diǎn)(3)<br/>            p = (p != t && t != (t = tail)) ? t : q;<br/>    }<br/>}<br/>

From the constructor, we know that there is a sentinel node with null item at the beginning, and both head and tail point to this node.

The above is the detailed content of Java multithreading and concurrency interview questions (Question 4, with answers). For more information, please follow other related articles on the PHP Chinese website!