Administrator
发布于 2023-01-28 / 55 阅读
0
0

BlockingQueue Best Practise

BlockingQueue

注意:该随笔内容完全引自http://wsmajunfeng.iteye.com/blog/1629354,写的很好,非常感谢,复制过来算是个积累,怕以后找不到。

一. 前言

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

二. 认识BlockingQueue

阻塞队列,顾名思义,首先它是一个队列,而一个队列在数据结构中所起的作用大致如下图所示:

img

从上图我们可以很清楚看到,通过一个共享的队列,可以使得数据由队列的一端输入,从另外一端输出;

常用的队列主要有以下两种:(当然通过不同的实现方式,还可以延伸出很多不同类型的队列,DelayQueue就是其中的一种)

先进先出(FIFO):先插入的队列的元素也最先出队列,类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。

后进先出(LIFO):后插入队列的元素最先出队列,这种队列优先处理最近发生的事件。

​ 多线程环境中,通过队列可以很容易实现数据共享,比如经典的“生产者”和“消费者”模型中,通过队列可以很便利地实现两者之间的数据共享。

假设我们有若干生产者线程,另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程,利用队列的方式来传递数据,就可以很方便地解决他们之间的数据共享问题。

但如果生产者和消费者在某个时间段内,万一发生数据处理速度不匹配的情况呢?理想情况下,如果生产者产出数据的速度大于消费者消费的速度,并且当生产出来的数据累积到一定程度的时候,那么生产者必须暂停等待一下(阻塞生产者线程),以便等待消费者线程把累积的数据处理完毕,反之亦然。

然而,在concurrent包发布以前,在多线程环境下,我们每个程序员都必须去自己控制这些细节,尤其还要兼顾效率和线程安全,而这会给我们的程序带来不小的复杂度。好在此时,强大的concurrent包横空出世了,而他也给我们带来了强大的BlockingQueue。

(在多线程领域:所谓阻塞,在某些情况下会挂起线程(即阻塞),一旦条件满足,被挂起的线程又会自动被唤醒),下面两幅图演示了BlockingQueue的两个常见阻塞场景:


img

如上图所示:当队列中没有数据的情况下,消费者端的所有线程都会被自动阻塞(挂起),直到有数据放入队列



img

如上图所示:当队列中填满数据的情况下,生产者端的所有线程都会被自动阻塞(挂起),直到队列中有空的位置,线程被自动唤醒

这也是我们在多线程环境下,为什么需要BlockingQueue的原因。作为BlockingQueue的使用者,我们再也不需要关心什么时候需要阻塞线程,什么时候需要唤醒线程,因为这一切BlockingQueue都给你一手包办了。既然BlockingQueue如此神通广大,让我们一起来见识下它的常用方法:

三. BlockingQueue的核心方法

1.放入数据

(1)offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.(本方法不阻塞当前执行方法

的线程);      
  (2)offer(E o, long timeout, TimeUnit unit):可以设定等待的时间,如果在指定的时间内,还不能往队列中加入BlockingQueue,则返回失败。

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

2. 获取数据

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

(2)poll(long timeout, TimeUnit unit):从BlockingQueue取出一个队首的对象,如果在指定时间内,队列一旦有数据可取,则立即返回队列中的数据。否则知道时间

超时还没有数据可取,返回失败。

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

(4)drainTo():一次性从BlockingQueue获取所有可用的数据对象(还可以指定获取数据的个数),通过该方法,可以提升获取数据效率;不需要多次分批加锁或释放锁。

四. 常见BlockingQueue

在了解了BlockingQueue的基本功能后,让我们来看看BlockingQueue家庭大致有哪些成员?

img

  1. ArrayBlockingQueue

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

