Java线程池策略

线程池在Java开发中随处可见，其执行策略可以总结如下：

当提交一个新任务到线程池时：

1. 判断核心线程数是否已满，未满则创建一个新的线程来执行任务
2. 否则判断工作队列是否已满，未满则加入队列
3. 否则判断线程数是否以达到最大线程，没有则创建一个新的线程来执行任务
4. 否则交给饱和策略来处理

源码分析就不展开了。

一般来说，我们建议使用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密集型的任务，可能更适合使用这样策略的线程池。