ThreadPoolExecutor 线程池调度器

发表于 | 分类于 ， | 阅读次数:
字数统计: 1,960 | 阅读时长 ≈ 7

可设置的主要参数

  • corePoolSize

核心线程数，核心线程会一直存活，即使没有任务需要处理。当线程数小于核心线程数时，即使现有的线程空闲，线程池也会优先创建新线程来处理任务，而不是直接交给现有的线程处理。

核心线程在allowCoreThreadTimeout被设置为true时会超时退出，默认情况下不会退出。

  • maxPoolSize

当线程数大于或等于核心线程，且任务队列已满时，线程池会创建新的线程，直到线程数量达到maxPoolSize。如果线程数已等于maxPoolSize，且任务队列已满，则已超出线程池的处理能力，线程池默认会拒绝处理任务而抛出异常。可以自定义处理多余任务的RejectedExecutionHandler

该参数在某些情况下是无效的。

  • keepAliveTime

当已经执行过的线程空闲后，空闲时间小于keepAliveTime，又来了新的任务时，会直接启动该线程。

当线程空闲时间达到keepAliveTime，该线程会退出，直到线程数量等于corePoolSize。

如果allowCoreThreadTimeout设置为true，则所有线程均会退出直到线程数量为0。

  • allowCoreThreadTimeout

是否允许核心线程空闲退出，默认值为false。

  • queueCapacity

任务队列容量，由它设定一个有界的队列。从maxPoolSize的描述上可以看出，任务队列的容量会影响到线程的变化，因此任务队列的长度也需要恰当的设置。

  • RejectedExecutionHandler

    1
    2
    3
    public interface RejectedExecutionHandler {
        void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
    }

    任务拒绝处理策略。在源码中定义了如下几种策略：

    • AbortPolicy (默认策略)

      1
      2
      3
      4
      5
      6
      7
      8
      public static class AbortPolicy implements RejectedExecutionHandler {
              public AbortPolicy() { }
              public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
                  throw new RejectedExecutionException("Task " + r.toString() +
                                                       " rejected from " +
                                                       e.toString());
              }
          }

      该策略 直接抛出了异常。异常中止

    • CallerRunsPolicy

      1
      2
      3
      4
      5
      6
      7
      8
      public static class CallerRunsPolicy implements RejectedExecutionHandler {
          public CallerRunsPolicy() { }
          public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
              if (!e.isShutdown()) {
                  r.run();
              }
          }
      }

      在线程池未关闭时，直接在当前线程执行任务。所谓当前线程，即提交任务时的线程

    • DiscardPolicy

      1
      2
      3
      4
      5
      public static class DiscardPolicy implements RejectedExecutionHandler {
          public DiscardPolicy() { }
          public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
          }
      }

      不做任何处理，即跳过任务r

    • DiscardOldestPolicy

      1
      2
      3
      4
      5
      6
      7
      8
      9
      public static class DiscardOldestPolicy implements RejectedExecutionHandler {
          public DiscardOldestPolicy() { }
          public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
              if (!e.isShutdown()) {
                  e.getQueue().poll();
                  e.execute(r);
              }
          }
      }

      在线程池未关闭时，移除处于队头的任务，再执行任务r。相当于把之前队头的任务跳过了，把r又加入了队列

      ​

      ​

参数对线程池任务调度的影响

当任务数大于等于corePoolSize，这时，除核心线程执行了一定数量的任务后，

余下的需要执行的任务数为int count = tasksCount - corePoolSize，设队列最大容量为capacity：

  • 若capacity >= count，则只会创建corePoolSize个线程来执行任务。多的任务放在队列中。

    这时maxPoolSize，就无效了

  • 若capacity < count，这时有int remain = tasksCount - capacity：

    • 若 remain <= maxPoolSize，创建remain个线程来执行任务。多的任务放在队列中
    • 若remain > maxPoolSize，创建maxPoolSize个线程来执行任务；这时将有remain-maxPoolSize个任务无法加入队列，被RejectedExecutionHandler处理。

    ​

Executors生成ThreadPoolExecutor

(most from http://dongxuan.iteye.com/blog/901689)

在构建时，corePoolSize、maximumPoolSizes 和BlockingQueue的选择，直接影响线程调度的策略

Executors 有三个方法可以生成 ThreadPoolExecutor

生成一个固定线程数的 ThreadPoolExecutor

public static ExecutorService newFixedThreadPool(int nThreads) {

return new ThreadPoolExecutor(nThreads, nThreads,

0L, TimeUnit.MILLISECONDS,

new LinkedBlockingQueue());

}

LinkedBlockingQueue实现是线程安全的，实现了先进先出等特性，是作为生产者消费者的首选，LinkedBlockingQueue 可以指定容量，也可以不指定，不指定的话，默认最大是Integer.MAX_VALUE，其中主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。在ThreadPoolExecutor中主要使用的是BlockingQueue的offer()和 poll()、take()

LinkedBlockingQueue 无界队列

生成一个只含一个线程的ThreadPoolExecutor

public static ExecutorService newSingleThreadExecutor() {

return new FinalizableDelegatedExecutorService

(new ThreadPoolExecutor(1, 1,

0L, TimeUnit.MILLISECONDS,

new LinkedBlockingQueue()));

}

LinkedBlockingQueue 无界队列

生成一个无界的并直接提交的ThreadPoolExecutor

public static ExecutorService newCachedThreadPool() {

return new ThreadPoolExecutor(0, Integer.MAX_VALUE,

60L, TimeUnit.SECONDS,

new SynchronousQueue());

}

这个线程池根据需要（新任务到来时）创建新的线程，如果有空闲线程则会重复使用，线程空闲了60秒后会被回收

SynchronousQueue 的特点是，每个插入操作必须等待另一个线程的对应移除操作。比如，要添加一个元素，接下来如果继续想尝试添加则会阻塞，直到另一个线程取走一个元素，反之亦然

SynchronousQueue 不会保留任务进队列，会直接提交到工作线程。

如果当前没有可工作线程，那么在 corePoolSize<maxPoolSize时，会直接创建出新的线程以执行任务

SynchronousQueue 直接提交策略

关于工作队列(workQueue)

所有 BlockingQueue 都可用于传输和保持提交的任务。可以使用此队列与池大小进行交互：

  • 如果运行的线程少于 corePoolSize，则 Executor 始终首选添加新的线程，而不进行排队。（什么意思？如果当前运行的线程小于corePoolSize，则任务根本不会存放，添加到queue中，而是直接开始运行thread。
  • 如果运行的线程等于或多于 corePoolSize，则 Executor 始终首选将请求加入队列，而不添加新的线程。
  • 如果无法将请求加入队列，则创建新的线程，除非创建此线程超出 maximumPoolSize，在这种情况下，任务将被拒绝。

排队有三种通用策略：

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

版权声明
 协议：No Fuck License

由 stone 创作并维护
本文首发于 stone世界 博客（ http://stone86.top ）
版权所有，侵权必究。如要转载，请声明出处

Fork me on GitHub