天天看点

SynchronousQueue,LinkedBlockingQueue,ArrayListBlockingQueue比较SynchronousQueue 在一个消息生产者和一个消息消费者模式中,在双核情况下,队列中一般只有一个元素时,其吞吐量是LinkedBlockingQueue 和 ArrayBlockingQueue 20倍。 LinkedBlockingQueue只有队列中元素增加时,其吞吐量才赶上,由此可见,SynchronousQueue在只有单个模式下,性能最优。

SynchronousQueue

        SynchronousQueue是无界的,是一种无缓冲的等待队列,但是由于该Queue本身的特性,在某次添加元素后必须等待其他线程取走后才能继续添加;可以认为SynchronousQueue是一个缓存值为1的阻塞队列,但是 isEmpty()方法永远返回是true,remainingCapacity() 方法永远返回是0,remove()和removeAll() 方法永远返回是false,iterator()方法永远返回空,peek()方法永远返回null。

        声明一个SynchronousQueue有两种不同的方式,它们之间有着不太一样的行为。公平模式和非公平模式的区别:如果采用公平模式:SynchronousQueue会采用公平锁,并配合一个FIFO队列来阻塞多余的生产者和消费者,从而体系整体的公平策略;但如果是非公平模式(SynchronousQueue默认):SynchronousQueue采用非公平锁,同时配合一个LIFO队列来管理多余的生产者和消费者,而后一种模式,如果生产者和消费者的处理速度有差距,则很容易出现饥渴的情况,即可能有某些生产者或者是消费者的数据永远都得不到处理。

它将任务直接提交给线程而不保持它们。在此,如果不存在可用于立即运行任务的线程,则试图把任务加入队列将失败,因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。

LinkedBlockingQueue

        LinkedBlockingQueue是无界的,是一个无界缓存的等待队列。

        基于链表的阻塞队列,内部维持着一个数据缓冲队列(该队列由链表构成)。当生产者往队列中放入一个数据时,队列会从生产者手中获取数据,并缓存在队列内部,而生产者立即返回;只有当队列缓冲区达到最大值缓存容量时(LinkedBlockingQueue可以通过构造函数指定该值),才会阻塞生产者队列,直到消费者从队列中消费掉一份数据,生产者线程会被唤醒,反之对于消费者这端的处理也基于同样的原理。

        LinkedBlockingQueue之所以能够高效的处理并发数据,还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步,这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据,以此来提高整个队列的并发性能。

使用无界队列(例如,不具有预定义容量的 LinkedBlockingQueue)将导致在所有corePoolSize 线程都忙时新任务在队列中等待。这样,创建的线程就不会超过 corePoolSize。(因此,maximumPoolSize的值也就无效了。)当每个任务完全独立于其他任务,即任务执行互不影响时,适合于使用无界队列;例如,在 Web页服务器中。这种排队可用于处理瞬态突发请求,当命令以超过队列所能处理的平均数连续到达时,此策略允许无界线程具有增长的可能性。

ArrayListBlockingQueue

         ArrayListBlockingQueue是有界的,是一个有界缓存的等待队列。

         基于数组的阻塞队列,同LinkedBlockingQueue类似,内部维持着一个定长数据缓冲队列(该队列由数组构成)。ArrayBlockingQueue内部还保存着两个整形变量,分别标识着队列的头部和尾部在数组中的位置。

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

使用有限的 maximumPoolSizes时,有界队列(如 ArrayBlockingQueue)有助于防止资源耗尽,但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷:使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销,但是可能导致人工降低吞吐量。如果任务频繁阻塞(例如,如果它们是 I/O边界),则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小,CPU使用率较高,但是可能遇到不可接受的调度开销,这样也会降低吞吐量。  

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

SynchronousQueue 在一个消息生产者和一个消息消费者模式中,在双核情况下,队列中一般只有一个元素时,其吞吐量是LinkedBlockingQueue 和 ArrayBlockingQueue 20倍。

LinkedBlockingQueue只有队列中元素增加时,其吞吐量才赶上,由此可见,SynchronousQueue在只有单个模式下,性能最优。

SynchronousQueue,LinkedBlockingQueue,ArrayListBlockingQueue比较SynchronousQueue 在一个消息生产者和一个消息消费者模式中,在双核情况下,队列中一般只有一个元素时,其吞吐量是LinkedBlockingQueue 和 ArrayBlockingQueue 20倍。 LinkedBlockingQueue只有队列中元素增加时,其吞吐量才赶上,由此可见,SynchronousQueue在只有单个模式下,性能最优。

例子一:使用直接提交策略,也即SynchronousQueue。

首先SynchronousQueue是无界的,也就是说他存数任务的能力是没有限制的,但是由于该Queue本身的特性,在某次添加元素后必须等待其他线程取走后才能继续添加。在这里不是核心线程便是新创建的线程,但是我们试想一样下,下面的场景。

我们使用一下参数构造ThreadPoolExecutor:

