BlockingQueue

xiaoxiao2026-01-16 10

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

认识BlockingQueue阻塞队列，顾名思义，首先它是一个队列，而一个队列在数据结构中所起的作用大致如下图所示：从上图我们可以很清楚看到，通过一个共享的队列，可以使得数据由队列的一端输入，从另外一端输出；常用的队列主要有以下两种：（当然通过不同的实现方式，还可以延伸出很多不同类型的队列，DelayQueue就是其中的一种）　　先进先出（FIFO）：先插入的队列的元素也最先出队列，类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。　　后进先出（LIFO）：后插入队列的元素最先出队列，这种队列优先处理最近发生的事件。多线程环境中，通过队列可以很容易实现数据共享，比如经典的“生产者”和“消费者”模型中，通过队列可以很便利地实现两者之间的数据共享。假设我们有若干生产者线程，另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程，利用队列的方式来传递数据，就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内，万一发生数据处理速度不匹配的情况呢？理想情况下，如果生产者产出数据的速度大于消费者消费的速度，并且当生产出来的数据累积到一定程度的时候，那么生产者必须暂停等待一下（阻塞生产者线程），以便等待消费者线程把累积的数据处理完毕，反之亦然。然而，在concurrent包发布以前，在多线程环境下，我们每个程序员都必须去自己控制这些细节，尤其还要兼顾效率和线程安全，而这会给我们的程序带来不小的复杂度。好在此时，强大的concurrent包横空出世了，而他也给我们带来了强大的BlockingQueue。（在多线程领域：所谓阻塞，在某些情况下会挂起线程（即阻塞），一旦条件满足，被挂起的线程又会自动被唤醒）下面两幅图演示了BlockingQueue的两个常见阻塞场景：　　　　　　　如上图所示：当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列。　　　如上图所示：当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。这也是我们在多线程环境下，为什么需要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家庭大致有哪些成员？ BlockingQueue成员详细介绍 1. ArrayBlockingQueue ArrayBlockingQueue是对BlockingQueue的一个数组实现，它使用一把全局的锁并行对queue的读写操作，同时使用两个Condition阻塞容量为空时的取操作和容量满时的写操作,是一个典型的有界缓存区，固定大小的数组在其保持生产者插入的元素和使用者提取的元素。

二、具体实现 ArrayBlockingQueue底层定义如下：

Java代码 public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { // 使用循环数组来实现queue，初始时takeIndex和putIndex均为0 private final E[] items; private transient int takeIndex; private transient int putIndex; private int count; // 用于并发的锁和条件 private final ReentrantLock lock; private final Condition notEmpty; private final Condition notFull; final int inc(int i) { return (++i == items.length)? 0 : i; } public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = (E[]) new Object[capacity]; // 分配锁及该锁上的condition lock = new ReentrantLock(fair); notEmpty = lock.newCondition(); notFull = lock.newCondition(); } ... } ArrayBlockingQueue的取操作：

Java代码 public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { private E extract() { final E[] items = this.items; E x = items[takeIndex]; items[takeIndex] = null; takeIndex = inc(takeIndex); --count; // 激发notFull条件 notFull.signal(); return x; } public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { try { // 等待notEmpty的条件 while (count == 0) notEmpty.await(); } catch (InterruptedException ie) { notEmpty.signal(); // propagate to non-interrupted thread throw ie; } E x = extract(); return x; } finally { lock.unlock(); } } ... } ArrayBlockingQueue的写操作：

Java代码 public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { private void insert(E x) { items[putIndex] = x; putIndex = inc(putIndex); ++count; notEmpty.signal(); } public void put(E o) throws InterruptedException { if (o == null) throw new NullPointerException(); final E[] items = this.items; final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { try { // 等待notFull条件 while (count == items.length) notFull.await(); } catch (InterruptedException ie) { notFull.signal(); // propagate to non-interrupted thread throw ie; } insert(o); } finally { lock.unlock(); } } ... } 注意：ArrayBlockingQueue在读写操作上都需要锁住整个容器，因此吞吐量与一般的实现是相似的，适合于实现“生产者消费者”模式。

LinkedBlockingQueue

LinkedBlockingQueue实现是线程安全的，实现了FIFO（先进先出）等特性. 是作为生产者消费者的首选，LinkedBlockingQueue 可以指定容量，也可以不指定，不指定的话，默认最大是Integer.MAX_VALUE，其中主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。 import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.LinkedBlockingQueue; public class MyBlockingQueue extends Thread{ public static BlockingQueue<String> queue=new LinkedBlockingQueue<String>(3); private int index; public MyBlockingQueue(int i){ this.index=i; } public void run(){ try{ queue.put(String.valueOf(this.index)); System.out.println("put {"+this.index+"} into queue!"); }catch(Exception e){ e.printStackTrace(); } } public static void main(String args[]){ ExecutorService service=Executors.newCachedThreadPool(); for( int i=0; i<10; i++){ service.submit(new MyBlockingQueue(i)); } Thread thread = new Thread(){ public void run(){ try{ while(true){ Thread.sleep((int)(Math.random()*1000)); if(MyBlockingQueue.queue.isEmpty()) break; String str=MyBlockingQueue.queue.take(); System.out.println("take {" + str+"} out of queue!"); } }catch(Exception e){ e.printStackTrace(); } } }; service.submit(thread); service.shutdown(); } } DelayQueue DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。使用场景：　　DelayQueue使用场景较少，但都相当巧妙，常见的例子比如使用一个DelayQueue来管理一个超时未响应的连接队列。 PriorityBlockingQueue 基于优先级的阻塞队列（优先级的判断通过构造函数传入的Compator对象来决定），但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者，而只会在没有可消费的数据时，阻塞数据的消费者。因此使用的时候要特别注意，生产者生产数据的速度绝对不能快于消费者消费数据的速度，否则时间一长，会最终耗尽所有的可用堆内存空间。在实现PriorityBlockingQueue时，内部控制线程同步的锁采用的是公平锁。相关资源：python入门教程(PDF版)

最新回复(0)