美文网首页
JUC之阻塞队列

JUC之阻塞队列

作者: 西界__ | 来源:发表于2021-01-03 11:32 被阅读0次

阻塞队列

概念

阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。

试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列。

下图展示了如何通过阻塞队列来合作：

线程1往阻塞队列中添加元素，而线程2从阻塞队列中移除元素

意义

在多线程领域:所谓阻塞,在某些情况下会挂起 线程(即线程阻塞),一旦条件满足,被挂起的线程优惠被自动唤醒

为什么需要使用BlockingQueue？

好处是我们不需要关心什么时候需要阻塞线程,什么时候需要唤醒线程,因为BlockingQueue都一手给你包办好了。在concurrent包 发布以前,在多线程环境下,我们每个程序员都必须自己去控制这些细节,尤其还要兼顾效率和线程安全, 而这会给我们的程序带来不小的复杂度.

Java的阻塞队列

  • <mark>ArrayBlockingQueue: 由数组结构组成的有界阻塞队列</mark>
  • <mark>LinkedBlockingDeque: 由链表结构组成的有界(但大小默认值为Integer.MAX_VALUE)阻塞队列.</mark>
  • PriorityBlockingQueue:支持优先级排序的无界阻塞队列.
  • DelayQueue: 使用优先级队列实现的延迟无界阻塞队列.
  • <mark>SynchronousQueue:不存储元素的阻塞队列,也即是单个元素的队列.</mark>
  • LinkedTransferQueue:由链表结构组成的无界阻塞队列.
  • LinkedBlockingDeque:由了解结构组成的双向阻塞队列.

阻塞队列的核心方法

抛出异常 返回特殊值 一直阻塞 超时退出
插入方法 add(o) offer(o) put(o) offer(o,timeout, timeunit)
移除方法 remove(o) poll() take(o) poll(o,timeout, timeunit)
检查方法 element() peek() — —
方法效果 成功/队满 失败/队空
抛出异常 当阻塞队列满时，再往队列里add插入元素会抛IllegalStateException:Queue full 当阻塞队列空时，再往队列里remove移除元素会抛NoSuchElementException
返回特殊值 插入方法，成功ture失败false 移除方法，成功返回出队列的元素，队列里没有就返回null
一直阻塞 当阻塞队列满时，生产者线程继续往队列里put元素，队列会一直阻塞生产者线程直到put数据or响应中断退出 当阻塞队列空时，消费者线程试图从队列里take元素，队列会一直阻塞消费者线程直到队列可用
超时退出 当阻塞队列满时，队列会阻塞生产者线程一定时间，超过限时后生产者线程会退出

抛出异常

使用add()向阻塞队列添加数据，成功返回true。

当向队列满的时候抛出异常java.lang.IllegalStateException: Queue full。

使用remove()方法移除阻塞队列中的元素，从队列头先开始移除，移除成功后返回。

当队列为空时进行移除操作，抛出java.util.NoSuchElementException异常

使用element()返回当前阻塞队列的头元素。

当阻塞队列为空的时候使用抛出java.util.NoSuchElementException异常

/**
 * 阻塞队列
 */
public class BlockingQueueDemo {

    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        System.out.println(blockingQueue.add("a"));
        System.out.println(blockingQueue.add("b"));
        System.out.println(blockingQueue.add("c"));
        //java.lang.IllegalStateException: Queue full
        //System.out.println(blockingQueue.add("d"));
        //返回当前队列头元素
        System.out.println(blockingQueue.element());

        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        //java.util.NoSuchElementException
        //System.out.println(blockingQueue.remove());
        //java.util.NoSuchElementException
        System.out.println(blockingQueue.element());
    }
}

返回特殊值

使用offer()方法向阻塞队列中添加值，添加成功返回true

当队列满时继续添加，返回false而不是抛出异常

使用poll()方法向阻塞队列中移除元素，并返回。

队列为空时移除，返回null

使用peek()方法查看当前队列头元素。队列为空时返回null

public class BlockingQueueDemo {
    
    public static void main(String[] args) {

        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        System.out.println(blockingQueue.offer("a"));
        System.out.println(blockingQueue.offer("b"));
        System.out.println(blockingQueue.offer("c"));
        //队列满
        System.out.println(blockingQueue.offer("d"));//false
        //查看当前队列头元素
        System.out.println(blockingQueue.peek());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        //队列空
        System.out.println(blockingQueue.poll());//null
        System.out.println(blockingQueue.peek());//null
    }
}

一直阻塞

使用put()方法向阻塞队列中添加元素，添加成功没有返回值，当队列满时，一直阻塞直到添加成功或者被中断才能停止。

使用take()方法向阻塞队列中移除元素，移除成功返回值，当队列为空时，一直阻塞直到再次移除成功或者被中断才能停止。

public class BlockingQueueDemo {
    
    public static void main(String[] args) throws InterruptedException {

        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        System.out.println("全部添加完成！");
        //队列满
        //blockingQueue.put("d");//false

        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        //队列为空
        System.out.println(blockingQueue.take());
    }

超时退出

添加使用的是offer的重载方法offer(E e, long timeout, TimeUnit unit)

移除使用的是poll的重载方法poll(E e, long timeout, TimeUnit unit)

两者都是在队列满，队列空是添加，移除元素等待给定的时间，如果在给定的时间中添加成功返回ture失败返回false，移除成功则返回元素，失败返回null。

public class BlockingQueueDemo {

