Java并发编程-进阶篇（下）

ゝ一世哀愁。 2022-04-23 07:20 205阅读 0赞

本章节主要讲解的是常用的并发队列  
ConcurrentLinkedQueue并发无阻塞队列  
BlockingQueue是并发阻塞队列

## 1. 无阻塞队列 ConcurrentLinkedQueue ##

无阻塞、无锁、高性能、无界、线程安全，性能优于BlockingQueue，不允许null值。

内部使用的链表存储数据，因此不会有容量限制。

static final class Node<E> {
        volatile E item;
        volatile Node<E> next;
    
        /**
         * Constructs a node holding item.  Uses relaxed write because
         * item can only be seen after piggy-backing publication via CAS.
         */
        Node(E item) {
            ITEM.set(this, item);
        }
    
        /** Constructs a dead dummy node. */
        Node() {}
    
        void appendRelaxed(Node<E> next) {
            // assert next != null;
            // assert this.next == null;
            NEXT.set(this, next);
        }
    
        boolean casItem(E cmp, E val) {
            // assert item == cmp || item == null;
            // assert cmp != null;
            // assert val == null;
            return ITEM.compareAndSet(this, cmp, val);
        }
    }

添加方法，内部实现为CAS，不加锁的方式实现线程安全。

/**
     * 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) {
        final Node<E> newNode = new Node<E>(Objects.requireNonNull(e));
    
        for (Node<E> t = tail, p = t;;) {
            Node<E> q = p.next;
            if (q == null) {
                // p is last node
                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".
                    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 = (t != (t = tail)) ? t : head;
            else
                // Check for tail updates after two hops.
                p = (p != t && t != (t = tail)) ? t : q;
        }
    }

常用方法  
boolean add(E e) 添加元素，底层调用offer方法  
boolean offer(E e) 添加元素  
int size() 获取容量  
E poll() 读取头元素并移除  
E peek() 读取头元素，但不移除

## 2. ArrayBlockingQueue ##

基于数组实现  
阻塞、有界  
读写互斥  
不允许null值

利用定长数组维护缓存数据

常用方法  
public boolean add(E e) 如果队列已满，则抛出异常。  
public void put(E e) throws InterruptedException 如果队列满了，不抛出异常，但是会阻塞  
public boolean offer(E e) 如果队列已满，不阻塞也不抛出异常  
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException 如果队列已满，最多阻塞时间为timeout, 单位为TimeUnit，阻塞结束后如果队列仍然已满，不阻塞，不抛异常。  
public int size() 获取容量  
public E poll() 读取头元素并移除  
public E peek() 读取头元素，但不移除

## 3. LinkedBlockingQueue ##

基于链表实现  
阻塞  
长度可指定（如果初始化时不指定长度，则最大长度为Integer.MAX\_VALUE，如果指定了，则为指定长度）  
不允许null值

内部结构，使用链表维护缓存数据

/**
     * Linked list node class.
     */
    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; }
    }

常用方法，与ArrayBlockingQueue基本一致。  
public void put(E e) throws InterruptedException 如果队列满了，不抛出异常，但是会阻塞  
public boolean offer(E e) 如果队列已满，不阻塞也不抛出异常  
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException 如果队列已满，最多阻塞时间为timeout, 单位为TimeUnit，阻塞结束后如果队列仍然已满，不阻塞，不抛异常。  
public int size() 获取容量  
public E poll() 读取头元素并移除  
public E peek() 读取头元素，但不移除。如果头元素不存在（队列为空），返回null  
public E poll(long timeout, TimeUnit unit) throws InterruptedException 读取头元素，不过存在则阻塞timeout的时长。

## 4. SynchronousQueue ##

没有任何容量  
必须现有线程从队列中take(), 才能向队列中添加数据，否则会抛出队列已满的一场。  
不能使用peek()方法获取数据，会直接返回null

DEMO代码