ArrayBlockingQueue在生产者放入数据和消费者获取数据,都是共用同一个锁对象,由此也意味着两者无法真正并行运行,这点尤其不同于LinkedBlockingQueue;按照实现原理来分析,ArrayBlockingQueue完全可以采用分离锁,从而实现生产者和消费者操作的完全并行运行。Doug Lea之所以没这样去做,也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧,以至于引入独立的锁机制,除了给代码带来额外的复杂性外,其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于,前者在插入或删除元素时不会产生或销毁任何额外的对象实例,而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中,其对于GC的影响还是存在一定的区别。而在创建ArrayBlockingQueue时,我们还可以控制对象的内部锁是否采用公平锁,默认采用非公平锁。

2.LinkedBlockingQueue

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

作为开发者,我们需要注意的是,如果构造一个LinkedBlockingQueue对象,而没有指定其容量大小,LinkedBlockingQueue会默认一个类似无限大小的容量(Integer.MAX_VALUE),这样的话,如果生产者的速度一旦大于消费者的速度,也许还没有等到队列满阻塞产生,系统内存就有可能已被消耗殆尽了。

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

插入和移除操作的4中处理方式
1570705034507

抛出异常:当队列满时,如果再往队列里插入元素,会抛出IllegalStateException(“Queue
full”)异常。当队列空时,从队列里获取元素会抛出NoSuchElementException异常。

·返回特殊值:当往队列插入元素时,会返回元素是否插入成功,成功返回true。如果是移
除方法,则是从队列里取出一个元素,如果没有则返回null。

·一直阻塞:当阻塞队列满时,如果生产者线程往队列里put元素,队列会一直阻塞生产者
线程,直到队列可用或者响应中断退出。当队列空时,如果消费者线程从队列里take元素,队
列会阻塞住消费者线程,直到队列不为空。

·超时退出:当阻塞队列满时,如果生产者线程往队列里插入元素,队列会阻塞生产者线程
一段时间,如果超过了指定的时间,生产者线程就会退出

这里的peek,就是获取到队列中的第一个元素,但是不同于poll的是:调用了peek后,不会将元素从队列中移除,但是调用了poll后,会将元素从队列中移除。

三.BlockingQueue中API介绍

offer(E e): 将给定的元素设置到队列中,如果设置成功返回true, 否则返回false. e的值不能为空,否则抛出空指针异常。

offer(E e, long timeout, TimeUnit unit): 将给定元素在给定的时间内设置到队列中,如果设置成功返回true, 否则返回false.

add(E e): 将给定元素设置到队列中,如果设置成功返回true, 否则抛出异常。如果是往限定了长度的队列中设置值,推荐使用offer()方法。

put(E e): 将元素设置到队列中,如果队列中没有多余的空间,该方法会一直阻塞,直到队列中有多余的空间。

take(): 从队列中获取值,如果队列中没有值,线程会一直阻塞,直到队列中有值,并且该方法取得了该值。

poll(long timeout, TimeUnit unit): 在给定的时间里,从队列中获取值,如果没有取到会抛出异常。

remainingCapacity():获取队列中剩余的空间。

remove(Object o): 从队列中移除指定的值。

contains(Object o): 判断队列中是否拥有该值。

drainTo(Collection c): 将队列中值,全部移除,并发设置到给定的集合中。

下面的代码演示了如何使用BlockingQueue:

(1) 测试类


 1 import java.util.concurrent.BlockingQueue;
 2 import java.util.concurrent.ExecutorService;
 3 import java.util.concurrent.Executors;
 4 import java.util.concurrent.LinkedBlockingQueue; 
 5 
 6 public class BlockingQueueTest {
 7  
 8     public static void main(String[] args) throws InterruptedException {
 9         // 声明一个容量为10的缓存队列
10         BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);
11  
12         //new了三个生产者和一个消费者
13         Producer producer1 = new Producer(queue);
14         Producer producer2 = new Producer(queue);
15         Producer producer3 = new Producer(queue);
16         Consumer consumer = new Consumer(queue);
17  
18         // 借助Executors
19         ExecutorService service = Executors.newCachedThreadPool();
20         // 启动线程
21         service.execute(producer1);
22         service.execute(producer2);
23         service.execute(producer3);
24         service.execute(consumer);
25  
26         // 执行10s
27         Thread.sleep(10 * 1000);
28         producer1.stop();
29         producer2.stop();
30         producer3.stop();
31  
32         Thread.sleep(2000);
33         // 退出Executor
34         service.shutdown();
35     }
36 }

