做教师知识网站有哪些,做木材生意的外贸网站,站长之家whois查询,wordpress 空间商转载自 Java多线程-BlockingQueue-ArrayBlockingQueue-LinkedBlockingQueue 前言#xff1a; BlockingQueue很好的解决了多线程中#xff0c;如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类#xff0c;为我们快速搭建高质量的多线程程序带来极大的便利…转载自 Java多线程-BlockingQueue-ArrayBlockingQueue-LinkedBlockingQueue 前言 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家庭大致有哪些成员 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的缓存队列BlockingQueueString queue new LinkedBlockingQueueString(10);Producer producer1 new Producer(queue);Producer producer2 new Producer(queue);Producer producer3 new Producer(queue);Consumer consumer new Consumer(queue);// 借助ExecutorsExecutorService service Executors.newCachedThreadPool();// 启动线程service.execute(producer1);service.execute(producer2);service.execute(producer3);service.execute(consumer);// 执行10sThread.sleep(10 * 1000);producer1.stop();producer2.stop();producer3.stop();Thread.sleep(2000);// 退出Executorservice.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(BlockingQueueString 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 BlockingQueueString 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
。。。。。。。。。。。。
