Executors

最大线程数：

32bit：1450

64bit：100000

http://blog.csdn.net/lipc_/article/details/52025993

http://blog.chinaunix.net/uid-20577907-id-3519578.html

http://825635381.iteye.com/blog/2184680

http://dongxuan.iteye.com/blog/901689

ThreadPoolExecutor是Executors类的底层实现。

在JDK帮助文档中，有如此一段话：

“

强烈建议程序员使用较为方便的

Executors

工厂方法

Executors.newCachedThreadPool()

（无界线程池，可以进行自动线程回收）、

Executors.newFixedThreadPool(int)

（固定大小线程池）和

Executors.newSingleThreadExecutor()

（单个后台线程），它们均为大多数使用场景预定义了设置。

”

总结：

1、用ThreadPoolExecutor自定义线程池，看线程是的用途，如果任务量不大，可以用无界队列，如果任务量非常大，要用有界队列，防止OOM

2、如果任务量很大，还要求每个任务都处理成功，要对提交的任务进行阻塞提交，重写拒绝机制，改为阻塞提交。保证不抛弃一个任务

3、最大线程数一般设为2N+1最好，N是CPU核数

4、核心线程数，看应用，如果是任务，一天跑一次，设置为0，合适，因为跑完就停掉了，如果是常用线程池，看任务量，是保留一个核心还是几个核心线程数

5、如果要获取任务执行结果，用CompletionService，但是注意，获取任务的结果的要重新开一个线程获取，如果在主线程获取，就要等任务都提交后才获取，就会阻塞大量任务结果，队列过大OOM，所以最好异步开个线程获取结果

http://blog.chinaunix.net/uid-20577907-id-3519578.html

http://825635381.iteye.com/blog/2184680

一、简介

线程池类为 java.util.concurrent.ThreadPoolExecutor，常用构造方法为：

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,

long keepAliveTime, TimeUnit unit,

BlockingQueue

workQueue,

RejectedExecutionHandler handler)

corePoolSize： 线程池维护线程的最少数量

maximumPoolSize：线程池维护线程的最大数量

keepAliveTime： 线程池维护线程所允许的空闲时间

unit： 线程池维护线程所允许的空闲时间的单位

workQueue： 线程池所使用的缓冲队列

handler： 线程池对拒绝任务的处理策略

一个任务通过 execute(Runnable)方法被添加到线程池，任务就是一个 Runnable类型的对象，任务的执行方法就是 Runnable类型对象的run()方法。

当一个任务通过execute(Runnable)方法欲添加到线程池时：

如果此时线程池中的数量小于corePoolSize，即使线程池中的线程都处于空闲状态，也要创建新的线程来处理被添加的任务。

如果此时线程池中的数量等于 corePoolSize，但是缓冲队列 workQueue未满，那么任务被放入缓冲队列。

如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。

如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过 handler所指定的策略来处理此任务。

也就是：处理任务的优先级为：

核心线程corePoolSize、任务队列workQueue、最大线程maximumPoolSize，如果三者都满了，使用handler处理被拒绝的任务。

当线程池中的线程数量大于 corePoolSize时，如果某线程空闲时间超过keepAliveTime，线程将被终止。这样，线程池可以动态的调整池中的线程数。

unit可选的参数为java.util.concurrent.TimeUnit中的几个静态属性：

NANOSECONDS、MICROSECONDS、MILLISECONDS、SECONDS。

workQueue我常用的是：java.util.concurrent.ArrayBlockingQueue

handler有四个选择：

ThreadPoolExecutor.AbortPolicy()

抛出java.util.concurrent.RejectedExecutionException异常

ThreadPoolExecutor.CallerRunsPolicy()

重试添加当前的任务，他会自动重复调用execute()方法

ThreadPoolExecutor.DiscardOldestPolicy()

抛弃旧的任务

ThreadPoolExecutor.DiscardPolicy()

抛弃当前的任务

二、一般用法举例

点击(

此处

)折叠或打开

1. package

   demo

   ;

2. 1. import

      java

      .

      io

      .

      Serializable

      ;

3. import

   java

   .

   util

   .

   concurrent

   .

   ArrayBlockingQueue

   ;