1.     new ThreadPoolExecutor(   

2.                 2, 3, 30, TimeUnit.SECONDS,    

3.                 new  SynchronousQueue<Runnable>(),    

4.                 new RecorderThreadFactory("CookieRecorderPool"),    

  1.             new ThreadPoolExecutor.CallerRunsPolicy());  

new ThreadPoolExecutor(

  2, 3, 30, TimeUnit.SECONDS,

  new SynchronousQueue<Runnable>(),

  new RecorderThreadFactory("CookieRecorderPool"),

  new ThreadPoolExecutor.CallerRunsPolicy());

 当核心线程已经有2个正在运行.

  1. 此时继续来了一个任务(A),根据前面介绍的“如果运行的线程等于或多于 corePoolSize,则Executor始终首选将请求加入队列,而不添加新的线程。”,所以A被添加到queue中。
  2. 又来了一个任务(B),且核心2个线程还没有忙完,OK,接下来首先尝试1中描述,但是由于使用的SynchronousQueue,所以一定无法加入进去。
  3. 此时便满足了上面提到的“如果无法将请求加入队列,则创建新的线程,除非创建此线程超出maximumPoolSize,在这种情况下,任务将被拒绝。”,所以必然会新建一个线程来运行这个任务。
  4. 暂时还可以,但是如果这三个任务都还没完成,连续来了两个任务,第一个添加入queue中,后一个呢?queue中无法插入,而线程数达到了maximumPoolSize,所以只好执行异常策略了。

所以在使用SynchronousQueue通常要求maximumPoolSize是无界的,这样就可以避免上述情况发生(如果希望限制就直接使用有界队列)。对于使用SynchronousQueue的作用jdk中写的很清楚:此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。

什么意思?如果你的任务A1,A2有内部关联,A1需要先运行,那么先提交A1,再提交A2,当使用SynchronousQueue我们可以保证,A1必定先被执行,在A1么有被执行前,A2不可能添加入queue中。

例子二:使用无界队列策略,即LinkedBlockingQueue

这个就拿newFixedThreadPool来说,根据前文提到的规则:

如果运行的线程少于 corePoolSize,则 Executor 始终首选添加新的线程,而不进行排队。那么当任务继续增加,会发生什么呢?

如果运行的线程等于或多于 corePoolSize,则 Executor 始终首选将请求加入队列,而不添加新的线程。OK,此时任务变加入队列之中了,那什么时候才会添加新线程呢?

如果无法将请求加入队列,则创建新的线程,除非创建此线程超出 maximumPoolSize,在这种情况下,任务将被拒绝。这里就很有意思了,可能会出现无法加入队列吗?不像SynchronousQueue那样有其自身的特点,对于无界队列来说,总是可以加入的(资源耗尽,当然另当别论)。换句说,永远也不会触发产生新的线程!corePoolSize大小的线程数会一直运行,忙完当前的,就从队列中拿任务开始运行。所以要防止任务疯长,比如任务运行的实行比较长,而添加任务的速度远远超过处理任务的时间,而且还不断增加,不一会儿就爆了。

例子三:有界队列,使用ArrayBlockingQueue。

这个是最为复杂的使用,所以JDK不推荐使用也有些道理。与上面的相比,最大的特点便是可以防止资源耗尽的情况发生。

举例来说,请看如下构造方法:

1.     new ThreadPoolExecutor(   

2.                 2, 4, 30, TimeUnit.SECONDS,    

3.                 new ArrayBlockingQueue<Runnable>(2),    

4.                 new RecorderThreadFactory("CookieRecorderPool"),    

5.                 new ThreadPoolExecutor.CallerRunsPolicy());  

new ThreadPoolExecutor(

    2, 4, 30, TimeUnit.SECONDS,

    new ArrayBlockingQueue<Runnable>(2),

    new RecorderThreadFactory("CookieRecorderPool"),

    new ThreadPoolExecutor.CallerRunsPolicy());

假设,所有的任务都永远无法执行完。

对于首先来的A,B来说直接运行,接下来,如果来了C,D,他们会被放到queue中,如果接下来再来E,F,则增加线程运行E,F。但是如果再来任务,队列无法再接受了,线程数也到达最大的限制了,所以就会使用拒绝策略来处理。

SynchronousQueue,LinkedBlockingQueue,ArrayListBlockingQueue比较SynchronousQueue 在一个消息生产者和一个消息消费者模式中,在双核情况下,队列中一般只有一个元素时,其吞吐量是LinkedBlockingQueue 和 ArrayBlockingQueue 20倍。 LinkedBlockingQueue只有队列中元素增加时,其吞吐量才赶上,由此可见,SynchronousQueue在只有单个模式下,性能最优。
SynchronousQueue,LinkedBlockingQueue,ArrayListBlockingQueue比较SynchronousQueue 在一个消息生产者和一个消息消费者模式中,在双核情况下,队列中一般只有一个元素时,其吞吐量是LinkedBlockingQueue 和 ArrayBlockingQueue 20倍。 LinkedBlockingQueue只有队列中元素增加时,其吞吐量才赶上,由此可见,SynchronousQueue在只有单个模式下,性能最优。