(2)生产者类


 1 import java.util.Random;
 2 import java.util.concurrent.BlockingQueue;
 3 import java.util.concurrent.TimeUnit;
 4 import java.util.concurrent.atomic.AtomicInteger;
 5  
 6 /**
 7  * 生产者线程
 8  * 
 9  * @author jackyuj
10  */
11 public class Producer implements Runnable {
12     
13     private volatile boolean  isRunning = true;//是否在运行标志
14     private BlockingQueue queue;//阻塞队列
15     private static AtomicInteger count = new AtomicInteger();//自动更新的值
16     private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
17  
18     //构造函数
19     public Producer(BlockingQueue queue) {
20         this.queue = queue;
21     }
22  
23     public void run() {
24         String data = null;
25         Random r = new Random();
26  
27         System.out.println("启动生产者线程!");
28         try {
29             while (isRunning) {
30                 System.out.println("正在生产数据...");
31                 Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));//取0~DEFAULT_RANGE_FOR_SLEEP值的一个随机数
32  
33                 data = "data:" + count.incrementAndGet();//以原子方式将count当前值加1
34                 System.out.println("将数据:" + data + "放入队列...");
35                 if (!queue.offer(data, 2, TimeUnit.SECONDS)) {//设定的等待时间为2s,如果超过2s还没加进去返回false
36                     System.out.println("放入数据失败:" + data);
37                 }
38             }
39         } catch (InterruptedException e) {
40             e.printStackTrace();
41             Thread.currentThread().interrupt();
42         } finally {
43             System.out.println("退出生产者线程!");
44         }
45     }
46  
47     public void stop() {
48         isRunning = false;
49     }
50 }

(3)消费者类


 1 import java.util.Random;
 2 import java.util.concurrent.BlockingQueue;
 3 import java.util.concurrent.TimeUnit;
 4  
 5 /**
 6  * 消费者线程
 7  * 
 8  * @author jackyuj
 9  */
10 public class Consumer implements Runnable {
11     
12     private BlockingQueue<String> queue;
13     private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;
14  
15     //构造函数
16     public Consumer(BlockingQueue<String> queue) {
17         this.queue = queue;
18     }
19  
20     public void run() {
21         System.out.println("启动消费者线程!");
22         Random r = new Random();
23         boolean isRunning = true;
24         try {
25             while (isRunning) {
26                 System.out.println("正从队列获取数据...");
27                 String data = queue.poll(2, TimeUnit.SECONDS);//有数据时直接从队列的队首取走,无数据时阻塞,在2s内有数据,取走,超过2s还没数据,返回失败
28                 if (null != data) {
29                     System.out.println("拿到数据:" + data);
30                     System.out.println("正在消费数据:" + data);
31                     Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
32                 } else {
33                     // 超过2s还没数据,认为所有生产线程都已经退出,自动退出消费线程。
34                     isRunning = false;
35                 }
36             }
37         } catch (InterruptedException e) {
38             e.printStackTrace();
39             Thread.currentThread().interrupt();
40         } finally {
41             System.out.println("退出消费者线程!");
42         }
43     }
44  
45     
46 }

3. DelayQueue

DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据的操作(消费者)才会被阻塞。

使用场景:

DelayQueue使用场景较少,但都相当巧妙,常见的例子比如使用一个DelayQueue来管理一个超时未响应的连接队列。

4. PriorityBlockingQueue

基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定),但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者,而只会在没有可消费的数据时,阻塞数据的消费者。因此使用的时候要特别注意,生产者生产数据的速度绝对不能快于消费者消费数据的速度,否则时间一长,会最终耗尽所有的可用堆内存空间。在实现PriorityBlockingQueue时,内部控制线程同步的锁采用的是公平锁。

5. SynchronousQueue

