Java-线程池

Java中的线程池

线程池是一个管理线程的池子，它的作用有以下几点：

• 管理线程，避免增加创建线程和销毁线程的资源损耗
• 提高响应速度
• 重复利用

创建线程池

常见的创建线程池的方式有以下几种：

Executors.newCachedThreadPool()

• 核心线程数为0
• 最大线程数为Integer.MAX_VALUE
• 阻塞队列是SynchronousQueue
• 非核心线程空闲存活时间为60s

/**
     * Creates a thread pool that creates new threads as needed, but
     * will reuse previously constructed threads when they are
     * available.  These pools will typically improve the performance
     * of programs that execute many short-lived asynchronous tasks.
     * Calls to {@code execute} will reuse previously constructed
     * threads if available. If no existing thread is available, a new
     * thread will be created and added to the pool. Threads that have
     * not been used for sixty seconds are terminated and removed from
     * the cache. Thus, a pool that remains idle for long enough will
     * not consume any resources. Note that pools with similar
     * properties but different details (for example, timeout parameters)
     * may be created using {@link ThreadPoolExecutor} constructors.
     *
     * @return the newly created thread pool
     */
    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

当提交任务的速度大于处理任务的速度时，每次提交一个任务，就必然会创建一个线程。极端情况下会创建过多的线程，耗尽CPU和内存资源。由于空闲60秒的线程会被终止，长时间保持空闲的CachedThreadPool不会占用任何资源

它的工作步骤是：

1. 提交任务
2. 判断是否有核心线程，没有的话，任务加载到SynchronousQueue队列
3. 有的话，判断是否有空闲线程，如果有，就取出任务执行
4. 如果没有空闲线程，就新建一个线程执行
5. 执行完任务的线程，还可以存活60秒，如果在这期间，接到任务，可以继续存活下去，否则，被销毁

Executors.newFixedThreadPool()

• 核心数和最大线程数大小一样
• 没有所谓的非空闲时间，即keepAliveTime为0
• 阻塞队列为无界队列LinkedBlockingQueue

/**
     * Creates a thread pool that reuses a fixed number of threads
     * operating off a shared unbounded queue, using the provided
     * ThreadFactory to create new threads when needed.  At any point,
     * at most {@code nThreads} threads will be active processing
     * tasks.  If additional tasks are submitted when all threads are
     * active, they will wait in the queue until a thread is
     * available.  If any thread terminates due to a failure during
     * execution prior to shutdown, a new one will take its place if
     * needed to execute subsequent tasks.  The threads in the pool will
     * exist until it is explicitly {@link ExecutorService#shutdown
     * shutdown}.
     *
     * @param nThreads the number of threads in the pool
     * @param threadFactory the factory to use when creating new threads
     * @return the newly created thread pool
     * @throws NullPointerException if threadFactory is null
     * @throws IllegalArgumentException if {@code nThreads <= 0}
     */
    public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>(),
                                      threadFactory);
    }

它的工作步骤是：

1. 提交任务
2. 如果线程数少于核心线程，创建核心线程执行任务
3. 如果线程数等于核心线程，把任务添加到LinkedBlockingQueue阻塞队列
4. 如果线程执行完任务，去阻塞队列取任务，继续执行

使用无界队列的线程池会导致内存飙升

Executors.newScheduledThreadPool()

• 最大线程数为Integer.MAX_VALUE
• 阻塞队列是DelayedWordQueue
• keepAliveTime为0
• scheduleAtFixedRate() ：按某种速率周期执行
• scheduleWithFixedDelay()：在某个延迟后执行

它的工作步骤是：

1. 添加一个任务
2. 线程池中的线程从 DelayQueue 中取任务
3. 线程从 DelayQueue 中获取 time 大于等于当前时间的task
4. 执行完后修改这个 task 的 time 为下次被执行的时间
5. 这个 task 放回DelayQueue队列中

Executors.newSingleThreadPool()

• 核心线程数为1
• 最大线程数也为1
• 阻塞队列是LinkedBlockingQueue
• keepAliveTime为0

