第1 部分配置ThreadPoolExecutor

1.1 创建

ThreadPoolExecutor为一些Executor提供了基本的实现，比如，newCachedThreadPool,newFixedThreadPool等等。ThreadPoolExecutor允许各种定制

它的构造函数如下：

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {}

最常用的newCachedThreadPool。

1.2 管理队列任务

据线程池的大小来选择合适的队列有利于充分利用资源和防止耗尽资源。

基本的排队方法有3种：

无界队列
有界队列
同步移交

对于Executor，newCachedThreadPool工厂方法是一种很好的默认选择，它能提供比固定大小的线程池更好的排队性能。

1.3 饱和策略

我们在ThreadPoolExecutor的构造函数中看到了最后一个参数。 RejectedExecutionHandler handler。这个就是饱和策略。

JDK提供了几种不同的实现：

DiscardOldestPolicy
AbortPolicy
CallerRunsPolicy
discardPolicy

AbortPolicy是默认的饱和策略，就是中止任务，该策略将抛出RejectedExecutionException。调用者可以捕获这个异常然后去编写代码处理异常。

当新提交的任务无法保存到队列中等待执行时，DiscardPolicy会稍稍的抛弃该任务，DiscardOldestPolicy则会抛弃最旧的（下一个将被执行的任务），然后尝试重新提交新的任务。如果工作队列是那个优先级队列时，搭配DiscardOldestPolicy饱和策略会导致优先级最高的那个任务被抛弃，所以两者不要组合使用。

CallerRunsPolicy是“调用者运行”策略，实现了一种调节机制。它不会抛弃任务，也不会抛出异常。而是将任务回退到调用者。它不会在线程池中执行任务，而是在一个调用了Executor的线程中执行该任务。

 1 /**
 2  * 调用者运行的饱和策略
 3  * @ClassName: ThreadDeadlock2
 4  * TODO
 5  * @author xingle
 6  * @date 2014-11-20 下午4:18:11
 7  */
 8 public class ThreadDeadlock2 {
 9     ExecutorService exec = new ThreadPoolExecutor(0, 2, 60L, TimeUnit.SECONDS,  
10             new SynchronousQueue<Runnable>(),new ThreadPoolExecutor.CallerRunsPolicy());
11 
12     private void putrunnable() {
13         for (int i = 0; i < 4; i++) {
14             exec.submit(new Runnable() {
15 
16                 @Override
17                 public void run() {
18                     // TODO Auto-generated method stub
19                     while (true) {
20                         System.out.println(Thread.currentThread().getName());
21                         try {
22                             Thread.sleep(500);
23                         } catch (InterruptedException e) {
24                             // TODO Auto-generated catch block
25                             e.printStackTrace();
26                         }
27                     }
28                 }
29             });
30         }
31     }
32     public static void main(String[] args) {
33         new ThreadDeadlock2().putrunnable();
34     }
35 }

执行结果：

当工作队列被填满之后，没有预定义的饱和策略来阻塞execute。通过使用 Semaphore （信号量）来限制任务的到达率，就可以实现这个功能。在下面的BoundedExecutor 中给出了这种方法，该方法使用了一个无界队列，并设置信号量的上界设置为线程池的大小加上可排队任务的数量，这是因为信号量需要控制正在执行的和正在等待执行的任务数量。

 1 /**
 2  * 8.4 使用Semaphore来控制任务的提交速率
 3  * @ClassName: BoundedExecutor
 4  * TODO
 5  * @author xingle
 6  * @date 2014-11-20 下午2:46:19
 7  */
 8 public class BoundedExecutor {
 9     private final Executor exec;
10     private final Semaphore semaphore;
11     int bound;
12 
13     public BoundedExecutor(Executor exec,int bound){
14         this.exec = exec;
15         this.semaphore = new Semaphore(bound);
16         this.bound = bound;
17     }
18     
19     public void submitTask(final Runnable command) throws InterruptedException{
20         //通过 acquire() 获取一个许可
21         semaphore.acquire();
22         System.out.println("----------当前可用的信号量个数:"+semaphore.availablePermits());
23         try {
24             exec.execute(new Runnable() {
25 
26                 @Override
27                 public void run() {
28                     try {
29                         System.out.println("线程" + Thread.currentThread().getName() +"进入，当前已有" + (bound-semaphore.availablePermits()) + "个并发");
30                         command.run();
31                     } finally {
32                         //release() 释放一个许可
33                         semaphore.release();
34                          System.out.println("线程" + Thread.currentThread().getName() +   
35                                     "已离开，当前已有" + (bound-semaphore.availablePermits()) + "个并发");  
36                     }
37                 }
38             });
39         } catch (RejectedExecutionException e) {
40             semaphore.release();
41         }        
42     }
43 }

测试程序：

 1 public class BoundedExecutor_main {
 2     public static void main(String[] args) throws InterruptedException{
 3         ExecutorService exec = Executors.newCachedThreadPool();    
 4         BoundedExecutor e = new BoundedExecutor(exec, 3);
 5         
 6         for(int i = 0;i<5;i++){
 7             final int c = i;
 8             e.submitTask(new Runnable() {
 9 
10                 @Override
11                 public void run() {
12                     System.out.println("执行任务:" +c);
13                 }
14             });
15         }
16     }
17 }

执行结果：

参考：

1.《java 并发编程》 8.2-8.3

2.聊聊并发（三）——JAVA线程池的分析和使用

3.Java 并发编程之线程池的使用（二）

4.Java线程(九)：Semaphore-信号灯机制

并发编程 13—— 线程池的使用 之 配置ThreadPoolExecutor 和 饱和策略

第1 部分 配置ThreadPoolExecutor

1.1 创建

1.2 管理队列任务

1.3 饱和策略

并发编程 13—— 线程池的使用之配置ThreadPoolExecutor 和饱和策略

第1 部分配置ThreadPoolExecutor