BlogGitHub

《Java 并发编程实战》笔记 - 线程池的使用

线程安全性

要编写线程安全的代码，其核心在于要 对状态访问操作进行管理，特别是对共享（Shared）和可变（Mutable）状态的访问。「共享」意味着变量可以由许多个线程同时访问，「可变」意味着变量的值在其生命周期内可以发生变化。

从非正式的意义上来说，对象的状态是指存储在状态变量（例如实例或者静态域）中的数据。对象的状态可能包括其他依赖对象的域，例如某个 HashMap 的状态不仅存储在 HashMap 对象本身，还存储在许多 Map.Entry 对象中。

实现线程安全的方式有以下三种：

  1. 不在线程之间共享状态变量；

  2. 将状态变量设置为不可变的；

  3. 在访问状态变量时，使用同步机制。

Java 中主要的同步机制是 synchronized 关键字，它是一种独占的加锁方式，但是「同步」还可以包括 volatile 类型的变量、Lock 显示锁、AtomicLong 等原子变量。

什么是线程安全性

简单版：当多个线程访问某个类时，这个类始终都能表现出正确的行为，那么就称这个类是线程安全的。

详细版：当多个线程访问某个类时，不管运行时环境采用何种调度方式或者这些线程将如何交替执行，并且在主调代码中不需要任何额外的同步或协同，这个类都能表现出正确的行为，那么就称这个类是线程安全的。

线程池的使用：配置 ThreadPoolExecutor

ThreadPoolExecutor 为 Executor 提供了基本的实现，这些 Executor 是由 Executors 中的 newCachedThreadPool、newFixedThreadPool、newScheduledThreadExecutor 等工厂方法返回的。ThradPoolExecutor 是一个灵活的、稳定的线程池，允许进行各种定制。

如果默认的执行策略不能满足需求，那么可以通过 ThreadPoolExecutor 的构造参数，根据自己的需求定制一个对象。ThreadPoolExecutor 的所有构造参数如下：

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

线程的创建与销毁

线程池的基本大小 corePoolSize、最大大小 maximumPoolSize、以及存活时间 keepAliveTime + unit 等因素共同负责线程的创建与销毁：

  1. corePoolSize 是线程池的目标大小，即在没有任务执行时的线程池大小，并且只有在工作队列 workQueue 满了的情况下，才会创建超出这个数量的线程；

  2. maximumPoolSize 表示可同时活动的线程数量的上限；

  3. 如果某个线程的空闲时间超过了存活时间 keepAliveTime + unit，那么将被标记为可回收，并且在当线程池的当前大小超过了 corePoolSize 时，这个线程将被终止。

newFixedThreadPool 工厂方法将线程池的 corePoolSize 和 maximumPoolSize 设置为参数中指定的值，并且创建的线程池不会超时。

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

newCachedThreadPool 工厂方法将线程池的 maximumPoolSize 设置为 Integer.MAX_VALUE，corePoolSize 设置为 0，并且超时时间为一分钟。

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

管理任务队列

如果新请求的到达速率超过了线程池的处理速率，那么新到来的请求将累积起来。在线程池中，这些请求会在一个 Runnable 队列 workQueue 中等待，而不会像线程那样去竞争 CPU 资源。ThreadPoolExecutor 允许提供一个 BlockingQueue 来保存等待执行的任务，基本的任务排队方式有三种：无界队列、有界队列、同步移交。队列的选择与其它线程池配置参数有关。

newFixedThreadPool 和 newSingleThreadExecutor 工厂方法将使用一个无界的 LinkedBlockingQueue。如果所有工作者线程都处于忙碌状态，并且任务持续快速地到达，那么 workQueue 将无限制地增加，这可能会导致程序内存溢出。

一种更加稳妥的方式是使用有界队列，例如 ArrayBlockingQueue、有界的 LinkedBlockingQueue、PriorityBlockingQueue。使用有界队列有助于避免资源耗尽的情况发生，在有界队列填满之后，线程池中的饱和策略 RejectedExecutionHandler 将会开始发挥作用。使用有界队列时，队列的大小与线程池的大小必须一起调节，如果线程池较小，队列较大，那么将有助于降低 CPU 的使用率，同时还可以减少上下文切换，但是可能会限制程序的吞吐量。

对于非常大或者无界的线程池，可以使用 SynchronousQueue 来避免任务排队，以及直接将任务从生产者移交给工作者线程。

当使用像 LinkedBlockingQueue 或者 ArrayBlockingQueue 这样的 FIFO 队列时，任务的执行顺序与它们的到达顺序相同。如果想进一步控制任务的执行顺序，还可以使用 PriorityBlockingQueue。

饱和策略

当有界队列被填满后，饱和策略开始发挥作用。JDK 提供了几种不同的 RejectedExecutionHandler 实现，每种实现都包含不同的饱和策略：

  1. CallerRunsPolicy 将当前任务回退给调用线程，并且会在调用线程执中行任务；

  2. AbortPolicy（默认） 抛出 RejectedExecutionException 异常；

  3. DiscardPolicy 抛弃当前任务

  4. DiscardOldestPolicy 抛弃 workQueue 队首的任务，然后重新提交当前任务。

线程工厂

当线程池需要创建一个线程时，都是通过线程工厂 ThreadFactory 来完成的。默认的线程工厂方法将创建一个新的、非守护的线程，并且不包含特殊的配置。

private static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

在许多情况下，都需要使用定制的线程工厂，例如希望为线程池中的线程指定一个 UncaughtExceptionHandler，或者实例化一个定制的 Thread 类用于执行调试信息的记录，或者只是希望给线程取一个更有意义的名称，用来解析线程的转储信息和错误信息。

一图概览

         +------------------------------------------------------------------------+
runnbale |                                              +-----------------------+ |
-------->|                                 True         | +------+ +------+     | |
         |--> workerCount < corePoolSize? ------------->| |Worker| |Worker| ... | |
runnbale |                 |               new Worker() | +------+ +------+     | |
-------->|           False |                            +-----------------------+ |
         |                 |                               |   |   |         ^    |
runnbale |                 |                            workQueue.take       |    |
-------->|                 |                               |   |   |         |    |
         |                 |                               v   v   v         |    |
         |                 |                 +---------------------------+   |    |
         |                 v         True    | +--------+ +--------+     |   |    |
         |         workQueue.offer? -------->| |Runnable| |Runnable| ... |   |    |
         |                 |                 | +--------+ +--------+     |   |    |
         |           False |                 +---------------------------+   |    |
         |                 |                      BlockingQueue<Runnable>    |    |
         |                 |                                                 |    |
         |                 V                    True                         |    |
         |      workerCount < maximumPoolSize? ------------------------------+    |
         |                 |                    new Worker()                      |
         |           False |                                                      |
         |                 |                                                      |
         |                 v                                                      |
         |        +-------------+------------+--------------+                     |
         |        |             |            |              |                     |
         |        v             v            v              v                     |
         |  CallerRunsPolicy  AbortPolicy  DiscardPolicy  DiscardOldestPolicy     |
         +------------------------------------------------------------------------+

最后更新于