JUC学习系列一（阻塞队列BlockingQueue）

2019年3月10日 275次阅读来源: JUC

阻塞队列 BlockingQueue

java.util.concurrent 包里的 BlockingQueue 接口表示一个线程安放入和提取实例的队列。本小节我将给你演示如何使用这个 BlockingQueue。

BlockingQueue 用法

BlockingQueue 通常用于一个线程生产对象，而另外一个线程消费这些对象的场景。下图是对这个原理的阐述：

《JUC学习系列一（阻塞队列BlockingQueue）》

一个线程往里边放，另外一个线程从里边取的一个 BlockingQueue。

一个线程将会持续生产新对象并将其插入到队列之中，直到队列达到它所能容纳的临界点。也就是说，它是有限的。如果该阻塞队列到达了其临界点，负责生产的线程将会在往里边插入新对象时发生阻塞。它会一直处于阻塞之中，直到负责消费的线程从队列中拿走一个对象。

负责消费的线程将会一直从该阻塞队列中拿出对象。如果消费线程尝试去从一个空的队列中提取对象的话，这个消费线程将会处于阻塞之中，直到一个生产线程把一个对象丢进队列。

BlockingQueue 的方法

BlockingQueue 具有 4 组不同的方法用于插入、移除以及对队列中的元素进行检查。如果请求的操作不能得到立即执行的话，每个方法的表现也不同。这些方法如下：

	抛异常	特定值	阻塞	超时
插入	add(o)	offer(o)	put(o)	offer(o, timeout, timeunit)
移除	remove(o)	poll(o)	take(o)	poll(timeout, timeunit)
检查	element(o)	peek(o)

四组不同的行为方式解释：

抛异常：如果试图的操作无法立即执行，抛一个异常。
特定值：如果试图的操作无法立即执行，返回一个特定的值(常常是 true / false)。
阻塞：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行。
超时：如果试图的操作无法立即执行，该方法调用将会发生阻塞，直到能够执行，但等待时间不会超过给定值。返回一个特定值以告知该操作是否成功(典型的是 true / false)。

无法向一个 BlockingQueue 中插入 null。如果你试图插入 null，BlockingQueue 将会抛出一个 NullPointerException。

可以访问到 BlockingQueue 中的所有元素，而不仅仅是开始和结束的元素。比如说，你将一个对象放入队列之中以等待处理，但你的应用想要将其取消掉。那么你可以调用诸如 remove(o) 方法来将队列之中的特定对象进行移除。但是这么干效率并不高(译者注：基于队列的数据结构，获取除开始或结束位置的其他对象的效率不会太高)，因此你尽量不要用这一类的方法，除非你确实不得不那么做。

BlockingQueue 的实现

BlockingQueue 是个接口，你需要使用它的实现之一来使用 BlockingQueue。java.util.concurrent 具有以下 BlockingQueue 接口的实现(Java 6)：

ArrayBlockingQueue
DelayQueue
LinkedBlockingQueue
PriorityBlockingQueue
SynchronousQueue

《JUC学习系列一（阻塞队列BlockingQueue）》

一、数组阻塞队列ArrayBlockingQueue

同过一个简单的生产消费模型，可以看出效果。当队列里的存放到达最大容量是，生产者会阻塞着，得消费者取走消息。容器里面有新的空间后，继续存放。效果复制以下代码，执行main方法就行。

生产者：

import java.util.concurrent.BlockingQueue;
/**
 * @Author Loujitao
 * @Date 2018/6/27
 * @Time  17:42
 * @Description:
 */
public class Producer implements Runnable{

    private BlockingQueue MQList=null;

    public Producer(BlockingQueue MQList) {
        this.MQList = MQList;
    }

