【Java并发编程学习 6】并发队列学习

﹏ヽ暗。殇╰゛Y 2023-07-17 04:45 2阅读 0赞

### 并发队列学习 ###

*   *   *  并发队列
         *   *  1 什么是并发队列：
             *  2 并发队列和并发集合的区别：
             *  2 并发队列介绍
             *  3 阻塞队列和非阻塞队列区别
             *  4 Java常用的并发队列：
             *   *  4.1 JDK11的Queue子父关系：
             *  5 常用并发列队的介绍：
             *  6 常用非堵塞队列的使用：
             *   *  6.1 ArrayDeque
                 *   *  ArrayDeque的重要方法：
                     *  ArrayDeque代码示例：
                 *  6.2 ConcurrentLinkedQueue
                 *   *  ConcurrentLinkedQueue重要方法:
                     *  ConcurrentLinkedQueue代码示例：
                 *  6.3 PriorityQueue
                 *   *  PriorityQueue重要方法:
                     *  PriorityQueue代码示例:
                     *  PriorityQueue自定义代码比较器示例:
             *  7 常用阻塞列队的使用
             *   *  7.1 ArrayBlockingQueue
                 *   *  ArrayBlockingQueue重要方法：
                     *  ArrayBlockingQueue代码示例：
                 *  7.2 LinkedBlockingQueue
                 *   *  LinkedBlockingQueue的重要方法：
                     *  LinkedBlockingQueue代码示例：
             *  8 使用BlockingQueue模拟生产者与消费者代码示例：

### 并发队列 ###

#### 1 什么是并发队列： ####

1.  消息队列很多人知道：消息队列是分布式系统中重要的组件，是系统与系统直接的通信
2.  并发队列是什么：并发队列多个线程以有次序共享数据的重要组件

#### 2 并发队列和并发集合的区别： ####

那就有可能要说了，我们并发集合不是也可以实现多线程之间的数据共享吗，其实也是有区别的：

1.  队列遵循“先进先出”的规则，可以想象成排队检票，队列一般用来解决大数据量采集处理和显示的。
2.  并发集合就是在多个线程中共享数据的

#### 2 并发队列介绍 ####

在并发队列上JDK提供了Queue接口，一个是以Queue接口下的BlockingQueue接口为代表的阻塞队列，另一个是高性能（无堵塞）队列。

#### 3 阻塞队列和非阻塞队列区别 ####

1.  当队列阻塞队列为空的时，从队列中获取元素的操作将会被阻塞。
2.  或者当阻塞队列是满时，往队列里添加元素的操作会被阻塞。
3.  或者试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。
4.  试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来

#### 4 Java常用的并发队列： ####

其实判断是否是堵塞队列很简单，只要继承了BlockingQueue的都是堵塞队列

##### 4.1 JDK11的Queue子父关系： #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzEyMjA5MA_size_16_color_FFFFFF_t_70]  
AbstractQueue（翻译：抽象队列）：实现了AbstractQueue的不一定是堵塞或者非堵塞列队  
BlockingQueue（翻译：堵塞队列）：实现了BlockingQueue接口的一定是堵塞列队

#### 5 常用并发列队的介绍： ####

1.  非堵塞队列：  
    1.ArrayDeque, （数组双端队列）  
    2.PriorityQueue, （优先级队列）  
    3.ConcurrentLinkedQueue, （基于链表的并发队列）
2.  堵塞队列：  
    1.DelayQueue, （基于时间优先级的队列，延期阻塞队列）  
    2.ArrayBlockingQueue, （基于数组的并发阻塞队列）  
    3.LinkedBlockingQueue, （基于链表的FIFO阻塞队列）  
    4.LinkedBlockingDeque, （基于链表的FIFO双端阻塞队列）  
    5.PriorityBlockingQueue, （带优先级的无界阻塞队列）  
    6.SynchronousQueue （并发同步阻塞队列）

#### 6 常用非堵塞队列的使用： ####

##### 6.1 ArrayDeque #####

ArrayDeque （非堵塞队列）是JDK容器中的一个双端队列实现，内部使用数组进行元素存储，不允许存储null值，可以高效的进行元素查找和尾部插入取出，是用作队列、双端队列、栈的绝佳选择，性能比LinkedList还要好。

###### ArrayDeque的重要方法： ######

add(E e) 在队列尾部添加一个元素  
offer(E e) 在队列尾部添加一个元素，并返回是否成功  
remove() 删除队列中第一个元素，并返回该元素的值，如果元素为null，将抛出异常(其实底层调用的是removeFirst())  
poll() 删除队列中第一个元素，并返回该元素的值,如果元素为null，将返回null(其实调用的是pollFirst())  
element() 获取第一个元素，如果没有将抛出异常  
peek() 获取第一个元素，如果返回null

