文档章节

ArrayBlockingQueue与LinkedBlockingQueue

go4it
 go4it
发布于 2017/09/10 13:52
字数 1268
阅读 5
收藏 0
点赞 0
评论 0

本文主要简单介绍下ArrayBlockingQueue与LinkedBlockingQueue。

对比

queue阻塞与否是否有界线程安全保障适用场景注意事项
ArrayBlockingQueue阻塞有界一把全局锁生产消费模型,平衡两边处理速度用于存储队列元素的存储空间是预先分配的,使用过程中内存开销较小(无须动态申请存储空间)
LinkedBlockingQueue阻塞可配置存取采用2把锁生产消费模型,平衡两边处理速度无界的时候注意内存溢出问题,用于存储队列元素的存储空间是在其使用过程中动态分配的,因此它可能会增加JVM垃圾回收的负担。
ConcurrentLinkedQueue非阻塞无界CAS对全局的集合进行操作的场景size() 是要遍历一遍集合,慎用

内存方面

  • ArrayBlockingQueue 用于存储队列元素的存储空间是预先分配的,使用过程中内存开销较小(无须动态申请存储空间)
  • LinkedBlockingQueue 用于存储队列元素的存储空间是在其使用过程中动态分配的,因此它可能会增加JVM垃圾回收的负担。

有界无界

  • ArrayBlockingQueue 有界,适合已知最大存储容量的场景
  • LinkedBlockingQueue 可有界可以无界

吞吐量

LinkedBlockingQueue在大多数并发的场景下吞吐量比ArrayBlockingQueue,但是性能不稳定。

Linked queues typically have higher throughput than array-based queues but less predictable performance in most concurrent applications.

输入图片说明

测试结果表明,LinkedBlockingQueue的可伸缩性要高于ArrayBlockingQueue。初看起来,这个结果有些奇怪:链表队列在每次插入元素时,都必须分配一个链表节点对象,这似乎比基于数组的队列执行了更多的工作。然而,虽然它拥有更好的内存分配与GC等开销,但与基于数组的队列相比,链表队列的put和take等方法支持并发性更高的访问,因为一些优化后的链接队列算法能将队列头节点的更新操作与尾节点的更新操作分离开来。由于内存分配操作通常是线程本地的,因此如果算法能通过多执行一些内存分配操作来降低竞争程度,那么这种算法通常具有更高的可伸缩性。

并发方面

  • ArrayBlockingQueue 采用一把锁,两个condition
    /** Main lock guarding all access */
    final ReentrantLock lock;

    /** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;
    
    /**
     * Inserts element at current put position, advances, and signals.
     * Call only when holding lock.
     */
    private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        notEmpty.signal();
    }

    /**
     * Extracts element at current take position, advances, and signals.
     * Call only when holding lock.
     */
    private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        notFull.signal();
        return x;
    }

此外还支持公平锁

