Java多线程-BlockingQueue-ArrayBlockingQueue-LinkedBlockingQueue

时间  2019-11-26
标签 java 多线程 blockingqueue arrayblockingqueue linkedblockingqueue 栏目 Java 繁體版
原文   原文链接

前言：html

  BlockingQueue很好的解决了多线程中，如何高效安全“传输”数据的问题。经过这些高效而且线程安全的队列类，为咱们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了BlockingQueue家庭中的两个重要成员，包括他们各自的功能以及常见使用场景。

认识BlockingQueuejava

  阻塞队列，顾名思义，首先它是一个队列，而一个队列在数据结构中所起的做用大体以下图所示：程序员

  从图中咱们能够很清楚看到，经过一个共享的队列，可使得数据由队列的一端输入，从另一端输出；经常使用的队列主要有如下两种：（固然经过不一样的实现方式，还能够延伸出不少不一样类型的队列，DelayQueue就是其中的一种）数组

　　先进先出（FIFO）：先插入的队列的元素也最早出队列，相似于排队的功能。从某种程度上来讲这种队列也体现了一种公平性。缓存

　　后进先出（LIFO）：后插入队列的元素最早出队列，这种队列优先处理最近发生的事件。

  多线程环境中，经过队列能够很容易实现数据共享，好比经典的“生产者”和“消费者”模型中，经过队列能够很便利地实现二者之间的数据共享。假设咱们有若干生产者线程，另外又有若干个消费者线程。若是生产者线程须要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就能够很方便地解决他们之间的数据共享问题。但若是生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的状况呢？理想状况下，若是生产者产出数据的速度大于消费者消费的速度，而且当生产出来的数据累积到必定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。安全

  然而，在concurrent包发布之前，在多线程环境下，咱们每一个程序员都必须去本身控制这些细节，尤为还要兼顾效率和线程安全，而这会给咱们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给咱们带来了强大的BlockingQueue。（在多线程领域：所谓阻塞，在某些状况下会挂起线程，一旦条件知足，被挂起的线程又会自动被唤醒）数据结构

  下面两幅图演示了BlockingQueue的两个常见阻塞场景：多线程

如图所示：当队列中没有数据的状况下，消费者端的全部线程都会被自动阻塞（挂起），直到有数据放入队列。并发

如图所示：当队列中填满数据的状况下，生产者端的全部线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。dom

  这也是咱们在多线程环境下，为何须要BlockingQueue的缘由。做为BlockingQueue的使用者，咱们不再须要关心何时须要阻塞线程，何时须要唤醒线程，由于这一切BlockingQueue都给你一手包办了。既然BlockingQueue如此神通广大，让咱们一块儿来见识下它的经常使用方法：

BlockingQueue的核心方法：

放入数据：

　　offer(anObject):表示若是可能的话,将anObject加到BlockingQueue里,即若是BlockingQueue能够容纳,则返回true,不然返回false.（本方法不阻塞当前执行方法的线程）

　　offer(E o, long timeout, TimeUnit unit),能够设定等待的时间，若是在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。

　　put(anObject):把anObject加到BlockingQueue里,若是BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.

获取数据：

　　poll(time):取走BlockingQueue里排在首位的对象,若不能当即取出,则能够等time参数规定的时间,取不到时返回null;

　　poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，若是在指定时间内，队列一旦有数据可取，则当即返回队列中的数据。不然知道时间超时尚未数据可取，返回失败。

　　take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入;

　　drainTo():一次性从BlockingQueue获取全部可用的数据对象（还能够指定获取数据的个数），经过该方法，能够提高获取数据效率；不须要屡次分批加锁或释放锁。

常见BlockingQueue

在了解了BlockingQueue的基本功能后，让咱们来看看BlockingQueue家庭大体有哪些成员？