###### ArrayDeque代码示例： ######

package com.lijie;
    
    import java.util.ArrayDeque;
    
    public class Test01 { 
        public static void main(String[] args) { 
            ArrayDeque q = new ArrayDeque();
            q.add("张三");
            q.offer("李四");
            q.offer("小红");
    
            //从头获取元素,删除该元素
            System.out.println(q.poll());
            //从头获取元素,不刪除该元素
            System.out.println(q.peek());
            //从头获取元素,不刪除该元素
            System.out.println(q.peek());
            //获取总长度
            System.out.println(q.size());
        }
    }

##### 6.2 ConcurrentLinkedQueue #####

ConcurrentLinkedQueue （非堵塞队列）: 是一个适用于高并发场景下的队列，通过无锁的方式，实现了高并发状态下的高性能。ConcurrentLinkedQueue的性能要好于BlockingQueue接口，它是一个基于链接节点的无界线程安全队列。该队列的元素遵循先进先出的原则。该队列不允许null元素。

###### ConcurrentLinkedQueue重要方法: ######

add() 和offer() 都是是添加元素方法  
poll() 是提取元素，提取完删除元素方法  
peek() 是提取元素，提取完不删除方法

###### ConcurrentLinkedQueue代码示例： ######

package com.lijie;
    
    import java.util.concurrent.ConcurrentLinkedDeque;
    
    public class Test02 { 
        public static void main(String[] args) { 
            ConcurrentLinkedDeque q = new ConcurrentLinkedDeque();
            q.add("张三");
            q.offer("李四");
            q.offer("小红");
            //从头获取元素,删除该元素
            System.out.println(q.poll());
            //从头获取元素,不刪除该元素
            System.out.println(q.peek());
            //从头获取元素,不刪除该元素
            System.out.println(q.peek());
            //获取总长度
            System.out.println(q.size());
        }
    }

##### 6.3 PriorityQueue #####

PriorityQueue （非堵塞队列） 一个基于优先级的无界优先级队列。优先级队列的元素按照其自然顺序进行排序，或者根据构造队列时提供的 Comparator 进行排序，具体取决于所使用的构造方法。该队列不允许使用 null 元素也不允许插入不可比较的对象

###### PriorityQueue重要方法: ######

peek()//返回队首元素  
poll()//返回队首元素，队首元素出队列  
add()//添加元素  
size()//返回队列元素个数  
isEmpty()//判断队列是否为空，为空返回true,不空返回false

###### PriorityQueue代码示例: ######

package com.lijie;
    
    
    import java.util.PriorityQueue;
    
    public class Test03 { 
        public static void main(String[] args) { 
            PriorityQueue<String> q = new PriorityQueue<String>();
            //入列
            q.offer("1");
            q.offer("5");
            q.offer("2");
            q.offer("3");
            q.offer("4");
    
            //出列
            System.out.println(q.poll());  // 1
            System.out.println(q.poll());  // 2
            System.out.println(q.poll());  // 3
            System.out.println(q.poll());  // 4
            System.out.println(q.poll());  // 5
        }
    }

观察打印结果， 入列：15234， 出列是12345， 也是说出列时他自己做了相关的判断

###### PriorityQueue自定义代码比较器示例: ######

package com.lijie;
    
    
    import java.util.Comparator;
    import java.util.PriorityQueue;
    
    public class Test04 { 
        //自定义比较器，降序排列
        static Comparator<Integer> cmp = new Comparator<Integer>() { 
            public int compare(Integer e1, Integer e2) { 
                return e2 - e1;
            }
        };
    
        public static void main(String[] args) { 
            //构造参数添加比较器
            PriorityQueue<Integer> q = new PriorityQueue<>(cmp);
            //入列
            q.offer(1);
            q.offer(2);
            q.offer(3);
            q.offer(4);
            q.offer(5);
    
            //出列
            System.out.println(q.poll());  // 5
            System.out.println(q.poll());  // 4
            System.out.println(q.poll());  // 3
            System.out.println(q.poll());  // 2
            System.out.println(q.poll());  // 1
        }
    }

观察打印结果， 入列：12345， 出列是54321， 也是说出列时他自己做了相关的判断

#### 7 常用阻塞列队的使用 ####

##### 7.1 ArrayBlockingQueue #####

ArrayBlockingQueue是一个有边界的阻塞队列，它的内部实现是一个数组。有边界的意思是它的容量是有限的，我们必须在其初始化的时候指定它的容量大小，容量大小一旦指定就不可改变。ArrayBlockingQueue是以先进先出的方式存储数据

###### ArrayBlockingQueue重要方法： ######