    @Override
    public void run() {
        try {
            for (int a=1;a<10;a++){
                //这里我让线程休眠0.5s，因为从数据库查询数据或者操作数据都是耗时的
                Thread.sleep(500);
                String str=Thread.currentThread().getName();
                MQList.put(str+"的"+a);
                System.out.println("生产者"+str+"存放了"+a);
            }
            System.out.println("生产者生产完成消息了");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消费者：

import java.util.concurrent.BlockingQueue;

public class Cunsumer implements Runnable{

    private BlockingQueue MQList=null;

    public Cunsumer(BlockingQueue MQList) {
        this.MQList = MQList;
    }

    @Override
    public void run() {
        try {
            while (true){
                Thread.sleep(2000);
            System.out.println("消费者"+Thread.currentThread().getName()+"取得消息："+MQList.take());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

执行代码：

package com.steve.concurrentCollection;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueTest {

    public static void main(String[] args) {
        function01();
    }

    /**
     * @Author Loujitao
     * @Date 2018/6/27
     * @Time  17:44
     * @Description: ArrayBlockingQueue的底层使用的是ReentrantLock重入锁
     *   而且是默认false，然后走sync = fair ? new FairSync() : new NonfairSync();的非公平锁
     */
    public static void function01(){
        BlockingQueue bq=new ArrayBlockingQueue(5);
        //producer是一个线程任务类，执行的方法是往队列里放10条数据
        Producer producer=new Producer(bq);
        //consumer是线程任务类，执行方法是往队列里拿数据   他两的队列是同一个队列，
        Cunsumer cunsumer=new Cunsumer(bq);

        new Thread(producer).start();
        new Thread(cunsumer).start();
    }


}

结果：《JUC学习系列一（阻塞队列BlockingQueue）》

总结：这个实例中生产者等待的时间是500ms，消费者2000ms；是消费慢生产快的情况。执行的过程中，我发现生产者很快就存放玩了五条数据，而消费者还在消费第一条数据，但是队列已经满了，所以此时生产者阻塞在这里，等待队列有空余位置。然后的情况就是，消费者消费掉一条数据，生产者立马就放入数据。直到九条数据存放完毕，生产者的线程终止，只剩下消费者慢慢消费。生产和消费的数据都是安全的。

只要将两个数据调换，就能模拟出生产慢、消费快的情景。

多生产者、单消费者情景

public static void function01(){
        BlockingQueue bq=new ArrayBlockingQueue(5);
        //producer是一个线程任务类，执行的方法是往队列里放10条数据
        Producer producer=new Producer(bq);
        Producer producer2=new Producer(bq);
        //consumer是线程任务类，执行方法是往队列里拿数据   他两的队列是同一个队列，
        Cunsumer cunsumer=new Cunsumer(bq);

        new Thread(producer,"pro1").start();
        new Thread(producer2,"pro2").start();
        new Thread(cunsumer).start();
    }

模拟了一下这种情况，数据是安全的。
多生产多消费者的情况，只要在新new出相应的对象，放到线程中执行即可，感兴趣的可以尝试一下。

实际生产中可能应用的场景：（本人还没碰到，一些中间件底层可能用到，但没时间看源码，今后再做补充吧）

二、链式阻塞队列（LinkedBlockingQueue）

public static void function02(){
        BlockingQueue bq=new LinkedBlockingQueue();
        //producer是一个线程任务类，执行的方法是往队列里放10条数据
        Producer producer=new Producer(bq);
        Producer producer2=new Producer(bq);
        //consumer是线程任务类，执行方法是往队列里拿数据   他两的队列是同一个队列，
        Cunsumer cunsumer=new Cunsumer(bq);

        new Thread(producer,"pro1").start();
//        new Thread(producer2,"pro2").start();
        new Thread(cunsumer).start();
    }

还是使用以上的
生产-消费模型，只是我这次创建的队列是无界的链式队列。运行的结果如下：

《JUC学习系列一（阻塞队列BlockingQueue）》

总结：因为使用的队列是无界的，它能容纳的个数是Integer.MAX_VALUE，整型的最大值。所以不存在生产者没地方存放消息，而需要阻塞的情况。生产者依次存放消息，存满了线程结束；消费者慢慢消费，直到所有的情况消费完毕。这种情况是生产快、消费慢的情况。

如果你将500ms和2000ms调换一下，即消费快、生产慢的时候，就会发现消费者一直队列中一有消息进来，就被消费者消费掉，没有消息就一直阻塞等待着。

像多生产者和多消费者这些情况，同上一个队列，可以实例化多个实例，模拟下不同场景，但是数据一定是安全的。

两个生产者一个消费者，在消费快的情况下：

《JUC学习系列一（阻塞队列BlockingQueue）》

实际生产中可能应用的场景：（待续）

三、优先级队列（PriorityBlockingQueue）

带优先级的无界阻塞队列，每次出队都返回优先级最高的元素，是二叉树最小堆的实现，研究过数组方式存放最小堆节点的都知道，直接遍历队列元素是无序的。该实现类需要自己实现一个继承了 Comparator 接口的类，在插入资源时会按照自定义的排序规则来对资源数组进行排序。

示例：

比较元素类

public class MyTask implements Comparable<MyTask> {

    private int money;
    private String name;
    @Override
    public int compareTo(MyTask o) {
        return this.money>o.money?-1:(this.money==o.money?0:1);
    }
    public int getMoney() {
        return money;
    }
    public void setMoney(int money) {
        this.money = money;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }
    @Override
    public String toString() {
        return "money=" + money +
                ",name=" + name +" ";
    }
}

测试类：

public class PriorityBlockingQueueTest {
    
    /**   
     * @Author Loujitao  
     * @Date 2018/7/2
     * @Time  9:58  
     * @Description: https://blog.csdn.net/qq_38872310
     */  
    public static void main(String[] args)throws Exception {
        BlockingQueue<MyTask> bq=new PriorityBlockingQueue<MyTask>();

        MyTask t0=new MyTask();
        t0.setMoney(0);
        t0.setName("VIP LV0");
        MyTask t1=new MyTask();
        t1.setMoney(1);
        t1.setName("VIP LV1");
        MyTask t2=new MyTask();
        t2.setMoney(2);
        t2.setName("VIP LV2");
        MyTask t3=new MyTask();
        t3.setMoney(3);
        t3.setName("VIP LV3");
        //这里我特意打乱了顺序
        bq.add(t2);
        bq.add(t1);
        bq.add(t3);
        bq.add(t0);
        //优先级最高的先执行了，这里t3的VIP等级高先走了。（社会你懂的）
        //优先级队列，只有调用一次take()方法，才会进行排序
        bq.take();//System.out.println(bq.take());
        System.out.println(bq);
    }

}

执行结果：

《JUC学习系列一（阻塞队列BlockingQueue）》

总结：bq.take()的返回值是你队列里存放的对象，这里的是mytask。可以用System.out.println(bq.take())查看。

PriorityBlockingQueue始终保证出队的元素是优先级最高的元素，并且可以定制优先级的规则，内部通过使用一个二叉树最小堆算法来维护内部数组，这个数组是可扩容的，当当前元素个数>=最大容量时候会通过算法扩容。值得注意的是为了避免在扩容操作时候其他线程不能进行出队操作，实现上使用了先释放锁，然后通过cas保证同时只有一个线程可以扩容成功。

如果想深入探究，可以了解下二叉树的数据结构，自己看下这个队列的源码。

使用场景：

四、延迟队列（ DelayQueue）

DelayQueue是一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部是延迟期满后保存时间最长的Delayed 元素。为了具有调用行为，存放到DelayDeque的元素必须继承Delayed接口。Delayed接口使对象成为延迟对象，它使存放在DelayQueue类中的对象具有了激活日期。该接口强制执行下列两个方法。

CompareTo(Delayed o)：Delayed接口继承了Comparable接口，因此有了这个方法。
getDelay(TimeUnit unit):这个方法返回到激活日期的剩余时间，时间单位由单位参数指定

示例：

用户对象

public class LOLGamer implements Delayed {
    private String name;
    //上机座位号
    private String id;
    //截止时间
    private long endTime;
    //定义时间工具类
    private TimeUnit timeUnit = TimeUnit.SECONDS;
    public LOLGamer(String name, String id, long endTime) {
        this.name = name;
        this.id = id;
        this.endTime = endTime;
    }
    @Override
    public int compareTo(Delayed o) {
        LOLGamer lg=(LOLGamer) o;
        return this.getDelay(this.timeUnit)-o.getDelay(this.timeUnit)>0?1:0;
    }
    @Override
    public long getDelay(TimeUnit unit) {
        return endTime-System.currentTimeMillis();
    }

    public String getName() {        return name;    }
    public void setName(String name) {        this.name = name;    }
    public String getId() {        return id;    }
    public void setId(String id) {        this.id = id;    }
    public long getEndTime() {        return endTime;    }
    public void setEndTime(long endTime) {        this.endTime = endTime;    }
}

执行代码

public class GameBar implements  Runnable{

    private DelayQueue<LOLGamer> queue = new DelayQueue<LOLGamer>();
    public boolean yinye =true;

    public void shangji(String name,String id,int money){
        //1000 * money + System.currentTimeMillis()  表示从当前开始上1000ms，截止下机时间就是这个和
        LOLGamer man = new LOLGamer(name, id, 1000 * money + System.currentTimeMillis());
        System.out.println("玩家"+man.getName()+" 机位号"+man.getId()+"交钱"+money+"块,开始上机...");
        //上机同时将消费记录存到延迟队列里
        this.queue.add(man);
    }
    public void xiaji(LOLGamer man){
        System.out.println("玩家"+man.getName()+" 机位号"+man.getId()+"时间到下机...");
    }
    @Override
    public void run() {
        while(yinye){
            try {
                //延迟时间到了，才能获取数据  只有有数据才能取到
                LOLGamer man = queue.take();
                //时间到了的墙纸下机了
                xiaji(man);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    public static void main(String args[]){
        try{
            System.out.println("网鱼网吧欢迎你");
            GameBar wangyu = new GameBar();
            Thread shangwang = new Thread(wangyu);
            shangwang.start();

            wangyu.shangji("张三", "A001", 1);
            wangyu.shangji("李四", "A002", 10);
            wangyu.shangji("王五", "A003", 5);
        }
        catch(Exception e){
            e.printStackTrace();
        }
    }

}

输出：

总结：DelayQueue类的主要作用：是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。这种队列是有序的，即队头对象的延迟到期时间最长。注意：不能将null元素放置到这种队列中。

Delayed，一种混合风格的接口，用来标记那些应该在给定延迟时间之后执行的对象。此接口的实现必须定义一个 compareTo 方法，该方法提供与此接口的 getDelay 方法一致的排序。

应用场景：　DelayQueue阻塞队列在我们系统开发中也常常会用到，例如：缓存系统的设计，缓存中的对象，超过了空闲时间，需要从缓存中移出；任务调度系统，能够准确的把握任务的执行时间。

多考试上机考试的场景：来自于http://ideasforjava.iteye.com/blog/657384，模拟一个考试的日子，考试时间为120分钟，30分钟后才可交卷，当时间到了，或学生都交完卷了考试结束。

具有时间的缓存场景：来自于http://www.cnblogs.com/jobs/archive/2007/04/27/730255.html，向缓存添加内容时，给每一个key设定过期时间，系统自动将超过过期时间的key清除。

五、同步队列（SynchronousQueue）

特点：1、SynchronousQueue没有容量。与其他BlockingQueue不同，SynchronousQueue是一个不存储元素的BlockingQueue.每一个put操作必须要等待一个take操作，否则不能继续添加元素，反之亦然。

2、因为没有容量，所以对应 peek, contains, clear, isEmpty … 等方法其实是无效的。例如clear是不执行任何操作的，contains始终返回false,peek始终返回null。
3、SynchronousQueue分为公平和非公平，默认情况下采用非公平性访问策略，当然也可以通过构造函数来设置为公平性访问策略（为true即可）。

4、若使用 TransferQueue, 则队列中永远会存在一个 dummy node（这点后面详细阐述）。

示例：

/**   
     * @Author Loujitao  
     * @Date 2018/7/2
     * @Time  10:45  
     * @Description:
     * 不像ArrayBlockingQueue、LinkedBlockingDeque之类的阻塞队列依赖AQS实现并发操作，
     * SynchronousQueue直接使用CAS实现线程的安全访问。它是个没有容量的队列，
     */  
    public static void main(String[] args) throws Exception {
       final BlockingQueue<String> bq=new SynchronousQueue<>();
        //第一种情况 这种直接添加，会出异常Queue full
        //bq.add("steve");
        
        //第二种情况  如果有线程先取，会阻塞着，这样就能放进去
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    System.out.println( bq.take());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
        Thread.sleep(1000);
        //可以理解为交接给了等待的线程,不是真正的存放进去了
        bq.put("stevetao");
    }

结果截图：

《JUC学习系列一（阻塞队列BlockingQueue）》

总结：SynchronousQueue由于其独有的线程一一配对通信机制，在大部分平常开发中，可能都不太会用到，但线程池技术中会有所使用，由于内部没有使用AQS，而是直接使用CAS，所以代码理解起来会比较困难。这里只是做了简介和使用，具体底层实现原理，我觉得有必要在搞清CAS和此队列源码后，单独写一篇博客以作说明。

应用场景：SynchronousQueue非常适合做交换工作，生产者的线程和消费者的线程同步以传递某些信息、事件或者任务。

双向链表阻塞队列简单了解了一下，这里就不写了。

    原文作者：JUC
    原文地址: https://blog.csdn.net/qq_38872310/article/details/80832703
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。