一、背景

        一张网络图片需要展示在页面上，通常需要开启一个线程去执行下载任务。然而当存在几十上百张图片的时候，每张图片都去开启一个独立的线程去进行下载任务，虽然可行但这并不是一个很好的做法，因为创建一个线程本身就是一个比较大的开销，其次一直创建新的线程，内存也会猛增。
        那么怎么做比较好呢？对，就是使用线程池来优化，通过重复利用线程对象，既可以避免创建新线程带来的开销问题，也可以通过控制池内最大线程数来避免内存一直增长可能带来的内存溢出问题。

二、定义

        所谓线程池，就是定义一个“池”，里面维护一个或者多个线程对象，执行完的线程对象往往不会立即销毁，而是可以重复利用，去执行下一个任务。主要解决了多任务场景下，减少每个任务调用的开销，以及执行大量异步任务时提供增强的性能。
Java 中已经为我们封装好了线程池，其中最核心的就是 java.util.concurrent.ThreadPoolExecutor 类，所以我们先从 ThreadPoolExecutor 类来一点点熟悉和掌握 Java 线程池的使用。

三、ThreadPoolExecutor 类

1. 构造方法

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

    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory)

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

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

前三个构造方法都会调用到第四个构造方法，我们具体看一下每个参数的意思：

  • corePoolSize - 池中所保存的线程数，包括空闲线程。
  • maximumPoolSize - 池中允许的最大线程数。
  • keepAliveTime - 当线程数大于核心时，此为终止前多余的空闲线程等待新任务的最长时间。
  • unit - keepAliveTime 参数的时间单位。
  • workQueue - 执行前用于保持任务的队列。此队列仅保持由 execute 方法提交的 Runnable 任务。

  • threadFactory - 执行程序创建新线程时使用的工厂。

  • handler - 由于超出线程范围和队列容量而使执行被阻塞时所使用的处理程序。

1.1 corePoolSize 和 maximumPoolSize

        ThreadPoolExecutor 将根据 corePoolSize 和 maximumPoolSize 设置的边界自动调整池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时，如果运行的线程少于 corePoolSize，则创建新线程来处理请求，即使其他辅助线程是空闲的。线程多于 corePoolSize 而少于 maximumPoolSize 时，则仅当队列满时才创建新线程。当线程池中的线程数目达到 maximumPoolSize 后，不再创建新线程，任务会放到缓存队列当中。
        如果设置的 corePoolSize 和 maximumPoolSize 相同，则创建了固定大小的线程池。如果将 maximumPoolSize 设置为基本的无界值（如 Integer.MAX_VALUE），则允许池适应任意数量的并发任务。在大多数情况下，核心和最大池大小仅基于构造来设置，不过也可以使用 setCorePoolSize(int) 和 setMaximumPoolSize(int) 进行动态更改。
        默认情况下，线程池中并没有任何线程，而是等待有任务到来才创建线程去执行任务，除非调用了 prestartAllCoreThreads() 或者 prestartCoreThread() 方法，这两个方法是预创建的作用。

1.2 keepAliveTime

        当线程数大于核心数时，此为终止前多余的空闲线程等待新任务的最长时间。只有当线程池中的线程数大于 corePoolSize 时，keepAliveTime 才会起作用，直到线程池中的线程数不大于 corePoolSize。即一个线程空闲的时间达到 keepAliveTime，则会终止，直到线程池中的线程数不超过 corePoolSize。但是如果调用了 allowCoreThreadTimeOut(boolean) 方法，在线程池中的线程数不大于 corePoolSize 时，keepAliveTime 参数也会起作用，直到线程池中的线程数为 0。

1.3 unit

        keepAliveTime 参数的时间单位。其中 TimeUtil 类是一个枚举类，有以下取值：

TimeUnit.DAYS;              //天
TimeUnit.HOURS;             //小时
TimeUnit.MINUTES;           //分钟
TimeUnit.SECONDS;           //秒
TimeUnit.MILLISECONDS;      //毫秒
TimeUnit.MICROSECONDS;      //微妙
TimeUnit.NANOSECONDS;       //纳秒
1.4 workQueue

    存储等待执行的任务的堵塞队列，使用该队列和池大小进行交互。

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

    这个队列一般有如下选择：

  • SynchronousQueue
    直接提交。工作队列的默认选项是 SynchronousQueue，它将任务直接提交给线程而不保持它们。在此，如果不存在可用于立即运行任务的线程，则试图把任务加入队列将失败，因此会构造一个新的线程。此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
  • LinkedBlockingQueue
    无界队列。使用无界队列（例如，不具有预定义容量的 LinkedBlockingQueue）将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程就不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如，在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。
  • ArrayBlockingQueue
    有界队列。当使用有限的 maximumPoolSizes 时，有界队列（如 ArrayBlockingQueue）有助于防止资源耗尽，但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷：使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销，但是可能导致人工降低吞吐量。如果任务频繁阻塞（例如，如果它们是 I/O 边界），则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小，CPU 使用率较高，但是可能遇到不可接受的调度开销，这样也会降低吞吐量。