package com.jimmy.concurrentdemo;
    
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.TimeUnit;
    
    public class SynchronousQueueDemo {
    
        /**
         * 在没有其他线程调用queue.take() | queue.poll()方法时，put(E e)方法会一直阻塞
         */
        static void testPut() {
            SynchronousQueue<Integer> queue = new SynchronousQueue<>();
            try {
                System.out.println("test put is running");
                queue.put(1);
                System.out.println("test put is over");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    
        /**
         * 只有take()方法先执行，后序的put()方法才能继续执行，而不被阻塞
         * 如果没有put()方法执行，take()方法将阻塞
         * 如果take()方法执行完，还有put()方法没有执行，则阻塞
         */
        static void testTakeAndPut() {
            SynchronousQueue<Integer> queue = new SynchronousQueue<>();
    
            new Thread(() -> {
                System.out.println("take thread running");
                try {
                    System.out.println("take - " + queue.take());
                    System.out.println("take - " + queue.take());
                    System.out.println("take - " + queue.take());
                    System.out.println("take - " + queue.take());
                    System.out.println("take - " + queue.take());
                    System.out.println("take thread is over");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
    
            new Thread(() -> {
                System.out.println("put thread running");
                try {
                    queue.put(6);
                    queue.put(7);
                    queue.put(8);
                    queue.put(9);
                    queue.put(10);
                    System.out.println("put thread is over");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    
        /**
         * 只有poll()方法先执行，后序的put()方法才能继续执行，而不被阻塞
         * 如果没有put()方法执行，poll()方法将返回null
         */
        static void testTakeAndPoll() {
            SynchronousQueue<Integer> queue = new SynchronousQueue<>();
    
            new Thread(() -> {
                System.out.println("poll thread running");
                System.out.println("poll - " + queue.poll());
                System.out.println("poll - " + queue.poll());
                System.out.println("poll - " + queue.poll());
                System.out.println("poll - " + queue.poll());
                System.out.println("poll - " + queue.poll());
                System.out.println("poll - " + queue.poll());
                System.out.println("poll - " + queue.poll());
                System.out.println("pool thread is over");
            }).start();
    
            new Thread(() -> {
                System.out.println("put thread running");
                try {
                    queue.put(1);
                    queue.put(2);
                    queue.put(3);
                    queue.put(4);
                    queue.put(5);
                    System.out.println("put thread is over");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    
        /**
         * 只有poll()方法先执行，后序的put()方法才能继续执行，而不被阻塞
         * 如果没有put()方法执行, poll(long timeout, TimeUnit timeunit)方法在阻塞timeout时间之后将返回null
         */
        static void testTakeAndPollWithExcepiton() {
            SynchronousQueue<Integer> queue = new SynchronousQueue<>();
    
            new Thread(() -> {
                System.out.println("poll thread running");
                try {
                    System.out.println("poll - " + queue.poll());
                    System.out.println("poll - " + queue.poll());
                    System.out.println("poll - " + queue.poll());
                    System.out.println("poll - " + queue.poll());
                    System.out.println("poll - " + queue.poll());
                    System.out.println("poll - " + queue.poll());
                    System.out.println("poll - " + queue.poll(2, TimeUnit.SECONDS));
                    System.out.println("pool thread is over");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
    
            new Thread(() -> {
                System.out.println("put thread running");
                try {
                    queue.put(1);
                    queue.put(2);
                    queue.put(3);
                    queue.put(4);
                    queue.put(5);
                    System.out.println("put thread is over");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    
        public static void main(String[] args) {
            testPut();
            testTakeAndPut();
            testTakeAndPoll();
            testTakeAndPollWithExcepiton();
        }
    }

## 5. PriorityBlockingQueue ##

阻塞队列  
无界（默认长度为11， 也可以手动指定，能够动态扩容， 内存资源被耗尽会抛出OutOfMemoryError）  
不允许使用null  
不允许插入不可比较的对象(插入对象需要实现Comparable接口，否则会抛出ClassCastException)

DEMO

package com.jimmy.concurrentdemo;
    
    import java.util.concurrent.PriorityBlockingQueue;
    
    public class PriorityBlockingQueueDemo {
    
        /**
         * 插入顺序1->3->2->4
         * 出队顺序1->2->3->4
         */
        static void test() {
            System.out.println("test()");
            PriorityBlockingQueue<Element> queue = new PriorityBlockingQueue<>();
            //入队，如果插入null,会抛出空指针异常
            queue.add(new Element(1, 1));
            queue.offer(new Element(3, 3));
            queue.put(new Element(2, 2));
            queue.put(new Element(4, 4));
            System.out.println("入队次序：" + queue.toString());
    
            //出队
            System.out.println("出队次序：1-" + queue.poll().toString());
            System.out.println("出队次序：2-" + queue.poll().toString());
            System.out.println("出队次序：3-" + queue.poll().toString());
            System.out.println("出队次序：4-" + queue.poll().toString());
            System.out.println("test() finish");
        }
    
        /**
         * 插入顺序4->2->1->3
         * 执行take()方法之后，队列顺序变为1->2->3，进行了重排序
         * 出队顺序1->2->3->4
         */
        static void testTake() {
            System.out.println("testTake()");
            PriorityBlockingQueue<Element> queue = new PriorityBlockingQueue<>();
            //入队，如果插入null,会抛出空指针异常
            queue.add(new Element(4, 4));
            queue.offer(new Element(2, 2));
            queue.put(new Element(1, 1));
            queue.put(new Element(3, 4));
            System.out.println("入队次序：" + queue.toString());
    
            //出队
            try {
                System.out.println("出队次序：1-" + queue.take().toString());
                System.out.println("当前队列次序：" + queue.toString());
                System.out.println("出队次序：2-" + queue.take().toString());
                System.out.println("出队次序：3-" + queue.take().toString());
                System.out.println("出队次序：4-" + queue.take().toString());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("testTake() finish");
        }
    
        public static void main(String[] args) {
            test();
            testTake();
        }
    }
    
    class Element implements Comparable<Element> {
        private int id;
        private int value;
    
        public Element(int id, int value) {
            this.id = id;
            this.value = value;
        }
    
        public int getId() {
            return id;
        }
    
        public void setId(int id) {
            this.id = id;
        }
    
        public int getValue() {
            return value;
        }
    
        public void setValue(int value) {
            this.value = value;
        }
    
        @Override
        public int compareTo(Element o) {
            return Integer.compare(this.id, o.id);
        }
    
        @Override
        public String toString() {
            return "Element{" +
                    "id=" + id +
                    ", value=" + value +
                    '}';
        }
    }

常用方法  
public boolean add(E e) 队列无界  
public void put(E e) throws InterruptedException 队列无界  
public boolean offer(E e) 队列无界  
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException 队列无界  
public int size() 获取容量  
public E poll() 读取头元素并移除，如果队列为空，不阻塞也不抛异常，返回null  
public E peek() 读取头元素，但不移除，如果队列为空，不阻塞也不抛异常，返回null

## 6. DelayQueue ##

阻塞队列  
无界  
只有延迟期满才能从队列中取出元素，该队列的头部是延迟期满后保存时间最长的Delayed元素（即到期时间最早的元素）。  
如果延迟期都没有满，则队列没有头部，获取元素将放回null  
不允许使用null元素

内部元素需要实现Delayed接口

场景：缓存到期删除、任务超时处理、空闲链接关闭等

DEMO

package com.jimmy.concurrentdemo;
    
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    public class DelayQueueDemo {
        public DelayQueue<DelayedElement> queue = new DelayQueue<>();
    
        public void add(DelayedElement e) {
            System.out.println("add: " + e.toString());
            queue.add(e);
        }
    
        public DelayedElement remove() {
            DelayedElement e = null;
            try {
                e = queue.take();
                System.out.println("remove: " + e.toString());
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
            return e;
        }
    
        public void inQueue() {
            System.out.println("elements in queue: " + queue.toString());
        }
    
        public static void main(String[] args) {
            DelayQueueDemo demo = new DelayQueueDemo();
            // 入队顺序：1->2->3->4
            demo.add(new DelayedElement(1, "a", 5000 + System.currentTimeMillis()));
            demo.add(new DelayedElement(2, "b", 1000 + System.currentTimeMillis()));
            demo.add(new DelayedElement(3, "c", 10000 + System.currentTimeMillis()));
            demo.add(new DelayedElement(4, "d", 3000 + System.currentTimeMillis()));
    
            //出队顺序: 2->4->1->3
            //先到期的先出队
            //执行完take()方法后进行排序
            while (true) {
                if (demo.queue.size() == 0) break;
                demo.remove();
                demo.inQueue();
            }
    
        }
    }
    
    class DelayedElement implements Delayed {
        private int id;
        private String value;
        private long endTime;
    
        public DelayedElement(int id, String value, long endTime) {
            this.id = id;
            this.value = value;
            this.endTime = endTime;
        }
    
        public int getId() {
            return id;
        }
    
        public void setId(int id) {
            this.id = id;
        }
    
        public String getValue() {
            return value;
        }
    
        public void setValue(String value) {
            this.value = value;
        }
    
        public long getEndTime() {
            return endTime;
        }
    
        public void setEndTime(long endTime) {
            this.endTime = endTime;
        }
    
        @Override
        public long getDelay(TimeUnit unit) {
            return this.endTime - System.currentTimeMillis();
        }
    
        @Override
        public int compareTo(Delayed o) {
            DelayedElement e = (DelayedElement) o;
            return Long.compare(this.endTime, ((DelayedElement) o).getEndTime());
        }
    
        @Override
        public String toString() {
            return "Element{" +
                    "id=" + id +
                    ", value='" + value + '\'' +
                    ", endTime=" + endTime +
                    '}';
        }
    }

输出结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0NDkzNjM_size_16_color_FFFFFF_t_70]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0NDkzNjM_size_16_color_FFFFFF_t_70]: /images/20220323/e3b5b55261cc45a08e3b8ce6055e008e.png