一种无缓冲的等待队列,类似于无中介的直接交易,有点像原始社会中的生产者和消费者,生产者拿着产品去集市销售给产品的最终消费者,而消费者必须亲自去集市找到所要商品的直接生产者,如果一方没有找到合适的目标,那么对不起,大家都在集市等待。相对于有缓冲的BlockingQueue来说,少了一个中间经销商的环节(缓冲区),如果有经销商,生产者直接把产品批发给经销商,而无需在意经销商最终会将这些产品卖给那些消费者,由于经销商可以库存一部分商品,因此相对于直接交易模式,总体来说采用中间经销商的模式会吞吐量高一些(可以批量买卖);但另一方面,又因为经销商的引入,使得产品从生产者到消费者中间增加了额外的交易环节,单个产品的及时响应性能可能会降低。

声明一个SynchronousQueue有两种不同的方式,它们之间有着不太一样的行为。公平模式和非公平模式的区别:

如果采用公平模式:SynchronousQueue会采用公平锁,并配合一个FIFO队列来阻塞多余的生产者和消费者,从而体系整体的公平策略;

但如果是非公平模式(SynchronousQueue默认):SynchronousQueue采用非公平锁,同时配合一个LIFO队列来管理多余的生产者和消费者,而后一种模式,如果生产者和消费者的处理速度有差距,则很容易出现饥渴的情况,即可能有某些生产者或者是消费者的数据永远都得不到处理。

五. 小结

BlockingQueue不光实现了一个完整队列所具有的基本功能,同时在多线程环境下,他还自动管理了多线间的自动等待于唤醒功能,从而使得程序员可以忽略这些细节,关注更高级的功能。

BlockingQueue是线程安全的

BlockingQueue是线程安全的,这句话是什么意思呢?

blockingQueue 可以实现以下功能:

1.当blockingQueue 的内容为空时,消费者自动阻塞;当blockingQueue 有元素的时候,被阻塞的消费者线程自动被唤醒。

2.当有多个生产者线程往blockingQueue 加值的时候,不需要考虑加锁,因为blockingQueue 自己的加值方法内部就已经使用了锁。例如:

 public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();

PriorityBlockingQueue

PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序
升序排列。也可以自定义类实现compareTo()方法来指定元素排序规则,或者初始化
PriorityBlockingQueue时,指定构造参数Comparator来对元素进行排序。需要注意的是不能保证
同优先级元素的顺序。

存储的元素 实现Comparable接口

package com.wp.queue;

public class Person implements Comparable<Person>{

	private String name;
	
	private int age;
	
	public Person(String name, int age) {
		super();
		this.name = name;
		this.age = age;
	}

	@Override
	public int compareTo(Person o) {
		
		return this.age-o.age;
	}

	@Override
	public String toString() {
		return "Person [name=" + name + ", age=" + age + "]";
	}
	
}

package com.wp.queue;

import java.util.concurrent.PriorityBlockingQueue;

public class MainTest {

	public static void main(String[] args) {
		PriorityBlockingQueue<Person> priorityBlockingQueue = new PriorityBlockingQueue();
		
		Person person1 = new Person("zhangsan", 20);
		Person person2 = new Person("lisi", 15);
		Person person3 = new Person("wangwu", 30);
		Person person4 = new Person("mazi", 10);
		
		priorityBlockingQueue.offer(person1);
		priorityBlockingQueue.offer(person2);
		priorityBlockingQueue.offer(person3);
		priorityBlockingQueue.offer(person4);
		
		
		while (priorityBlockingQueue.size() != 0) {
			System.out.println(priorityBlockingQueue.remove());
		}
		
	}
	
}


PriorityBlockingQueue的构造方法中加入Comparator

package com.wp.queue;

import java.util.Comparator;
import java.util.concurrent.PriorityBlockingQueue;

public class MainTestNext {