/**
     * Creates an {@code ArrayBlockingQueue} with the given (fixed)
     * capacity and the specified access policy.
     *
     * @param capacity the capacity of this queue
     * @param fair if {@code true} then queue accesses for threads blocked
     *        on insertion or removal, are processed in FIFO order;
     *        if {@code false} the access order is unspecified.
     * @throws IllegalArgumentException if {@code capacity < 1}
     */
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }
  • LinkedBlockingQueue 头尾各1把锁
    /** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();

    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();

    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();
    
    /**
     * Inserts the specified element at the tail of this queue if it is
     * possible to do so immediately without exceeding the queue's capacity,
     * returning {@code true} upon success and {@code false} if this queue
     * is full.
     * When using a capacity-restricted queue, this method is generally
     * preferable to method {@link BlockingQueue#add add}, which can fail to
     * insert an element only by throwing an exception.
     *
     * @throws NullPointerException if the specified element is null
     */
    public boolean offer(E e) {
        if (e == null) throw new NullPointerException();
        final AtomicInteger count = this.count;
        if (count.get() == capacity)
            return false;
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
            if (count.get() < capacity) {
                enqueue(node);
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }
    
    public E poll(long timeout, TimeUnit unit) throws InterruptedException {
        E x = null;
        int c = -1;
        long nanos = unit.toNanos(timeout);
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                if (nanos <= 0)
                    return null;
                nanos = notEmpty.awaitNanos(nanos);
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

应用实例

Executors

里头用了LinkedBlockingQueue

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }
    
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    
    public static ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

使用LinkedBlockingQueue实现logger

public class BungeeLogger extends Logger {

    private final ColouredWriter writer;
    private final Formatter formatter = new ConciseFormatter();
//    private final LogDispatcher dispatcher = new LogDispatcher(this);

    private final BlockingQueue<LogRecord> queue = new LinkedBlockingQueue<>();

    volatile boolean running = true;

    Thread recvThread = new Thread(){
        @Override
        public void run() {
            while (!isInterrupted() && running) {
                LogRecord record;
                try {
                    record = queue.take();
                } catch (InterruptedException ex) {
                    continue;
                }

                doLog(record);
            }
            for (LogRecord record : queue) {
                doLog(record);
            }
        }
    };

    public BungeeLogger() throws IOException {
        super("BungeeCord", null);
        this.writer = new ColouredWriter(new ConsoleReader());

        try {
            FileHandler handler = new FileHandler("proxy.log", 1 << 24, 8, true);
            handler.setFormatter(formatter);
            addHandler(handler);
        } catch (IOException ex) {
            System.err.println("Could not register logger!");
            ex.printStackTrace();
        }
        recvThread.start();

        Runtime.getRuntime().addShutdownHook(new Thread(){
            @Override
            public void run() {
                running = false;
            }
        });
    }

    @Override
    public void log(LogRecord record) {
        if (running) {
            queue.add(record);
        }
    }

    void doLog(LogRecord record) {
        super.log(record);
        writer.print(formatter.format(record));
    }
}

doc

© 著作权归作者所有

共有 人打赏支持
go4it
粉丝 50
博文 643
码字总数 434875
作品 0
深圳
Java集合--阻塞队列(LinkedBlockingQueue)

1. LinkedBlockingQueue 上篇中,说到了ArrayBlockingQueue阻塞队列。在ArrayBlockingQueue中,底层使用了数组结构来实现。 那么,提到数组了就不得不提及链表。作为两对成双成对的老冤家,链...

贾博岩 ⋅ 2017/12/10 ⋅ 0

BlockingQueue详解

本例再次实现前面介绍的篮子程序,不过这个篮子中最多能放得苹果数不是1,可以随意指定。当篮子满时,生产者进入等待状态,当篮子空时,消费者等待。 BlockingQueue定义的常用方法如下: ad...

wow863597 ⋅ 2016/05/09 ⋅ 0

集合框架 Queue---BlockingQueue详解

本例介绍一个特殊的队列:BlockingQueue,如果BlockingQueue是空的,从BlockingQueue取东西的操作将会被阻断进入等待状态,直到BlockingQueue进了东西才会被唤醒,同样,如果BlockingQueue是...

长平狐 ⋅ 2012/11/28 ⋅ 0

并发十三:并发容器Queue实现分析

Queue J.U.C中分为阻塞队里和非阻塞队列。 阻塞队列在满时进行入列操作会被阻塞,空时进行出列操作会被阻塞,很适合并发编程中最常见的生产者-消费者模式。 非阻塞队使用CAS无锁算法避免锁竞...

wangjie2016 ⋅ 04/14 ⋅ 0

使用Java的BlockingQueue实现生产者-消费者

BlockingQueue也是java.util.concurrent下的主要用来控制线程同步的工具。 BlockingQueue有四个具体的实现类,根据不同需求,选择不同的实现类 1、ArrayBlockingQueue:一个由数组支持的有界阻...

mrliuze ⋅ 2015/08/11 ⋅ 0

阻塞队列之LinkedBlockingQueue源码分析

LinkedBlockingQueue是一个基于单向链表实现的可选容量的阻塞队列,队列的头节点是等待时间最长的元素,队列的尾节点是等待时间最短的元素。新元素直接插入到尾节点的后面,成为新的尾节点,...

qq_30572275 ⋅ 05/24 ⋅ 0

BlockingQueue学习

引言 在Concurrent包中,BlockingQueue很好的解决了在多线程中,如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利。同时...

jiangmitiao ⋅ 2015/08/30 ⋅ 0

理解BlockingQueue

在Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利。 what is ...

zyqJustin ⋅ 2016/05/05 ⋅ 0

源码|并发一枝花之BlockingQueue

今天来介绍Java并发编程中最受欢迎的同步类——堪称并发一枝花之BlockingQueue。 JDK版本:oracle java 1.8.0_102 继续阅读之前,需确保你对锁和条件队列的使用方法烂熟于心,特别是条件队列...

猴子007 ⋅ 2017/10/25 ⋅ 0

源码|并发一枝花之BlockingQueue

今天来介绍Java并发编程中最受欢迎的同步类——堪称并发一枝花之BlockingQueue。 JDK版本:oracle java 1.8.0_102 继续阅读之前,需确保你对锁和条件队列的使用方法烂熟于心,特别是条件队列...

monkeysayhi ⋅ 2017/11/27 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

知乎Java数据结构

作者:匿名用户 链接:https://www.zhihu.com/question/35947829/answer/66113038 来源:知乎 著作权归作者所有。商业转载请联系作者获得授权,非商业转载请注明出处。 感觉知乎上嘲讽题主简...

颖伙虫 ⋅ 今天 ⋅ 0

Confluence 6 恢复一个站点有关使用站点导出为备份的说明

推荐使用生产备份策略。我们推荐你针对你的生产环境中使用的 Confluence 参考 Production Backup Strategy 页面中的内容进行备份和恢复(这个需要你备份你的数据库和 home 目录)。XML 导出备...

honeymose ⋅ 今天 ⋅ 0

JavaScript零基础入门——(九)JavaScript的函数

JavaScript零基础入门——(九)JavaScript的函数 欢迎回到我们的JavaScript零基础入门,上一节课我们了解了有关JS中数组的相关知识点,不知道大家有没有自己去敲一敲,消化一下?这一节课,...

JandenMa ⋅ 今天 ⋅ 0

火狐浏览器各版本下载及插件httprequest

各版本下载地址:http://ftp.mozilla.org/pub/mozilla.org//firefox/releases/ httprequest插件截至57版本可用

xiaoge2016 ⋅ 今天 ⋅ 0

Docker系列教程28-实战:使用Docker Compose运行ELK

原文:http://www.itmuch.com/docker/28-docker-compose-in-action-elk/,转载请说明出处。 ElasticSearch【存储】 Logtash【日志聚合器】 Kibana【界面】 答案: version: '2'services: ...

周立_ITMuch ⋅ 今天 ⋅ 0

使用快嘉sdkg极速搭建接口模拟系统

在具体项目研发过程中,一旦前后端双方约定好接口,前端和app同事就会希望后台同事可以尽快提供可供对接的接口方便调试,而对后台同事来说定好接口还仅是个开始、设计流程,实现业务逻辑,编...

fastjrun ⋅ 今天 ⋅ 0

PXE/KickStart 无人值守安装

导言 作为中小公司的运维,经常会遇到一些机械式的重复工作,例如:有时公司同时上线几十甚至上百台服务器,而且需要我们在短时间内完成系统安装。 常规的办法有什么? 光盘安装系统 ===> 一...

kangvcar ⋅ 昨天 ⋅ 0

使用Puppeteer撸一个爬虫

Puppeteer是什么 puppeteer是谷歌chrome团队官方开发的一个无界面(Headless)chrome工具。Chrome Headless将成为web应用自动化测试的行业标杆。所以我们很有必要来了解一下它。所谓的无头浏...

小草先森 ⋅ 昨天 ⋅ 0

Java Done Right

* 表示难度较大或理论性较强。 ** 表示难度更大或理论性更强。 【Java语言本身】 基础语法,面向对象,顺序编程,并发编程,网络编程,泛型,注解,lambda(Java8),module(Java9),var(...

风华神使 ⋅ 昨天 ⋅ 0

Linux系统日志

linux 系统日志 /var/log/messages /etc/logrotate.conf 日志切割配置文件 https://my.oschina.net/u/2000675/blog/908189 logrotate 使用详解 dmesg 命令 /var/log/dmesg 日志 last命令,调......

Linux学习笔记 ⋅ 昨天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部