/**
     * Creates an Executor that uses a single worker thread operating
     * off an unbounded queue. (Note however that if this single
     * thread terminates due to a failure during execution prior to
     * shutdown, a new one will take its place if needed to execute
     * subsequent tasks.)  Tasks are guaranteed to execute
     * sequentially, and no more than one task will be active at any
     * given time. Unlike the otherwise equivalent
     * {@code newFixedThreadPool(1)} the returned executor is
     * guaranteed not to be reconfigurable to use additional threads.
     *
     * @return the newly created single-threaded Executor
     */
    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

它的工作步骤是：

1. 提交任务
2. 线程池是否有一条线程在，如果没有，新建线程执行任务
3. 如果有，将任务加载到阻塞队列
4. 从队列取任务，执行完一个继续执行下一个

线程池主要参数

跟踪这几个创建线程池的源码就会发现，它们其实都是通过ThreadPoolExecutor类实现的。我们看下它的构造函数。

/**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *        pool
     * @param keepAliveTime when the number of threads is greater than
     *        the core, this is the maximum time that excess idle threads
     *        will wait for new tasks before terminating.
     * @param unit the time unit for the {@code keepAliveTime} argument
     * @param workQueue the queue to use for holding tasks before they are
     *        executed.  This queue will hold only the {@code Runnable}
     *        tasks submitted by the {@code execute} method.
     * @param threadFactory the factory to use when the executor
     *        creates a new thread
     * @param handler the handler to use when execution is blocked
     *        because the thread bounds and queue capacities are reached
     * @throws IllegalArgumentException if one of the following holds:<br>
     *         {@code corePoolSize < 0}<br>
     *         {@code keepAliveTime < 0}<br>
     *         {@code maximumPoolSize <= 0}<br>
     *         {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException if {@code workQueue}
     *         or {@code threadFactory} or {@code handler} is null
     */
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

线程池就是上面几个核心参数：

corePoolSize

核心线程数目。默认情况下，在创建了线程之后，线程池中的线程数为0，当有任务来之后，就会创建一个线程去执行任务，当线程池中的线程数目达到corePoolSize后，就会把到达的任务放进缓存队列中

maximumPoolSize

线程池最大的线程数。表示线程最多能创建多少个线程

keepAliveTime

表示线程没有任务执行时最多保持多久时间会终止。默认情况下，只有

unit

参数keepAliveTime的时间单位，有7种取值

• TimeUnit.DAYS
• TimeUnit.HOURS
• TimeUnit.MINUTES
• TimeUnit.SECONDS
• TimeUnit.MILLISECONDS
• TimeUnit.MICROSECONDS
• TimeUnit.NANOSECONDS

workQueue

一个阻塞队列，用来存储等待执行的任务。一般来说有以下几种队列：

• ArrayBlockingQueue：ArrayBlockingQueue（有界队列）是一个用数组实现的有界阻塞队列，按FIFO排序量。
• LinkedBlockingQueue：基于链表结构的阻塞队列，按照FIFO排序任务，容量可以选择进行设置，不设置的话，将是一个无边界的阻塞队列，最大长度为Integer.MAX_VALUE，吞吐量通常要高于ArrayBlockingQueue；newFixedThreadPool线程池使用了这个队列
• DelayQueue：延迟队列，是一个任务定时周期的延迟执行的队列。根据指定的执行时间从小到大排序，否则根据插入到队列的先后排序。newScheduledThreadPool线程池使用了这个队列
• PriorityBlockingQueue：是具有优先级的无界阻塞队列
• SynchronousQueue：是一个不存储元素的阻塞队列，每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，newCachedThreadPool线程池使用了这个队列

threadFactory

线程工厂，主要用来创建线程

handler

表示当拒绝处理任务时的策略，有以下四种取值

• ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常
• ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，但是不抛出异常。
• ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务（重复此过程）
• ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

线程池状态

线程池有RUNNING、SHUTDOWN、STOP、TIDYING、TERMINATED等5个状态。

