Java队列Queue

蔚落 2021-08-24 17:01 504阅读 0赞

**欲速则不达，欲达则欲速！**

队列Queue是一个先进先出的数据结构；与list、set同一级别，继承了collection接口。

Queue的实现

*  阻塞队列（BolckingQueue）
 *  非阻塞队列

**阻塞队列（BolckingQueue）**

一、插入与移除

1、插入

队列不满时可执行插入元素线程，直到队列满。

2、移除

队列不为空时都可移除，直到队列为空。

![format_png][]

抛出异常：满的时候插入，空的时候取出都会抛出异常。

返回特殊值：插入成功返回true

一直阻塞：满时put和空时take会阻塞线程，直到队列可用。

二、JDK7提供了7个阻塞队列：

1、ArrayListBlockingQueue

ArrayBlockingQueue是一个由数组支持的有界阻塞队列。在读写操作上都需要锁住整个容器，因此吞吐量与一般的实现是相似的，适合于实现“生产者消费者”模式。

2、LinkedBlockingQueue

一个由链表结构组成的双向阻塞队列。

基于链表的阻塞队列，内部维持着数据缓冲队列。当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；当队列缓冲区达到最大缓冲容量时，才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生成者线程才会被唤醒，反之对消费者的处理也基于同样的原理。

LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者和消费者分别采用了独立的锁来控制数据同步，也就意味着在高并发的情况下生产者和消费者可以并行的操作队列中的数据，以此来提高整个队列的并发性能。

3、SynchronousQueue

一个不存储元素的阻塞队列，在某次添加元素后必须等待其他线程取走后才能继续添加。

4、 PriorityBlockingQueue

是一个带优先级的队列，而不是先进先出队列，该队列也没有上限，但是如果队列为空，那么取元素的操作take就会阻塞。

5、DelayQueue

是一个存放Delayed 元素的无界阻塞队列，只有在延迟期满时才能从中提取元素。

三、ArrayListBlockingQueue和LinkedBlockingQueue的区别？

1、队列中锁的实现不同

ArrayBlockingQueue生产者消费者使用同一个锁。

LinkedBlockingQueue生产用的是putLock，消费是takeLock。

2、在生产和消费时操作不同

ArrayBlockingQueue实现的队列中在生产和消费的时候，是直接将枚举对象插入或移除的；

LinkedBlockingQueue实现的队列中在生产和消费的时候，需要把枚举对象转换为节点进行插入或移除，会影响性能。

3、 队列大小初始化方式不同

ArrayBlockingQueue实现的队列必须指定大小

LinkedBlockingQueue可以不指定大小，默认是Integer.MAX\_VALUE

ArrayBlockingQueue性能要比LinkedBlockingQueue性能要好，执行速度更快，ArrayBlockingQueue优先使用！

**非阻塞队列**

ConcurrentLinkedQueue是一个基于链接节点的无界线程安全队列，它采用先进先出的规则对节点进行排序，当我们添加一个元素的时候，它会添加到队列的尾部；当我们获取一个元素时，它会返回队列头部的元素。

入队和出队都采用CAS（compare and set）更新，这样允许多个线程并发执行，并且不会因为加锁而阻塞线程，使得并发性能更好。

注：CAS用于实现多线程同步的原子指令，它将内存位置的内容与给定值进行比较，只有在相同的情况下，将该内存位置的内容修改为新的给定值。

**代码实例**

package OSChina.Client;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class BlockingQueueTest {
        static final SimpleDateFormat sdf = new SimpleDateFormat("HH:mm:ss");
        //定义装苹果的篮子
        public static class Basket{
            // 篮子，能够容纳3个苹果
           static BlockingQueue<String> queue = new ArrayBlockingQueue<String>(3);
            //生产苹果，放入篮子
            public void produce() throws InterruptedException{
                queue.put("An apple");
            }
            // 消费苹果，从篮子中取走
            public String consume() throws InterruptedException{
                String apple = queue.take();
                return apple;
            }
            public static int getAppleNumber(){
                return queue.size();
            }
        }
    
        public static void testBasket(){
            // 建立一个装苹果的篮子
            final Basket basket = new Basket();
            //定义苹果生产者
            class Producer implements Runnable{
                @Override
                public void run() {
                    try {
                        while (true){
                            System.out.println("生产者准备生产苹果："+sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                            basket.produce();
                            System.out.println("生产者生产苹果完毕：" + sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                            Thread.sleep(300);
                        }
                    }catch (InterruptedException  ex){
                    }
                }
            }
            //定义苹果消费者
            class Consumer implements Runnable{
                @Override
                public void run() {
                    try {
                        while (true){
                            System.out.println("消费者准备消费苹果：" + sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                            basket.consume();
                            System.out.println("消费完后有苹果：" + sdf.format(new Date())+"，篮子中苹果数量："+Basket.getAppleNumber());
                            Thread.sleep(1000);
                        }
                    }catch (InterruptedException ex){
                    }
                }
            }
    
            ExecutorService service = Executors.newFixedThreadPool(2);
            Producer producer = new Producer();
            Consumer consumer = new Consumer();
            service.submit(producer);
            service.submit(consumer);
            // 程序运行10s后，所有任务停止
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            service.shutdownNow();
        }
    
        public static void main(String[] args) {
            BlockingQueueTest.testBasket();
        }
    }

![format_png 1][]

[format_png]: /images/20210813/07b4cd9e62ef4f4b8af3f8b3c02f6104.png
[format_png 1]: /images/20210813/cd7e2a08c3a644fc8377fe0389d7eebc.png