Java高并发之BlockingQueue

简介

多线程中通过队列很容易共享数据,比如经典的生产者和消费者模型中,通过队列可以很方便的实现数据共享。假设我们有若干生产者线程,又有若干消费者线程,生产者线程可以通过队列将数据共享给消费者。但是生产者和消费者在某个时间段内,万一发生数据处理速度不匹配的情况呢?如果生产者生产数据的速度远大于消费者消费数据的速度,理想情况下是当生产者产生的数据到达一个阈值之后,那么生产者必须暂停一下(阻塞生产者线程),以便消费者可以把数据消费掉。在concurrent包出现之前,开发人员必须手动控制这些细节,导致开发高性能程序难度较大(兼顾效率和安全)。concurrent出来之后,带来了BlockingQueue(在多线程中,在某些情况下挂起线程(即阻塞),一旦条件满足,被挂起的线程又会被自动唤醒)

这里写图片描述

BlockingQueue即为阻塞队列,是一个先进先出的队列,在某些情况下,对阻塞队列的访问可能会造成阻塞,被阻塞的情况主要有两种。

• 当对列满时,进行入队操作时。当一个线程试图对一个已经满了的队列进行入队操作时,也将会被阻塞,除非有一个线程进行了出队列操作。
• 当队列空时,进行出队操作时。当一个线程试图对一个为空的队列进行出队列操作时,也将会被阻塞,除非有一个线程进行了出队列操作。

阻塞队列是线程安全的,主要用在生产者消费者的场景上。负责生产的线程不断的制造新对象并插入到阻塞队列中,直到到达队列的上限值,从而被阻塞,直到消费线程对队列进行消费。同理,负责消费的线程不断的从队列中取出对象进行消费,直到这个队列为空,这时消费队列会被阻塞,除非队列中有新的对象被加入进来。

public interface BlockingQueue<E> extends Queue<E> {}
public interface Queue<E> extends Collection<E> {}

BlockingQueue是一个接口,继承自Queue,所以实现类也可以作为Queue的实现来使用,而Queue又继承自Collection接口。

BlockingQueue对插入操作,移除操作,获取元素操作提供了四种不同的方法用于不同的场景,使用不同的方法,会有不同的效果。BlockingQueue的各个实现都遵循这些规则。

• Throws Exception：抛出异常。如果不能马上进行，则抛出异常。
• Special Value：如果不能马上进行，则返回特殊值，一般是True或False
• Blocks：如果不能马上进行，则操作会被阻塞，直到这个操作成功
• Times Out：如果不能马上进行，操作会被阻塞指定的时间。如果指定时间还未执行，则返回特殊值，一般是True或False。

对于BlockingQueue,关注点应该在它的put和take方法上,因为这两个方法是带阻塞。

BlockingQueue不接受null值的插入,相应的方法在碰到null的插入时会抛出NullPointerException异常,null值通常用于特殊值返回(表格中的第三列),代表poll失败。所以如果允许插入null值的话,那获取的时候,就不能很好的用null来判断到底是失败还是获取的值为null。

BlockingQueue实现了java.util.Collection接口,我们可以使用remove(x)来删除任意一个元素,但是这类操作并不高效,所以尽量在少数场合使用,比如一条消息已经入队,但是需要取消操作的时候。

BlockingQueue的实现都是线程安全的,但是批量的集合操作addAll,containsAll,retainAll,removeAll不一定是原子操作,如addAll(c)添加了一些元素后抛出异常,此时BlockingQueue中已经添加了部分元素,这个是允许的,取决于具体实现。

BlockingQueue在生产者-消费者的场景中,是支持多消费者和多消费者的,说的其实就是线程安全问题。BlockingQueue是一个比较简单的线程安全容器。作为BlockingQueue的使用者,我们再不用考虑何时阻塞线程,什么时候唤醒线程,因为这一些BlockingQueue都实现了。

无界队列,并不是大小不限制,只是它的大小是Integer.MAX_VALUE,即int类型能表示的最大值(2的31次方)-1

BlockingQueue具体实现类

• ArrayBlockingQueue
• LinkedBlockingQueue
• DelayQueue
• PriorityBlockingQueue
• SynchronousQueue

常用的是ArrayBlockingQueue和LinkedBlockingQueue

wx.jpg