concurrent: ThreadPoolExecutor 用法

   thread pool一般被用来解决两个问题：当处理大量的同步task的时候，它能够避免thread不断创建销毁的开销；而另外一个也许更重要的含义是，它其实表示了一个boundary，通过使用thread pool可以限制这些任务所消耗的资源,比如最大线程数，比如最大的消息缓冲池。
      需要指出的是，ThreadPoolExecutor不仅仅是简单的多个thread的集合，它还带有一个消息队列。
在Java中，如果只是需要一个简单的thread pool，ExecuteService可能更为合适，这是一个Interface。可以通过调用Executor的静态方法来获得一些简单的threadpool，如：
      ExecuteService pool = Executors.newFixedThreadPool(poolSize); 
      但如果要用定制的thread pool，则要使用ThreadPoolExecutor类

 

ThreadPoolExecutor的工作机制: 

 

 

构造方法为：

 
       ThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
              long keepAliveTime, TimeUnit unit,
              BlockingQueue<Runnable> workQueue,
              RejectedExecutionHandler handler)

参数说明：

 corePoolSize ： 线程池维护线程的最少数量
 maximumPoolSize ：线程池维护线程的最大数量
 keepAliveTime ： 线程池维护线程所允许的空闲时间
 unit ： 线程池维护线程所允许的空闲时间的单位
 workQueue ： 线程池所使用的缓冲队列
 handler ： 线程池对拒绝任务的处理策略

方法说明

 

 execute(Runnable):一个任务通过 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() ：抛弃当前的任务

 

网站摘取使用提示：转载自：http://www.iteye.com/topic/1118660

1.   pool threads启动后，以后的任务获取都会通过block queue中，获取堆积的runnable task.
    所以建议： block size >= corePoolSize ，不然线程池就没任何意义
2.  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。典型的配置错误)
   
3. 善用blockqueue和reject组合. 这里要重点推荐下CallsRun的Rejected Handler，从字面意思就是让调用者自己来运行。
    我们经常会在线上使用一些线程池做异步处理，比如我前面做的(业务层)异步并行加载技术分析和设计, 将原本串行的请求都变为了并行操作，但过多的并行会增加系统的负载(比如软中断，上下文切换)。所以肯定需要对线程池做一个size限制。但是为了引入异步操作后，避免因在block queue的等待时间过长，所以需要在队列满的时，执行一个callsRun的策略，并行的操作又转为一个串行处理，这样就可以保证尽量少的延迟影响。
    所以建议：  RejectExecutionHandler = CallsRun ,  blockqueue size = 2 * poolSize (为啥是2倍poolSize，主要一个考虑就是瞬间高峰处理，允许一个thread等待一个runnable任务)

一个简单的例子

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class TestThreadPool {

	private static int produceTaskSleepTime = 2;
	
	private static int produceTaskMaxNumber = 10;

	public static void main(String[] args) {

		// 构造一个线程池
		ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 4, 3,
				TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(3),
				new ThreadPoolExecutor.DiscardOldestPolicy());

		for (int i = 1; i <= produceTaskMaxNumber; i++) {
			try {
				String task = "task@ " + i;
				System.out.println("创建任务并提交到线程池中：" + task);
				threadPool.execute(new ThreadPoolTask(task));

				Thread.sleep(produceTaskSleepTime);
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
}

 

创建task类：

 

import java.io.Serializable;

public class ThreadPoolTask implements Runnable, Serializable {

	private Object attachData;

	ThreadPoolTask(Object tasks) {
		this.attachData = tasks;
	}

	public void run() {
		
		System.out.println("开始执行任务：" + attachData);
		
		attachData = null;
	}

	public Object getTask() {
		return this.attachData;
	}
}

 

执行结果：

               创建任务并提交到线程池中：task@ 1

开始执行任务：task@ 1

创建任务并提交到线程池中：task@ 2

开始执行任务：task@ 2

创建任务并提交到线程池中：task@ 3

创建任务并提交到线程池中：task@ 4

开始执行任务：task@ 3

创建任务并提交到线程池中：task@ 5

开始执行任务：task@ 4

创建任务并提交到线程池中：task@ 6

创建任务并提交到线程池中：task@ 7

创建任务并提交到线程池中：task@ 8

开始执行任务：task@ 5

开始执行任务：task@ 6

创建任务并提交到线程池中：task@ 9

开始执行任务：task@ 7

创建任务并提交到线程池中：task@ 10

开始执行任务：task@ 8

开始执行任务：task@ 9

开始执行任务：task@ 10

 

评论

1 楼 raoyutao 2012-09-13  

好