1. ArrayBlockingQueue

  基于数组的阻塞队列实现，在ArrayBlockingQueue内部，维护了一个定长数组，以便缓存队列中的数据对象，这是一个经常使用的阻塞队列，除了一个定长数组外，ArrayBlockingQueue内部还保存着两个整形变量，分别标识着队列的头部和尾部在数组中的位置。

  ArrayBlockingQueue在生产者放入数据和消费者获取数据，都是共用同一个锁对象，由此也意味着二者没法真正并行运行，这点尤为不一样于LinkedBlockingQueue；

  按照实现原理来分析，ArrayBlockingQueue彻底能够采用分离锁，从而实现生产者和消费者操做的彻底并行运行。之因此没这样去作，猜想是由于ArrayBlockingQueue的数据写入和获取操做已经足够轻巧，以致于引入独立的锁机制，除了给代码带来额外的复杂性外，其在性能上彻底占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不一样之处在于，前者在插入或删除元素时不会产生或销毁任何额外的对象实例，然后者则会生成一个额外的Node对象。这在长时间内须要高效并发地处理大批量数据的系统中，其对于GC的影响仍是存在必定的区别。而在建立ArrayBlockingQueue时，咱们还能够控制对象的内部锁是否采用公平锁，默认采用非公平锁。

2. LinkedBlockingQueue

  基于链表的阻塞队列，同ArrayListBlockingQueue相似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者当即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue能够经过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于一样的原理。而LinkedBlockingQueue之因此可以高效的处理并发数据，还由于其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的状况下生产者和消费者能够并行地操做队列中的数据，以此来提升整个队列的并发性能。

  做为开发者，咱们须要注意的是，若是构造一个LinkedBlockingQueue对象，而没有指定其容量大小，LinkedBlockingQueue会默认一个相似无限大小的容量（Integer.MAX_VALUE），这样的话，若是生产者的速度一旦大于消费者的速度，也许尚未等到队列满阻塞产生，系统内存就有可能已被消耗殆尽了。

  ArrayBlockingQueue和LinkedBlockingQueue是两个最普通也是最经常使用的阻塞队列，通常状况下，在处理多线程间的生产者消费者问题，使用这两个类足以。

BlockingQueueTest.java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueTest {

    public static void main(String[] args) throws InterruptedException {
        // 声明一个容量为10的缓存队列
        BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);

        Producer producer1 = new Producer(queue);
        Producer producer2 = new Producer(queue);
        Producer producer3 = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        // 借助Executors
        ExecutorService service = Executors.newCachedThreadPool();
        // 启动线程
        service.execute(producer1);
        service.execute(producer2);
        service.execute(producer3);
        service.execute(consumer);

        // 执行10s
        Thread.sleep(10 * 1000);
        producer1.stop();
        producer2.stop();
        producer3.stop();

        Thread.sleep(2000);
        // 退出Executor
        service.shutdown();
    }
}

  上面程序使用了Executors线程池的用法，详细内容能够参考：http://www.cnblogs.com/liqiu/p/3649360.html

Cosumer.java

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class Consumer implements Runnable {

    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    public void run() {
        System.out.println("启动消费者线程！");
        Random r = new Random();
        boolean isRunning = true;
        try {
            while (isRunning) {
                System.out.println("正从队列获取数据...");
                String data = queue.poll(2, TimeUnit.SECONDS);
                if (null != data) {
                    System.out.println("拿到数据：" + data);
                    System.out.println("正在消费数据：" + data);
                    Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
                } else {
                    // 超过2s还没数据，认为全部生产线程都已经退出，自动退出消费线程。
                    isRunning = false;
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            System.out.println("退出消费者线程！");
        }
    }

    private BlockingQueue<String> queue;
    private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
}

Producer.java

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
public class Producer implements Runnable {

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

    public void run() {
        String data = null;
        Random r = new Random();

        System.out.println("启动生产者线程！");
        try {
            while (isRunning) {
                System.out.println("正在生产数据...");
                Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));

                data = "data:" + count.incrementAndGet();
                System.out.println("将数据：" + data + "放入队列...");
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {
                    System.out.println("放入数据失败：" + data);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            System.out.println("退出生产者线程！");
        }
    }

    public void stop() {
        isRunning = false;
    }

    private volatile boolean isRunning = true;
    private BlockingQueue queue;
    private static AtomicInteger count = new AtomicInteger();
    private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;

}

 执行结果：

。。。。。。。。。。。。。。。
正从队列获取数据...
拿到数据：data:36
正在消费数据：data:36
正从队列获取数据...
拿到数据：data:37
正在消费数据：data:37
正从队列获取数据...
拿到数据：data:38
正在消费数据：data:38
。。。。。。。。。。。。
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程