    public static void main(String[] args) throws InterruptedException {

        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        System.out.println("全部添加完成！");
        //队列满 等待一秒
        System.out.println(blockingQueue.offer("d", 1l, TimeUnit.SECONDS));

        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        //队列为空 等待一秒
        System.out.println(blockingQueue.poll(1l, TimeUnit.SECONDS));
    }

}

SynchronousQueue

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

只有B把元素取走，A才能继续向队列终添加元素，否则一直阻塞。

public class SynchronousQueueDemo {

    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>();

        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName()+"\t put1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+"\t put2");
                blockingQueue.put("2");
                System.out.println(Thread.currentThread().getName()+"\t put3");
                blockingQueue.put("3");

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"AA").start();

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"\t "+blockingQueue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"\t "+blockingQueue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"\t "+blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"BB").start();
    }
}

阻塞队列版生产者消费者

/**
 * volatile/CAS/atomicInteger/BlockQueue/线程交互/原子引用
 */
public class ProdConsumer_BlockQueueDemo {

    public static void main(String[] args) {
        MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            System.out.println(Thread.currentThread().getName()+"\t 生产线程启动");
            try {
                myResource.myProd();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"Prod").start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName()+"\t 消费线程启动");
            try {
                myResource.myConsumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"Consumer").start();

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println();
        System.out.println();
        System.out.println("5秒钟到，大老板叫停！");
        try {
            myResource.stop();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

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

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

    public void myProd() throws Exception {
        String data = null;
        boolean retValue;
        while (FLAG) {
            data = atomicInteger.incrementAndGet() + "";
            retValue = blockingQueue.offer(data, 2l, TimeUnit.SECONDS);
            if (retValue) {
                System.out.println(Thread.currentThread().getName() + "\t 插入队列" + data + "成功");
            } else {
                System.out.println(Thread.currentThread().getName() + "\t 插入队列" + data + "失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName()+"\t大老板叫停了，表示FLAG=false，生产动作结束");

    }
    public void myConsumer() throws Exception{
        String result = null;
        while (FLAG){
            result = blockingQueue.poll(2L,TimeUnit.SECONDS);
            if (null==result||result.equalsIgnoreCase("")){
                System.out.println(Thread.currentThread().getName()+"\t 超过2秒钟没有取到蛋糕，消费退出");
                System.out.println();
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName()+"\t 消费队列蛋糕"+result+"成功");
        }
    }
    public void stop() throws Exception{
        this.FLAG = false;
    }
}

相关文章

  • JUC之阻塞队列

    阻塞队列 概念 阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，...

  • JUC 阻塞队列

    概述 阻塞队列解决的问题：在一个容量有限的仓库里面，实现满了就挂起生产线程，空了就挂起消费线程的兼顾性能和安全的数...

  • JUC--阻塞队列

    2018-10-03 原文推荐 死磕Java并发

  • JUC-阻塞队列

      阻塞队列数据结构示意图如下所示： 当阻塞队列是空时，从队列中获取元素的操作将会被阻塞； 当阻塞队列是满时，往队...

  • JUC下的阻塞队列-PriorityBlockingQueue

        PriorityBlockingQueue是一个支持优先级的无界阻塞队列，基于数组的二叉堆，其实就是线程安...

  • JUC下的阻塞队列-ArrayBlockingQueue

        ArrayBlockingQueue是数组实现的线程安全的有界的阻塞队列。下面是主要结构。    大概思想...

  • JUC下的阻塞队列-LinkedBlockingQueue

        LinkedBlockingQueue是一个单向链表实现的阻塞队列。该队列按 FIFO（先进先出）排序元素...

  • JUC下的阻塞队列-DelayQueue

        DelayQueue是一个无界阻塞队列，可以用做延时处理，延时处理就是可以为队列中元素设定一个过期时间，队...

  • JUC之队列

    1.队列 1.1 类图 1.2 核心接口 BlockingQueue 阻塞队列 添加区别boolean add(E...

  • 并发编程之并发队列

    常见的并发队列有2种：阻塞队列和非阻塞队列。阻塞队列使用锁实现，非阻塞队列使用CAS非阻塞算法实现。这2种队列都是...

网友评论

      本文标题：JUC之阻塞队列

      本文链接：https://www.haomeiwen.com/subject/goddnktx.html

      延伸阅读

      深度阅读

      • 您也可以注册成为美文阅读网的作者，发表您的原创作品、分享您的心情！

      栏目导航

      热点阅读

      关于我们|服务条款|联系我们|JUC之阻塞队列|投稿指南|网站地图|RSS订阅|排版工具|手机版

      提供经典美文摘抄,优美散文欣赏,现代诗歌精选,短篇小说,心情随笔,表白情书范文,故事会在线阅读欣赏

      Copyright © 2014-2023 Haomeiwen.com All Rights Reserved. 好美文阅读网 版权所有

      备案信息：桂公网安备 45052102000051号 · 桂ICP备13007215号-3

      本站所收录作品、热点评论等信息部分来源互联网，目的只是为了系统归纳学习和传递资讯

      所有作品版权归原创作者所有，与本站立场无关，如不慎侵犯了你的权益，请联系我们告知，我们将做删除处理！