Java中的线程池

实现原理

ThreadPoolExecutor执行execute：

如果当前运行的线程少于corePoolSize，则创建新线程来执行任务（需要获取全局锁）。
如果运行的线程等于或多余corePoolSize，则将任务加入BlockingQueue
如果无法将任务加入BlockingQueue（队列已满），则创建新的线程来处理任务（需要获取全局锁）。
如果创建新线程将使当前运行的线程超过maximumPoolSize，任务将被拒绝，并调用RejectedExecutionHandler.rejectedExecution()方法。

线程池的使用

创建

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

workQueue：任务队列(ArrayBlockingQueue、LinkedBlockingQueue、SynchronousQueue、PriorityBlockingQueue)
maximumPoolSize: 如果使用了无界的任务队列这个参数就没哟什么效果
RejectedExecutionHandler：饱和策略

提交任务

execute：无返回值
sumbit：有返回值

/**
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 */
public Future<?> submit(Runnable task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<Void> ftask = newTaskFor(task, null);
    execute(ftask);
    return ftask;
}

/**
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 */
public <T> Future<T> submit(Runnable task, T result) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task, result);
    execute(ftask);
    return ftask;
}

/**
 * @throws RejectedExecutionException {@inheritDoc}
 * @throws NullPointerException       {@inheritDoc}
 */
public <T> Future<T> submit(Callable<T> task) {
    if (task == null) throw new NullPointerException();
    RunnableFuture<T> ftask = newTaskFor(task);
    execute(ftask);
    return ftask;
}

protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
    return new FutureTask<T>(runnable, value);
}


public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    /*
     * Proceed in 3 steps:
     *
     * 1. If fewer than corePoolSize threads are running, try to
     * start a new thread with the given command as its first
     * task.  The call to addWorker atomically checks runState and
     * workerCount, and so prevents false alarms that would add
     * threads when it shouldn't, by returning false.
     *
     * 2. If a task can be successfully queued, then we still need
     * to double-check whether we should have added a thread
     * (because existing ones died since last checking) or that
     * the pool shut down since entry into this method. So we
     * recheck state and if necessary roll back the enqueuing if
     * stopped, or start a new thread if there are none.
     *
     * 3. If we cannot queue task, then we try to add a new
     * thread.  If it fails, we know we are shut down or saturated
     * and so reject the task.
     */
    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

通过线程池submit的任务，最终都被包裹成FutureTask，并且FutureTask实现了Runnable接口。
在调用execute方法后，将任务添加到工作队列中。

关闭线程池(shutdown、shutdownNow)

原理：遍历线程池中的工作线程，逐个调用线程的interrupt方法来终端，所以无法响应中断的任务可能永远无法终止。
区别：shutdown将线程池状态设为SHUTDOWN，然后中断所有没在执行的任务；shutdownNow首先将线程池的状态设为STOP，然后尝试停止所有正在执行或暂停任务的线程，并返回等待执行任务的列表。

合理配置线程池

任务特性

性质：CPU密集型任务、IO密集型任务、混合型任务
优先级：高中低
执行时间：长、短
依赖性：依赖其他资源，入数据库连接

CPU密集型任务应配置尽可能小的线程，如配置coreNum + 1个线程的线程池。
IO密集型应配置尽可能多的线程，如2coreNum，由于不是一直在执行任务。
混合型的任务，如何可以拆分，将其拆分为一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐量将高于串行执行的吞吐量。如果这两个执行时间相差太大，则没必要进行分解。
可以通过Runtime.getRuntime().availableProcessors()来获取当前设备的CPU个数。
建议使用有界队列，防止撑爆内存，拖垮系统，影响其他的任务。

线程池的监控

通过自定义线程池，重写线程池的beforeExecute、afterExecute和terminated方法，也可以在任务执行前、后和线程池给关闭前执行一些代码来进行监控。