RUNNING

在这个状态下的线程池可以接收新任务，并处理阻塞队列中的任务，此时调用线程池的shutdown()方法，可以切换到SHUTDOWN状态，调用shutdownNow()方法，可以切换到stop状态

SHUTDOWN

该状态的线程池不会接收新任务，但会处理阻塞队列中的任务。当队列为空时，并且线程池中执行的任务也为空，线程池就会进入TIDYING状态

STOP

该状态的线程不会接收新任务，也不会处理阻塞队列中的任务，而且会中断正在运行的任务

TIDYING

该状态表名所有的任务已经运行终止，记录的任务数量为0

TERMINATED

该状态表示线程池彻底终止

线程池各个状态的切换关系如下图所示

线程池的执行流程

总的来说，线程池执行步骤可以用下图来概括：

1. 提交一个任务，线程池里存活的核心线程数小于线程数corePoolSize时，线程池会创建一个核心线程去处理提交的任务
2. 如果线程池核心线程数已满，即线程数已经等于corePoolSize，一个新提交的任务，会被放进任务队列workQueue排队等待执行
3. 当线程池里面存活的线程数已经等于corePoolSize了，并且任务队列workQueue也满，判断线程数是否达到maximumPoolSize，即最大线程数是否已满，如果没到达，创建一个非核心线程执行提交的任务
4. 如果当前线程数达到了maximumPoolSize，还有新的任务过来的话，直接采用拒绝策略处理。

Java中的线程池是用ThreadPoolExecutor类来实现的. 本文就结合JDK 1.8对该类的源码来分析一下这个类内部对于线程的创建, 管理以及后台任务的调度等方面的执行原理。

先来看下线程池的类图：

Executors提供了一系列工厂方法用于创建线程池，返回的线程池都实现了ExecutorService接口。而后调用ExecutorService的execute(Runnable runnable)方法。一旦 Runnable 任务传递到 execute()方法，该方法便会自动在一个线程上执行。

线程池异常处理

在使用线程池处理任务的时候，任务代码可能抛出RuntimeException，抛出异常后，线程池可能捕获它，也可能创建一个新的线程来代替异常的线程，我们可能无法感知任务出现了异常，因此我们需要考虑线程池异常情况。

有四种处理线程池异常的方法：

1. 使用try-catch捕获异常
2. submit执行，Future.get接受异常
3. 重写ThreadPoolExecutor.afterExecute方法，处理传递的异常引用
4. 实例化时，传入自己的ThreadFactory，设置Thread.UncaughtExceptionHandler处理未检测的异常

如何合理设置线程池大小

对于不同性质的任务来说，CPU密集型任务应配置尽可能小的线程，如配置CPU个数+1的线程数，IO密集型任务应配置尽可能多的线程，因为IO操作不占用CPU，不要让CPU闲下来，应加大线程数量，如配置两倍CPU个数+1，而对于混合型的任务，如果可以拆分，拆分成IO密集型和CPU密集型分别处理，前提是两者运行的时间是差不多的，如果处理时间相差很大，则没必要拆分了。

若任务对其他系统资源有依赖，如某个任务依赖数据库的连接返回的结果，这时候等待的时间越长，则CPU空闲的时间越长，那么线程数量应设置得越大，才能更好的利用CPU。

最佳线程数目 = （（线程等待时间+线程CPU时间）/线程CPU时间 ）* CPU数目

比如平均每个线程CPU运行时间为0.5s，而线程等待时间（非CPU运行时间，比如IO）为1.5s，CPU核心数为8，那么根据上面这个公式估算得到：((0.5+1.5)/0.5)*8=32。这个公式进一步转化为：

最佳线程数目 = （线程等待时间与线程CPU时间之比 + 1）* CPU数目

可以得出一个结论：

线程等待时间所占比例越高，需要越多线程。线程CPU时间所占比例越高，需要越少线程。

参考资料

深入理解 Java 线程池：ThreadPoolExecutor

面试必备：Java线程池解析