remove():底层是用到了poll()方法，检索并且删除返回队列头的元素,与poll()方法不同的是，元素没有是进行抛异常NoSuchElementException。  
poll(): 检索并且删除返回队列头的元素,有就返回没有就返回null。  
take(): 检索并且删除返回队列头的元素,如果元素没有会一直等待，有就返回。  
peek(): 检索但不移除此队列的头部;如果此队列为空，则返回null。返回头部元素。

###### ArrayBlockingQueue代码示例： ######

package com.lijie;
    
    import java.util.concurrent.ArrayBlockingQueue;
    
    public class Test05 { 
        public static void main(String[] args){ 
            ArrayBlockingQueue<String> arrays = new ArrayBlockingQueue<String>(3);
            arrays.add("李四");
            arrays.add("张三");
            arrays.add("王五");
            arrays.add("小红");
            arrays.add("校长");
            //会抱错，在未消费之前不能添加列队了，列队已经排满，
        }
    }

##### 7.2 LinkedBlockingQueue #####

LinkedBlockingQueue阻塞队列大小的配置是可选的，如果我们初始化时指定一个大小，它就是有边界的，如果不指定，它就是无边界的。说是无边界，其实是采用了默认大小为Integer.MAX\_VALUE的容量 。它的内部实现是一个链表。

和ArrayBlockingQueue一样，LinkedBlockingQueue 也是以先进先出的方式存储数据。

###### LinkedBlockingQueue的重要方法： ######

add(): 在不超出队列长度的情况下插入元素，可以立即执行，成功返回true，如果队列满了就抛出异常。  
offer(): 在不超出队列长度的情况下插入元素的时候则可以立即在队列的尾部插入指定元素,成功时返回true，如果此队列已满，则返回false。  
put()： 插入元素的时候，如果队列满了就进行等待，直到队列可用。  
take(): 从队列中获取值，如果队列中没有值，线程会一直阻塞，直到队列中有值，并且该方法取得了该值。  
poll(long timeout, TimeUnit unit): 在给定的时间里，从队列中获取值，如果没有取到会抛出异常。  
remainingCapacity()：获取队列中剩余的空间。  
remove(Object o): 从队列中移除指定的值。  
contains(Object o): 判断队列中是否拥有该值。  
drainTo(Collection c): 将队列中值，全部移除，并发设置到给定的集合中。

###### LinkedBlockingQueue代码示例： ######

package com.lijie;
    
    import java.util.concurrent.LinkedBlockingQueue;
    
    public class Test06 { 
        public static void main(String[] args){ 
            LinkedBlockingQueue linkedBlockingQueue = new LinkedBlockingQueue(3);
            linkedBlockingQueue.add("张三");
            linkedBlockingQueue.add("李四");
            linkedBlockingQueue.add("王五");
            System.out.println(linkedBlockingQueue.size());
        }
    }

#### 8 使用BlockingQueue模拟生产者与消费者代码示例： ####

package com.lijie;
     
    import java.util.concurrent.ArrayBlockingQueue;
     
    public class BlockingQueueTest { 
     
        //final成员变量表示常量，只能被赋值一次，赋值后值不再改变。
        private static final int queueSize = 5;
        private static final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(queueSize);
        private static final int produceSpeed = 300;//生产速度
        private static final int consumeSpeed = 500;//消费速度
     
        //生产者
        public static class Producer implements Runnable { 
    
            public void run() { 
                while (true) { 
                    try { 
                        System.out.println("老板准备炸油条了，架子上还能放：" + (queueSize - queue.size()) + "根油条");
                        queue.put("1根油条");
                        System.out.println("老板炸好了1根油条，架子上还能放：" + (queueSize - queue.size()) + "根油条");
                        Thread.sleep(produceSpeed);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
     
        //消费者
        public static class Consumer implements Runnable { 
     
            public void run() { 
                while (true) { 
                    try { 
                        System.out.println("A 准备买油条了，架子上还剩" + queue.size() + "根油条");
                        queue.take();
                        System.out.println("A 买到1根油条，架子上还剩" + queue.size() + "根油条");
                        Thread.sleep(consumeSpeed);
     
                        System.out.println("B 准备买油条了，架子上还剩" + queue.size() + "根油条");
                        queue.take();
                        System.out.println("B 买到1根油条，架子上还剩" + queue.size() + "根油条");
                        Thread.sleep(consumeSpeed);
     
                        System.out.println("C 准备买油条了，架子上还剩" + queue.size() + "根油条");
                        queue.take();
                        System.out.println("C 买到1根油条，架子上还剩" + queue.size() + "根油条");
                        Thread.sleep(consumeSpeed);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
     
     
        public static void main(String[] args) { 
            Thread producer = new Thread(new Producer());
            Thread consumer = new Thread(new Consumer());
            producer.start();
            consumer.start();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzEyMjA5MA_size_16_color_FFFFFF_t_70]: /images/20230528/7d6677805b074b618a3880fcbe164fb2.png