Java线程池策略
线程池在Java开发中随处可见，其执行策略可以总结如下：
当提交一个新任务到线程池时：
- 判断核心线程数是否已满，未满则创建一个新的线程来执行任务
- 否则判断工作队列是否已满，未满则加入队列
- 否则判断线程数是否以达到最大线程，没有则创建一个新的线程来执行任务
- 否则交给饱和策略来处理
源码分析就不展开了。
一般来说，我们建议使用ThreadPoolExecutor来创建线程池，见如下的构造函数：
public ThreadPoolExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory,
RejectedExecutionHandler handler)
每个参数具体是什么含义，注释里面已经描述的很清楚了，通过这些参数的配和置，来创建我们所需要的线程池。
Executors中通过静态工厂方法提供了几种基本的线程池创建方法，这里取两种比较典型的简单介绍一下：
可重用固定线程数的线程池
可重用固定线程数的线程池即为FixedThreadPool.
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
corePoolSize和maximumPoolSize设置成相同，keepAliveTime设置为0L。也就是说当有新任务时，
如果线程数线程数corePoolSize创建新的线程，而空闲时多余的线程不会被回收。
FixedThreadPool队列配置为无界队列，有可能大量任务堆积，撑爆内存。
按需创建的线程池
根据需要创建新线程的线程池即为CachedThreadPool。
public static ExecutorService newCachedThreadPool() {
return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
60L, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>());
}
corePoolSize 为0，maximumPoolSize为Integer.MAX_VALUE.keepAliveTime设置为60s, 空闲线程超过60秒后会被回收。但是使用没有容量的SynchronousQueue做为工作队列。
这意味着如果提交任务的速度大于任务处理速度，会不断的创建新的线程，甚至耗尽CPU和内存资源。
这两种线程池的配置都太绝对了，因此建议使用ThreadPoolExecutor来创建线程池，限定更多的参数配置，以符合业务，资源的要求。比如：
- 对于FixedThreadPool，我们可以限制队列的大小，避免任务的无限堆积。
- 对于CachedThreadPool，我们可以限制maximumPoolSize，避免线程的无限创建。
但是，我们能否有一种全新的策略，
- 对于可重用固定线程数的线程池，有固定数量的线程，也可以有一定数量线程扩展的能力，毕竟有些业务场景不能容忍延时，无法放入队列。
- 对于按需创建的线程池，限定了最大线程数后，也可以有一定的队列缓存能力。
这样就需要一种新的线程池策略，使我们可以先优先扩充线程到maximumPoolSize，再offer到queue，否则执行拒绝逻辑。感谢开源，在tomcat和matan中的源码都发现了这种实现，分享之：
NewThreadExecutor
首先，自定义实现一个队列，如果线程还未达到maximumPoolSize，拒绝将任务加入队列，这里选用LinkedTransferQueue是因为其有更好的性能。
public class TaskQueue extends LinkedTransferQueue<Runnable> {
private transient volatile NewThreadExecutor executor;
public TaskQueue() {
super();
}
public void setNewThreadExecutor(NewThreadExecutor threadPoolExecutor) {
this.executor = threadPoolExecutor;
}
public boolean offer(Runnable o) {
int poolSize = executor.getPoolSize();
// we are maxed out on threads, simply queue the object
if (poolSize == executor.getMaximumPoolSize()) {
return super.offer(o);
}
// we have idle threads, just add it to the queue
// note that we don't use getActiveCount(), see BZ 49730
if (executor.getSubmittedTasksCount() <= poolSize) {
return super.offer(o);
}
// if we have less threads than maximum force creation of a new
// thread
if (poolSize < executor.getMaximumPoolSize()) {
return false;
}
// if we reached here, we need to add it to the queue
return super.offer(o);
}
public boolean force(Runnable o, long timeout, TimeUnit unit) {
if (executor.isShutdown()) {
throw new RejectedExecutionException("Executor not running, can't force a command into the queue");
}
// forces the item onto the queue, to be used if the task is rejected
return super.offer(o, timeout, unit);
}
}
然后继承实现ThreadPoolExecutor类，覆写其execute方法，源码如下
public class NewThreadExecutor extends ThreadPoolExecutor {
private AtomicInteger submittedTasksCount;
private int maxSubmittedTaskCount;
public NewThreadExecutor(int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
int queueCapacity,
ThreadFactory threadFactory,
RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, new TaskQueue(), threadFactory, handler);
((TaskQueue) getQueue()).setNewThreadExecutor(this);
submittedTasksCount = new AtomicInteger(0);
maxSubmittedTaskCount = queueCapacity + maximumPoolSize;
}
public void execute(Runnable command) {
int count = submittedTasksCount.incrementAndGet();
// There is no length limit for the LinkedTransferQueue
if (count > maxSubmittedTaskCount) {
submittedTasksCount.decrementAndGet();
getRejectedExecutionHandler().rejectedExecution(command, this);
}
try {
super.execute(command);
} catch (RejectedExecutionException rx) {
// there could have been contention around the queue
if (!((TaskQueue) getQueue()).force(command, 0, TimeUnit.MILLISECONDS)) {
submittedTasksCount.decrementAndGet();
getRejectedExecutionHandler().rejectedExecution(command, this);
}
} catch (Throwable t) {
submittedTasksCount.decrementAndGet();
throw t;
}
}
public int getSubmittedTasksCount() {
return this.submittedTasksCount.get();
}
protected void afterExecute(Runnable r, Throwable t) {
submittedTasksCount.decrementAndGet();
}
}
在某些场景下，比如依赖了远程资源而非CPU密集型的任务，可能更适合使用这样策略的线程池。
