正文

基于ThreadPoolExecutor，自定义线程池简单实现

IP云
IP云V管理员/3 阅读
08/22
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

一、线程池作用

在上一篇随笔中有提到多线程具有同一时刻处理多个任务的特点，即并行工作，因此多线程的用途非常广泛，特别在性能优化上显得尤为重要。然而，多线程处理消耗的时间包括创建线程时间T1、工作时间T2、销毁线程时间T3，创建和销毁线程需要消耗一定的时间和资源，如果能够减少这部分的时间消耗，性能将会进一步提高，线程池就能够很好解决问题。线程池在初始化时会创建一定数量的线程，当需要线程执行任务时，从线程池取出线程，当任务执行完成后，线程置回线程池成为空闲线程，等待下一次任务。JDK1.5提供了一个Executors工厂类来产生线程池，该工厂类提供5种静态方法来创建线程池，详细请参见：http://www.cnblogs.com/firstsheng618/p/3861097.html。

二、认识队列

队列具有先进先出(FIFO)的特点，不同于堆的后进先出(LIFO)，队列（Queue）是只允许在一端进行插入，而在另一端进行删除的运算受限的线性表。线程池就是通过队列的方式实现任务的调用。下面介绍几个常用的队列：

1、ArrayBlockingQueue：一个有数组支持的有界阻塞队列。此队列按 FIFO（先进先出）原则对元素进行排序，队列的头部 是在队列中存在时间最长的元素，队列的尾部 是在队列中存在时间最短的元素，新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。

2、ConcurrentLinkedQueue：一个基于链接节点的无界线程安全队列。此队列按照 FIFO（先进先出）原则对元素进行排序，队列的头部 是队列中时间最长的元素，队列的尾部 是队列中时间最短的元素，新的元素插入到队列的尾部，队列获取操作从队列头部获得元素，当多个线程共享访问一个公共 collection 时，ConcurrentLinkedQueue 是一个恰当的选择，此队列不允许使用 null 元素。

3、LinkedBlockingQueue：一个基于已链接节点的、范围任意的BlockingQueue。此队列按 FIFO（先进先出）排序元素，队列的头部 是在队列中时间最长的元素，队列的尾部 是在队列中时间最短的元素，新元素插入到队列的尾部，并且队列获取操作会获得位于队列头部的元素，链接队列的吞吐量通常要高于基于数组的队列，但是在大多数并发应用程序中，其可预知的性能要低。

4、SynchronousQueue：一种阻塞队列，其中每个插入操作必须等待另一个线程的对应移除操作 ，反之亦然。

三、队列排队策略

1、ThreadPoolExecutor.AbortPolicy：用于被拒绝任务的处理程序，它将抛出 RejectedExecutionException，线程池默认被拒绝任务的处理策略。

2、ThreadPoolExecutor.CallerRunsPolicy：用于被拒绝任务的处理程序，它直接在 execute 方法的调用线程(上一层线程)中运行被拒绝的任务；如果执行程序已关闭，则会丢弃该任务。

3、ThreadPoolExecutor.DiscardOldestPolicy：用于被拒绝任务的处理程序，它放弃最旧的未处理请求，然后重试 execute；如果执行程序已关闭，则会丢弃该任务。

4、ThreadPoolExecutor.DiscardPolicy：用于被拒绝任务的处理程序，默认情况下它将丢弃被拒绝的任务。

四、自定义线程

线程池工作策略：A. 如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。B. 如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。C. 如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

ArrayBlockingQueue和LinkedBlockingQueue是两个最常用的阻塞队列，一般情况下足以处理多线程间的生产者和消费者问题，LinkedBlockingQueue内部分别采用独立锁来控制数据同步，实现生产者端和消费者端并行工作，高效的处理并发数据。

基于ThreadPoolExecutor自定义线程池如下：

