阻塞队列的理解和使用

队列和阻塞队列

阻塞队列，顾名思义首先他是一个队列，而一个阻塞队列在数据结构中所起的作用如下图：

image

当阻塞队列为空时，从队列中获取元素的操作将会被阻塞。
当阻塞队列为满时，从队列里添加元素的操作将会被阻塞。

为什么用？ 有什么好处？

在多线程领域：所谓阻塞，在某些情况下会挂起线程（阻塞），一旦条件满足，被挂起的线程又会自动被唤醒。

为什么需要阻塞队列

好处是我们不需要关心什么时候需要阻塞线程，什么时候唤醒线程，因为这一切BlockingQueue都给你一手包办了。

阻塞队列的种类分析

• ArrayBlockingQueue：由数组结构组成的有界阻塞队列，
• LinkedBlockingQueue：由链表结构组成的有界（但是默认值大小为Integer,MAX_VALUE）阻塞队列。
• PriorityBlockingQueue：支持优先级排序的无界阻塞队列。
• DelayQueue：使用优先级队列支持延迟的无界阻塞队列。
• SynchronousQueue：不存储元素的队列，也即单个元素的队列。没有容量，每一个put操作必须等待一个take操作，否则不能添加元素，反之亦然。
• LinkedTransferQueue：由链表结构组成的无界阻塞队列。
• LinkedBlockingDeque：由链表结构组成的双向阻塞队列。

常用方法

在这里插入图片描述

最佳实践

使用阻塞队列实现消息的生产和消费

class MyShare {
    private volatile boolean FLAG = true;//默认开启，进行生产和消费
    AtomicInteger atomicInteger = new AtomicInteger();
    BlockingQueue<String> blockingQueue = null;

    public MyShare(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void myProd() throws Exception {
        String data = null;
        boolean result;
        while (FLAG) {
            data = String.valueOf(atomicInteger.incrementAndGet());
            result = blockingQueue.offer(data, 2L, TimeUnit.MICROSECONDS);
            if (result) {
                System.out.println(Thread.currentThread().getName() + "\t 插入队列data:" + data + "成功");
            } else {
                System.out.println(Thread.currentThread().getName() + "\t 插入队列data:" + data + "失败");
            }
            //一秒生产一个
            try {TimeUnit.SECONDS.sleep(1);}catch (InterruptedException e){e.printStackTrace();}


        }
        System.out.println(Thread.currentThread().getName() + "\t 生产被叫停，Flag为false");

    }

    public void myConsumer() throws Exception {
        String value = null;
        while (FLAG) {
            value = blockingQueue.poll(2, TimeUnit.SECONDS);
            if (value == null || value.equalsIgnoreCase("")) {
//                FLAG = false;
                System.out.println(Thread.currentThread().getName() + "\t 超过两秒没收到消息");
                System.out.println("");
                System.out.println("");
                System.out.println("");
                return;
            }
            System.out.println(Thread.currentThread().getName() + "\t 消费队列value：" + value + "成功");
        }

    }

    public void stop() {
        FLAG = false;
    }
}

/**
 * @author liujian
 * @descripts 生产者消费者阻塞队列版
 * @create 2019-06-24 22:43
 */
public class ProductConsumer_BlockingQueueDemo {
    public static void main(String[] args) {
        MyShare myShare = new MyShare(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            System.out.println("生产线程启动");
            try {
                myShare.myProd();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, "product").start();

        new Thread(() -> {
            System.out.println("消费线程启动");
            try {
                System.out.println("");
                System.out.println("");
                System.out.println("");
                myShare.myConsumer();
            } catch (Exception e) {
                e.printStackTrace();
            }

        }, "consumer").start();
        //休眠五秒
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        myShare.stop();
    }


}