4. import

   java

   .

   util

   .

   concurrent

   .

   ThreadPoolExecutor

   ;

5. import

   java

   .

   util

   .

   concurrent

   .

   TimeUnit

   ;

6. 1. public

      class

      TestThreadPool2

7. {

8. private

   static

   int

   produceTaskSleepTime

   =

   2

   ;

9. private

   static

   int

   produceTaskMaxNumber

   =

   10

   ;

10. 1. public

       static

       void

       main

       (

       String

       []

       args

       )

11. {

12. // 构造一个线程池

13. ThreadPoolExecutor

    threadPool

    =

    new

    ThreadPoolExecutor

    (

    2

    ,

    4

    ,

    3

    ,

    TimeUnit

    .

    SECONDS

    ,

    new

    ArrayBlockingQueue

    <

    Runnable

    >

    (

    3

    ),

14. new

    ThreadPoolExecutor

    .

    DiscardOldestPolicy

    ());

15. 1. for

       (

       int

       i

       =

       1

       ;

       i

       <

       =

       produceTaskMaxNumber

       ;

       i

       ++)

16. {

17. try

18. {

19. // 产生一个任务，并将其加入到线程池

20. String

    task

    =

    "task@ "

    +

    i

    ;

21. System

    .

    out

    .

    println

    (

    "put "

    +

    task

    );

22. threadPool

    .

    execute

    (

    new

    ThreadPoolTask

    (

    task

    ));

23. 1. // 便于观察，等待一段时间

24. Thread

    .

    sleep

    (

    produceTaskSleepTime

    );

25. }

26. catch

    (

    Exception

    e

    )

27. {

28. e

    .

    printStackTrace

    ();

29. }

30. }

31. }

32. }

33. 1. /**

34. * 线程池执行的任务

35. */

36. class

    ThreadPoolTask

    implements

    Runnable

    ,

    Serializable

37. {

38. private

    static

    final

    long

    serialVersionUID

    =

    0

    ;

39. private

    static

    int

    consumeTaskSleepTime

    =

    2000

    ;

40. // 保存任务所需要的数据

41. private

    Object

    threadPoolTaskData

    ;

42. 1. ThreadPoolTask

       (

       Object

       tasks

       )

43. {

44. this

    .

    threadPoolTaskData

    =

    tasks

    ;

45. }

46. 1. public

       void

       run

       ()

47. {

48. // 处理一个任务，这里的处理方式太简单了，仅仅是一个打印语句

49. System

    .

    out

    .

    println

    (

    Thread

    .

    currentThread

    ().

    getName

    ());

50. System

    .

    out

    .

    println

    (

    "start .."

    +

    threadPoolTaskData

    );

51. 1. try

52. {

53. // //便于观察，等待一段时间

54. Thread

    .

    sleep

    (

    consumeTaskSleepTime

    );

55. }

56. catch

    (

    Exception

    e

    )

57. {

58. e

    .

    printStackTrace

    ();

59. }

60. threadPoolTaskData

    =

    null

    ;

61. }

62. 1. public

       Object

       getTask

       ()

63. {

64. return

    this

    .

    threadPoolTaskData

    ;

65. }

66. }

说明：

1、在这段程序中，一个任务就是一个Runnable类型的对象，也就是一个ThreadPoolTask类型的对象。

2、一般来说任务除了处理方式外，还需要处理的数据，处理的数据通过构造方法传给任务。

3、在这段程序中，main()方法相当于一个残忍的领导，他派发出许多任务，丢给一个叫 threadPool的任劳任怨的小组来做。

这个小组里面队员至少有两个，如果他们两个忙不过来，任务就被放到任务列表里面。

如果积压的任务过多，多到任务列表都装不下(超过3个)的时候，就雇佣新的队员来帮忙。但是基于成本的考虑，不能雇佣太多的队员，至多只能雇佣 4个。

如果四个队员都在忙时，再有新的任务，这个小组就处理不了了，任务就会被通过一种策略来处理，我们的处理方式是不停的派发，直到接受这个任务为止(更残忍！呵呵)。