1.5 threadFactory

        创建新线程时使用的工厂。如果没有另外说明，则在同一个 ThreadGroup 中一律使用 Executors.defaultThreadFactory() 创建线程，并且这些线程具有相同的 NORM_PRIORITY 优先级和非守护进程状态。通过提供不同的 ThreadFactory，可以改变线程的名称、线程组、优先级、守护进程状态，等等。如果从 newThread 返回 null 时 ThreadFactory 未能创建线程，则执行程序将继续运行，但不能执行任何任务。

1.6 handler

    处理当超出线程范围和队列容量而使执行被阻塞时所使用的策略。

  • 在默认的 ThreadPoolExecutor.AbortPolicy 中，处理程序遭到拒绝将抛出运行时 RejectedExecutionException。
  • 在 ThreadPoolExecutor.CallerRunsPolicy 中，线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制，能够减缓新任务的提交速度。

  • 在 ThreadPoolExecutor.DiscardPolicy 中，不能执行的任务将被删除。

  • 在 ThreadPoolExecutor.DiscardOldestPolicy 中，如果执行程序尚未关闭，则位于工作队列头部的任务将被删除，然后重试执行程序（如果再次失败，则重复此过程）。

2. 执行方法

public void execute(Runnable command)
        在将来某个时间执行给定任务。可以在新线程中或者在现有池线程中执行该任务。 如果无法将任务提交执行，或者因为此执行程序已关闭，或者因为已达到其容量，则该任务由当前 RejectedExecutionHandler 处理。

  • 参数：
    command - 要执行的任务。
  • 抛出：
    RejectedExecutionException - 如果无法接收要执行的任务，则由 RejectedExecutionHandler 决定是否抛出
    NullPointerException - 如果命令为 null

3. 移除任务

public boolean remove(Runnable task)
        此方法可用作取消方案的一部分。它可能无法移除在放置到内部队列之前已经转换为其他形式的任务。例如，使用 submit 输入的任务可能被转换为维护 Future 状态的形式。但是，在此情况下，purge() 方法可用于移除那些已被取消的 Future。

  • 参数：
    task - 要移除的任务
  • 返回：
    如果已经移除任务，则返回 true

public void purge()
        尝试从工作队列移除所有已取消的 Future 任务。此方法可用作存储回收操作，它对功能没有任何影响。取消的任务不会再次执行，但是它们可能在工作队列中累积，直到 worker 线程主动将其移除。调用此方法将试图立即移除它们。但是，如果出现其他线程的干预，那么此方法移除任务将失败。

扩展：

        Future 是一个接口。它提供了检查计算是否完成的方法，以等待计算的完成，并获取计算的结果。计算完成后只能使用 get 方法来获取结果，如有必要，计算完成前可以阻塞此方法。取消则由 cancel 方法来执行。还提供了其他方法，以确定任务是正常完成还是被取消了。一旦计算完成，就不能再取消计算。如果为了可取消性而使用 Future 但又不提供可用的结果，则可以声明 Future<?> 形式类型、并返回 null 作为底层任务的结果。

        FutureTask 类是 Future 的一个实现，同时也实现了 Runnable，所以可通过 Executor 来执行。

4. 关闭任务

public void shutdown()
        按过去执行已提交任务的顺序发起一个有序的关闭，但是不接受新任务。如果已经关闭，则调用没有其他作用。

  • 抛出：
    SecurityException - 如果安全管理器存在并且关闭此 ExecutorService 可能操作某些不允许调用者修改的线程（因为它没有 RuntimePermission ("modifyThread")），或者安全管理器的 checkAccess 方法拒绝访问。

public List shutdownNow()
        尝试停止所有的活动执行任务、暂停等待任务的处理，并返回等待执行的任务列表。在从此方法返回的任务队列中排空（移除）这些任务。
        并不保证能够停止正在处理的活动执行任务，但是会尽力尝试。 此实现通过 Thread.interrupt() 取消任务，所以无法响应中断的任何任务可能永远无法终止。

  • 返回：
    从未开始执行的任务的列表。
  • 抛出：
    SecurityException - 如果安全管理器存在并且关闭此 ExecutorService 可能操作某些不允许调用者修改的线程（因为它没有 RuntimePermission ("modifyThread")），或者安全管理器的 checkAccess 方法拒绝访问。

5. 更多方法介绍

        其余一些API，可以参见中文API文档，地址：

http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ThreadPoolExecutor.html

四、使用示例

1. 代码

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
 
public class Main {
    public static void main(String[] args) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 5, 100, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<Runnable>(5));
        for (int i = 0; i < 10; i++) {
            executor.execute(new MyTask(i));
            System.out.println("线程池中线程数："+executor.getPoolSize()+"，队列中等待执行的任务数："+
                    executor.getQueue().size()+"，已执行完的任务数："+executor.getCompletedTaskCount());
        }
    }
}
 
class MyTask implements Runnable {
 
    private int id;
 
