并发举措七之阻塞队列_预占模式
赞
踩
前言：
阻塞队列的典型使用场景就是生产者/消费者模式。本文会介绍Java7种阻塞队列。
一、阻塞对列的实现机制
阻塞队列，关键字是阻塞，先理解阻塞的含义，在阻塞队列中，线程阻塞有这样的两种情况：
1.当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列。
2.当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。
二、阻塞对列常用api
阻塞队列提供了四种处理方法:
| 方法\处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
|---|---|---|---|---|
| 插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
| 移除方法 | remove() | poll() | take() | poll(time,unit) |
| 检查方法 | element() | peek() | 不可用 | 不可用 |
异常：是指当阻塞队列满时候，再往队列里插入元素，会抛出IllegalStateException("Queue full")异常。当队列为空时，从队列里获取元素时会抛出NoSuchElementException异常 。
- 返回特殊值：插入方法会返回是否成功，成功则返回true。移除方法，则是从队列里拿出一个元素，如果没有则返回null
- 一直阻塞：当阻塞队列满时，如果生产者线程往队列里put元素，队列会一直阻塞生产者线程，直到拿到数据，或者响应中断退出。当队列空时，消费者线程试图从队列里take元素，队列也会阻塞消费者线程，直到队列可用。
- 超时退出：当阻塞队列满时，队列会阻塞生产者线程一段时间，如果超过一定的时间，生产者线程就会退出。
添加数据操作：
1：publicabstract boolean add(E paramE)：将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回true，如果当前没有可用的空间，则抛出IllegalStateException。如果该元素是NULL，则会抛出NullPointerException异常。
2：public abstract boolean offer(E paramE)：将指定元素插入此队列中（如果立即可行且不会违反容量限制），成功时返回true，如果当前没有可用的空间，则返回false。
3：public abstract void put(E paramE) throws InterruptedException：将指定元素插入此队列中，将等待可用的空间（如果有必要）
4：offer(E o, long timeout, TimeUnit unit)：可以设定等待的时间，如果在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。
获取数据操作：
1：poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null;
2：poll(long timeout, TimeUnit unit)：从BlockingQueue取出一个队首的对象，如果在指定时间内，队列一旦有数据可取，则立即返回队列中的数据。否则直到时间超时还没有数据可取，返回失败。
3：take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入。
4.drainTo():一次性从BlockingQueue获取所有可用的数据对象（还可以指定获取数据的个数），通过该方法，可以提升获取数据效率；不需要多次分批加锁或释放锁。
三、阻塞队列实现类图
四、7种阻塞队列介绍
4.1、ArrayBlockingQueue ：由数组结构组成的有界阻塞队列（公平、非公平）
- ArrayBlockingQueue有界且固定，在构造函数时确认大小，确认后不支持改变，
- 此队列按照先进先出（FIFO）的原则对元素进行排序，在多线程环境下默认不保证“公平性”
- 通常情况下为了保证公平性会降低吞吐量。我们可以使用以下代码创建一个公平的阻塞队列：ArrayBlockingQueue fairQueue = new ArrayBlockingQueue(1000,true);
- ReentrantLock+Condition实现
4.2、LinkedBlockingQueue ：由链表结构组成的有界阻塞队列（两个独立锁提高并发）
- 基于链表的阻塞队列
- 按照先进先出（FIFO）的原则对元素进行排序。
- 够高效的处理并发数据。因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。
- LinkedBlockingQueue会默认一个类似无限大小的容量（Integer.MAX_VALUE）。
4.3、PriorityBlockingQueue ：支持优先级排序的无界阻塞队列。
- 基于二叉堆的无界阻塞队列， 最大堆：父节点的键值总是大于或等于任何一个子节点的键值， 最小堆：父节点的键值总是小于或等于任何一个子节点的键值， 添加操作则是不断“上冒”，而删除操作则是不断“下掉”
- 是一个支持优先级的无界队列。
- 默认情况下元素采取自然顺序升序排列，可以自定义实现compareTo()方法来指定元素进行排序规则，或者初始化PriorityBlockingQueue时，指定构造参数Comparator来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。
- ReentrantLock + Condition实现
4.4、DelayQueue：使用优先级队列实现的无界阻塞队列。
- 队列使用PriorityQueue来实现，队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。
- 支持延时获取元素的无界阻塞队列
- 应用场景：
1.缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。
2.定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，从比如TimerQueue就是使用DelayQueue实现的。
- ReentrantLock + Condition 实现
4.5、SynchronousQueue：不存储元素的阻塞队列。（注意同threadLocal的不同，threadLocal是在同一线程中共享数据代替数据专递来提升性能）
- 一个没有容量的阻塞队列，每一个put操作必须等待一个take操作，否则不能继续添加元素。
- 应用场景： 非常适合于传递性场景,比如在一个线程中使用的数据，传递给另外一个线程使用
- SynchronousQueue的吞吐量高于LinkedBlockingQueue 和ArrayBlockingQueue。
4.6、LinkedTransferQueue：由链表结构组成的无界阻塞队列。
- 链表组成的的无界阻塞队列, 相当于ConcurrentLinkedQueue、SynchronousQueue (公平模式下)、无界的LinkedBlockingQueues等的超集；
- 预占模式: 有就直接拿走，没有就占着这个位置直到拿到或者超时或者中断
- 相对于其他阻塞队列，LinkedTransferQueue多了tryTransfer和transfer方法。
1.transfer方法：如果当前有消费者正在等待接收元素（消费者使用take()方法或带时间限制的poll()方法时），transfer方法可以把生产者传入的元素立刻transfer（传输）给消费者。如果没有消费者在等待接收元素，transfer方法会将元素存放在队列的tail节点，并等到该元素被消费者消费了才返回。
2.tryTransfer方法。则是用来试探下生产者传入的元素是否能直接传给消费者。如果没有消费者等待接收元素，则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收，方法立即返回。而transfer方法是必须等到消费者消费了才返回。对于带有时间限制的tryTransfer(E e, long timeout, TimeUnit unit)方法，则是试图把生产者传入的元素直接传给消费者，但是如果没有消费者消费该元素则等待指定的时间再返回，如果超时还没消费元素，则返回false，如果在超时时间内消费了元素，则返回true。
4.7、LinkedBlockingDeque：由链表结构组成的双向阻塞队列
- 由链表组成的双向阻塞队列，所谓双向队列指的你可以从队列的两端插入和移出元素。双端队列因为多了一个操作队列的入口，在多线程同时入队时，也就减少了一半的竞争。
- 容量可选，在初始化时可以设置容量防止其过度膨胀，如果不设置，默认容量大小为Integer.MAX_VALUE
- 相比其他的阻塞队列，LinkedBlockingDeque多了addFirst，addLast，offerFirst，offerLast，peekFirst，peekLast等方法，以First单词结尾的方法，表示插入，获取（peek）或移除双端队列的第一个元素。以Last单词结尾的方法，表示插入，获取或移除双端队列的最后一个元素。另外插入方法add等同于addLast，移除方法remove等效于removeFirst。但是take方法却等同于takeFirst，不知道是不是Jdk的bug，使用时还是用带有First和Last后缀的方法更清楚。
- 应用： “工作窃取”模式
五、阻塞对应应用实例
- //生产者
- public class Producer implements Runnable{
- private final BlockingQueue<Integer> queue;
-
- public Producer(BlockingQueue q){
- this.queue=q;
- }
-
- @Override
- public void run() {
- try {
- while (true){
- Thread.sleep(1000);//模拟耗时
- queue.put(produce());
- }
- }catch (InterruptedException e){
-
- }
- }
-
- private int produce() {
- int n=new Random().nextInt(10000);
- System.out.println("Thread:" + Thread.currentThread().getId() + " produce:" + n);
- return n;
- }
- }
- //消费者
- public class Consumer implements Runnable {
- private final BlockingQueue<Integer> queue;
-
- public Consumer(BlockingQueue q){
- this.queue=q;
- }
-
- @Override
- public void run() {
- while (true){
- try {
- Thread.sleep(2000);//模拟耗时
- consume(queue.take());
- }catch (InterruptedException e){
-
- }
-
- }
- }
-
- private void consume(Integer n) {
- System.out.println("Thread:" + Thread.currentThread().getId() + " consume:" + n);
-
- }
- }
- //测试
- public class Main {
-
- public static void main(String[] args) {
- BlockingQueue<Integer> queue=new ArrayBlockingQueue<Integer>(100);
- Producer p=new Producer(queue);
- Consumer c1=new Consumer(queue);
- Consumer c2=new Consumer(queue);
-
- new Thread(p).start();
- new Thread(c1).start();
- new Thread(c2).start();
- }
- }
六、阻塞队列的优点
6.1 、降低多线程开发的难度
由于阻塞队列本身是线程安全的，队列可以安全地从一个线程向另外一个线程传递数据，所以我们的生产者/消费者直接使用线程安全的队列就可以，而不需要自己去考虑更多的线程安全问题。这也就意味着，考虑锁等线程安全问题的重任从你转移到了 队列 上，降低了我们开发的难度和工作量。
6.2 、隔离代码，实现业务代码解耦
队列还能起到一个隔离的作用。比如说我们开发一个银行转账的程序，那么生产者线程不需要关心具体的转账逻辑，只需要把转账任务，如账户和金额等信息放到（put）队列中就可以，而不需要去关心银行这个类如何实现具体的转账业务。而作为银行这个类来讲，它会去从队列里（take）取出来将要执行的具体的任务，再去通过自己的各种方法来完成本次转账。这样就实现了具体任务与执行任务类之间的解耦，任务被放在了阻塞队列中，而负责放任务的线程是无法直接访问到我们银行具体实现转账操作的对象的，实现了隔离，提高了安全性。