public class LinkedBqThreadPool extends ThreadPoolExecutor {
protected Logger log = Logger.getLogger(getClass());
/**
* 正在执行任务数量
*/
private AtomicInteger taskNum = new AtomicInteger(0); /**
* 构建线程池
* @param corePoolSize 池中所保存的核心线程数
* @param maximumPoolSize 池中允许的最大线程数
* @param keepActiveTime 非核心线程空闲等待新任务的最长时间
* @param timeunit keepActiveTime参数的时间单位
* @param blockingqueue 任务队列
*/
public LinkedBqThreadPool(int corePoolSize, int maximumPoolSize, long keepActiveTime, TimeUnit timeunit,
BlockingQueue<Runnable> blockingqueue) {
super(corePoolSize, maximumPoolSize, keepActiveTime, timeunit, blockingqueue);
} /**
* 构建线程池
* @param corePoolSize 池中所保存的核心线程数
* @param maximumPoolSize 池中允许的最大线程数
* @param keepActiveTime 非核心线程空闲等待新任务的最长时间(单位：秒)
* @param blockingqueue 任务队列
*/
public LinkedBqThreadPool(int corePoolSize, int maximumPoolSize, long keepActiveTime,
BlockingQueue<Runnable> blockingqueue) {
this(corePoolSize, maximumPoolSize, keepActiveTime, TimeUnit.SECONDS, blockingqueue);
} /**
* 构建线程池
* @param corePoolSize 池中所保存的核心线程数
* @param maximumPoolSize 池中允许的最大线程数
* @param keepActiveTime 非核心线程空闲等待新任务的最长时间(单位：秒)
*/
public LinkedBqThreadPool(int corePoolSize, int maximumPoolSize, long keepActiveTime) {
this(corePoolSize, maximumPoolSize, keepActiveTime, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
} /**
* 构建单线程的线程池
*/
public LinkedBqThreadPool() {
this(1, 1, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>());
} /**
* 任务执行，以原子方式将当前值加 1
*/
public void execute(Runnable task) {
taskNum.getAndIncrement();
super.execute(task);
} /**
* 任务执行之后
*/
public void afterExecute(Runnable task, Throwable throwable) {
taskNum.decrementAndGet();
log.debug("task : " + task.getClass().getSimpleName()
+ " completed,Throwable:" + throwable + ",taskNum:" + getTaskNum());
synchronized(this) {
notifyAll();
}
} /**
* 挂起当前线程，直到所有任务执行完成
*/
public void waitComplete() {
try {
synchronized(this){
while(getTaskNum() > 0){
wait(500);
}
}
} catch (InterruptedException e) {
log.error(e + ", taskNum:" + getTaskNum());
}
} /**
* @return 未执行的任务数
*/
public int getTaskNum() {
return taskNum.get();
} /**
* @param time 非核心线程空闲等待新任务的最长时间(单位：秒)
*/
public void setKeepAliveTime(int time) {
super.setKeepAliveTime(time, TimeUnit.SECONDS);
} /**
* @param size 池中所保存的核心线程数
*/
public void setCorePoolSize(int size) {
super.setCorePoolSize(size);
} /**
* @param size 池中允许的最大线程数
*/
public void setMaximumPoolSize(int size) {
super.setMaximumPoolSize(size);
}
}

LinkedBlockingQueue队列实现

public class ArrayBqThreadPool extends ThreadPoolExecutor {
protected Logger log = Logger.getLogger(getClass());
/**
* 待执行任务数量
*/
private AtomicInteger taskNum = new AtomicInteger(0); /**
* 构建线程池
* @param corePoolSize 池中所保存的核心线程数
* @param maximumPoolSize 池中允许的最大线程数
* @param keepActiveTime 非核心线程空闲等待新任务的最长时间(单位：秒)
* @param queueCapacity 队列容量，即等待执行任务数
*/
public ArrayBqThreadPool(int corePoolSize, int maximumPoolSize, long keepActiveTime, int queueCapacity) {
super(corePoolSize, maximumPoolSize, keepActiveTime, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueCapacity));
} /**
* 构建单线程的线程池
*/
public ArrayBqThreadPool(int queueCapacity) {
this(1, 1, 1, queueCapacity);
} /**
* 任务执行，以原子方式将当前值加 1
*/
public void execute(Runnable task) {
taskNum.getAndIncrement();
super.execute(task);
} /**
* 任务执行之后
*/
public void afterExecute(Runnable task, Throwable throwable) {
taskNum.decrementAndGet();
log.debug("task : " + task.getClass().getSimpleName()
+ " completed,Throwable:" + throwable + ",taskNum:" + getTaskNum());
synchronized(this) {
notifyAll();
}
} /**
* 挂起当前线程，直到所有任务执行完成
*/
public void waitComplete() {
try {
synchronized(this){
while(getTaskNum() > 0){
wait(500);
}
}
} catch (InterruptedException e) {
log.error(e + ", taskNum:" + getTaskNum());
}
} /**
* @return 待执行的任务数
*/
public int getTaskNum() {
return taskNum.get();
} /**
* @param time 非核心线程空闲等待新任务的最长时间(单位：秒)
*/
public void setKeepAliveTime(int time) {
super.setKeepAliveTime(time, TimeUnit.SECONDS);
} /**
* @param size 池中所保存的核心线程数
*/
public void setCorePoolSize(int size) {
super.setCorePoolSize(size);
} /**
* @param size 池中允许的最大线程数
*/
public void setMaximumPoolSize(int size) {
super.setMaximumPoolSize(size);
}
}

ArrayBlockingQueue队列实现