    public MyTask(int id) {
        this.id = id;
    }
 
    public void run() {
        System.out.println("开始执行：任务 " + id);
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("执行完毕：任务 " + id);
    }
 
}

2. 运行结果

开始执行：任务 0
线程池中线程数：1，队列中等待执行的任务数：0，已执行完的任务数：0
线程池中线程数：2，队列中等待执行的任务数：0，已执行完的任务数：0
开始执行：任务 1
线程池中线程数：3，队列中等待执行的任务数：0，已执行完的任务数：0
开始执行：任务 2
线程池中线程数：3，队列中等待执行的任务数：1，已执行完的任务数：0
线程池中线程数：3，队列中等待执行的任务数：2，已执行完的任务数：0
线程池中线程数：3，队列中等待执行的任务数：3，已执行完的任务数：0
线程池中线程数：3，队列中等待执行的任务数：4，已执行完的任务数：0
线程池中线程数：3，队列中等待执行的任务数：5，已执行完的任务数：0
线程池中线程数：4，队列中等待执行的任务数：5，已执行完的任务数：0
开始执行：任务 8
线程池中线程数：5，队列中等待执行的任务数：5，已执行完的任务数：0
开始执行：任务 9
执行完毕：任务 2
执行完毕：任务 0
执行完毕：任务 1
开始执行：任务 5
执行完毕：任务 9
开始执行：任务 4
开始执行：任务 3
执行完毕：任务 8
开始执行：任务 6
开始执行：任务 7
执行完毕：任务 3
执行完毕：任务 6
执行完毕：任务 5
执行完毕：任务 4
执行完毕：任务 7

3. 分析

  • 可以看到线程池中的线程数最大为5，不会超过 maximumPoolSize
  • 任务数量大于最大线程数时，多余的任务会在阻塞队列的保存

  • 如果堵塞队列也满了，再来任务的话就会使用默认的 AbortPolicy 处理，它会抛出 RejectedExecutionException 异常

  • 具体的池容量大小选择、任务缓存及排队策略、任务拒绝策略需要根据应用场景来挑选

五、扩展

        官方其实并不推荐我们直接使用 ThreadPoolExecotor 类，而是使用 Executors 类提供的几个静态方法来创建线程池，常用的有：

1. public static ExecutorService newCachedThreadPool()

    public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }

        创建一个可根据需要创建新线程的线程池，但是在以前构造的线程可用时将重用它们。对于执行很多短期异步任务的程序而言，这些线程池通常可提高程序性能。调用 execute 将重用以前构造的线程（如果线程可用）。如果现有线程没有可用的，则创建一个新线程并添加到池中。终止并从缓存中移除那些已有 60 秒钟未被使用的线程。因此，长时间保持空闲的线程池不会使用任何资源。注意，可以使用 ThreadPoolExecutor 构造方法创建具有类似属性但细节不同（例如超时参数）的线程池。

2. public static ExecutorService newSingleThreadExecutor()

    public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

        创建一个使用单个 worker 线程的 Executor，以无界队列方式来运行该线程。（注意，如果因为在关闭前的执行期间出现失败而终止了此单个线程，那么如果需要，一个新线程将代替它执行后续的任务）。可保证顺序地执行各个任务，并且在任意给定的时间不会有多个线程是活动的。与其他等效的 newFixedThreadPool(1) 不同，可保证无需重新配置此方法所返回的执行程序即可使用其他的线程。

3. public static ExecutorService newFixedThreadPool(int nThreads)

    public static ExecutorService newFixedThreadPool(int nThreads) {
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }

        创建一个可重用固定线程数的线程池，以共享的无界队列方式来运行这些线程。在任意点，在大多数 nThreads 线程会处于处理任务的活动状态。如果在所有线程处于活动状态时提交附加任务，则在有可用线程之前，附加任务将在队列中等待。如果在关闭前的执行期间由于失败而导致任何线程终止，那么一个新线程将代替它执行后续的任务（如果需要）。在某个线程被显式地 关闭之前，池中的线程将一直存在。

六、线程池大小选择

    一般需要根据任务的类型来配置线程池大小：

  • 如果是CPU密集型任务，就需要尽量压榨CPU，参考值可以设为 N(CPU)+1
  • 如果是IO密集型任务，参考值可以设置为 2*N(CPU)

        这只是一个参考值，具体的设置还需要根据实际情况进行调整，比如可以先将线程池大小设置为参考值，再观察任务运行情况和系统负载、资源利用率来动态调整。

​
Logo
开放原子开发者工作坊

开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！

更多推荐

操作系统大会&openEuler Summit 2024参会指南，请查收！

avatar 开放原子开发者工作坊

推动工业软件核心技术攻关，开源工业软件算法集成大赛正式启动！

推动工业软件核心技术攻关，开源工业软件算法集成大赛正式启动！

avatar 开放原子开发者工作坊

第二届openEuler生态大会（中国·湖南）成功举办

10月30日，第二届openEuler生态大会（中国·湖南）成功举办。

avatar 开放原子开发者工作坊