因为队员工作是需要成本的，如果工作很闲，闲到 3SECONDS都没有新的任务了，那么有的队员就会被解雇了，但是，为了小组的正常运转，即使工作再闲，小组的队员也不能少于两个。

4、通过调整 produceTaskSleepTime和 consumeTaskSleepTime的大小来实现对派发任务和处理任务的速度的控制，改变这两个值就可以观察不同速率下程序的工作情况。

5、通过调整4中所指的数据，再加上调整任务丢弃策略，换上其他三种策略，就可以看出不同策略下的不同处理方式。

6、对于其他的使用方法，参看jdk的帮助，很容易理解和使用。

先看一副图，描述了ThreadPoolExecutor的工作机制:

整个ThreadPoolExecutor的任务处理有4步操作：

• 第一步，初始的poolSize

  <

  corePoolSize，提交的runnable任务，会直接做为new一个Thread的参数，立马执行

• 第二步，当提交的任务数超过了corePoolSize，就进入了第二步操作。会将当前的runable提交到一个block queue中

• 第三步，如果block queue是个有界队列，当队列满了之后就进入了第三步。如果poolSize

  <

  maximumPoolsize时，会尝试new 一个Thread的进行救急处理，立马执行对应的runnable任务

• 第四步，如果第三步救急方案也无法处理了，就会走到第四步执行reject操作。

几点说明：(相信这些网上一搜一大把，我这里简单介绍下，为后面做一下铺垫)

• block queue有以下几种实现：

• ArrayBlockingQueue :

有界的数组队列

2.

LinkedBlockingQueue : 可支持有界/无界的队列，使用链表实现

3.

PriorityBlockingQueue : 优先队列，可以针对任务排序

4.

SynchronousQueue : 队列长度为1的队列，和Array有点区别就是：client thread提交到block queue会是一个阻塞过程，直到有一个worker thread连接上来poll task。

• RejectExecutionHandler是针对任务无法处理时的一些自保护处理：

• Reject 直接抛出Reject exception

• Discard 直接忽略该runnable，不可取

• DiscardOldest 丢弃最早入队列的的任务

• CallsRun 直接让原先的client thread做为worker线程，进行执行

容易被人忽略的点：

1. pool threads启动后，以后的任务获取都会通过block queue中，获取堆积的runnable task.

所以建议：

block size

>

= corePoolSize ，不然线程池就没任何意义

1. corePoolSize 和

maximumPoolSize的区别， 和大家正常理解的数据库连接池不太一样。

*

据dbcp pool为例，会有minIdle , maxActive配置。minIdle代表是常驻内存中的threads数量，maxActive代表是工作的最大线程数。

*

这里的corePoolSize就是连接池的maxActive的概念，它没有minIdle的概念(每个线程可以设置keepAliveTime，超过多少时间多有任务后销毁线程，但不会固定保持一定数量的threads)。

* 这里的maximumPoolSize，是一种救急措施的第一层。当threadPoolExecutor的工作threads存在满负荷，并且block queue队列也满了，这时代表接近崩溃边缘。这时允许临时起一批threads，用来处理runnable，处理完后立马退出。

所以建议：

maximumPoolSize

>

=

corePoolSize =期望的最大线程数。 (我曾经配置了corePoolSize=1,

maximumPoolSize=20, blockqueue为无界队列，最后就成了单线程工作的pool。典型的配置错误)

1. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler，从字面意思就是让调用者自己来运行。

我们经常会在线上使用一些线程池做异步处理，比如我前面做的

(业务层)异步并行加载技术分析和设计

,

将原本串行的请求都变为了并行操作，但过多的并行会增加系统的负载(比如软中断，上下文切换)。所以肯定需要对线程池做一个size限制。但是为了引入异步操作后，避免因在block queue的等待时间过长，所以需要在队列满的时，执行一个callsRun的策略，并行的操作又转为一个串行处理，这样就可以保证尽量少的延迟影响。

所以建议：

RejectExecutionHandler =

CallsRun ,

blockqueue size = 2 *

poolSize (为啥是2倍poolSize，主要一个考虑就是瞬间高峰处理，允许一个thread等待一个runnable任务)