	public static void main(String[] args) {
		Comparator<Animl> comparator = 
				Comparator.comparing(Animl::getType, (s,t) -> Integer.compare(s.length(),t.length()));
		// PriorityBlockingQueue的构造方法中加入Comparator
		PriorityBlockingQueue<Animl> priorityBlockingQueue = new PriorityBlockingQueue<Animl>(20,comparator);
		
		Animl animl = new Animl("tiger", 10);
		Animl animl2 = new Animl("cat", 20);
		Animl animl3 = new Animl("alon", 30);
		
		priorityBlockingQueue.offer(animl);
		priorityBlockingQueue.offer(animl2);
		priorityBlockingQueue.offer(animl3);

		while (priorityBlockingQueue.size() != 0) {
			System.out.println(priorityBlockingQueue.remove());
		}
		
	}
	
}



package com.wp.queue;

public class Animl {

	private String type;
	
	private int number;

	public Animl(String type, int number) {
		super();
		this.type = type;
		this.number = number;
	}

	@Override
	public String toString() {
		return "Animl [type=" + type + ", number=" + number + "]";
	}

	public String getType() {
		return type;
	}

	public void setType(String type) {
		this.type = type;
	}

	public int getNumber() {
		return number;
	}

	public void setNumber(int number) {
		this.number = number;
	}
	
}

BlockingDequeu

1.BlockingQueue阻塞队列,FIFO一端进入队列,一端出队列

BlockingQueue,进出口单一明确。队列尾部进入队列,队列头部出队列。FIFO

2.BlockingDequeu 双端阻塞队列,两端都可以进队列,也可以出队列

putFirst-> 队列头部进入队列

takeFirst -> 队列头部出队列

putLast -> 队列尾部进入队列

takeLast -> 队列尾部出队列

public interface BlockingDeque extends BlockingQueue, Deque

双端阻塞队列,继承了FIFO特性,又继承了双端队列特性

BlockingDeque的方法

对于在队列中插入、删除和检查元素操作BlockingQueue有4类不同的方法。如果操作不能立即执行,每一类方法的行为将不同。下面是方法列表:

operation Throws Exception Special Value Blocks Times Out
Insert addFirst(o) offerFirst(o) putFirst(o) offerFirst(o, timeout, timeunit)
Remove removeFirst(o) pollFirst(o) takeFirst(o) pollFirst(timeout, timeunit)
Examine getFirst(o) peekFirst(o)

LinkedBlockingDeque类实现了BlockingDeque接口。
LinkedBlockingDeque是一个双端队列,如果一个线程试图从一个空的队列中取出元素,它将被阻塞住,不管线程是从哪一端去获取元素。

调用了getFirst和peekFirst 不会将元素从队列中移除。

队列的使用

1.队列如果为空,这个时候需要在消费者线程中,使用阻塞;同时注意,在生成者线程中使用通知,解除消费者的阻塞。

常用的实现方式有 condition 的await和signal

也可以使用 队列的阻塞方法 takeFirst

2.向队列添加元素时,需要判断队列中是否已经存在了此元素。

常用实现方式有:

  
 stream.anyMatch
  private boolean idExists(ActionTask task) {
        return actionTaskQueue.stream().anyMatch(actionTask -> actionTask.getId().equals(task.getId()));
    }
    
 stream.anyMatch
  private boolean idExists(ActionTask task) {
        return actionTaskQueue.stream().anyMatch(actionTask -> {return actionTask.getId().equals(task.getId()));};
    }

注意比较 上面2种写法,如果使用了{},那么需要加上return,如果不用{},那么就不用写return

  1. 队列元素的增减,应该有2种操作主体,一个是消费者可以从队列中取任务,一个是生产者既可以向队列添加任务,还可以将队列中的任务移除

    消费者:

    // 从队列中 移除任务
     actionTaskQueue.remove(task);
    
    

   

生产者:

```java
  
// 向队列中添加任务
actionTaskQueue.addFirst(task);
isAccept = true;

  // 打断当前任务
// 将队列中的任务移除
ActionTask first = actionTaskQueue.pollFirst();
LOG.info("cancel task {}", first);

评论