转载

java线程池的原理学习(二)

接上文:java线程池的原理学习

ThreadPoolExecutor简单介绍

ThreadPoolExecutor ,线程池类,继承自 AbstractExecutorService

public class ThreadPoolExecutor extends AbstractExecutorService 

构造方法

ThreadPoolExecutor 提供了四种构造方法实现(这里只介绍一种):

 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.corePoolSize = corePoolSize;  this.maximumPoolSize = maximumPoolSize;  this.workQueue = workQueue;  this.keepAliveTime = unit.toNanos(keepAliveTime);  this.threadFactory = threadFactory;  this.handler = handler; } 

有必要对每个参数解释一下:

  • corePoolSize - 池中所保存的线程数,包括空闲线程。

  • maximumPoolSize - 池中允许的最大线程数。

  • keepAliveTime - 当线程数大于核心时,此为终止前多余的空闲线程等待新任务的最长时间。

  • unit - keepAliveTime 参数的时间单位。

  • workQueue - 执行前用于保持任务的队列。此队列仅保持由 execute 方法提交的 Runnable 任务。

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

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

配置规则

为了便于跨大量上下文使用,此类提供了很多可调整的参数和扩展钩子 (hook)。jdk文档中建议在通常情况下,使用 Executors 提供的工厂方法配置,也就是提供好了的线程池。若非要手动配置,需要遵循以下规则:

核心和最大池大小

ThreadPoolExecutor 将根据 corePoolSizemaximumPoolSize 设置的边界自动调整池大小。当新任务在方法 execute(java.lang.Runnable) 中提交时:

1.运行的线程少于 corePoolSize ,则创建新线程来处理请求,即使其他辅助线程是空闲的。

2:运行的线程多于 corePoolSize 而少于 maximumPoolSize ,则把任务放进队列,由空闲线程从队列中取任务,仅当队列满时才创建新线程。

3:如果设置的 corePoolSizemaximumPoolSize 相同,则创建了固定大小的线程池。

4:如果将 maximumPoolSize 设置为基本的无界值(如 Integer.MAX_VALUE ),则允许池适应任意数量的并发任务。

还要注意以下两点:

  1. 在大多数情况下,核心和最大池大小仅基于构造器来设置,不过也可以使用 setCorePoolSize(int)setMaximumPoolSize(int) 进行动态更改。

  2. 当池中的线程数大于 corePoolSize 的时候,多余的线程会等待 keepAliveTime 长的时间,如果无请求可处理就自行销毁。

创建新线程

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

ThreadFactory 是线程工厂,它是一个接口:

public interface ThreadFactory {     Thread newThread(Runnable r); }

ThreadPoolExecutor 中的 threadFactory 是由 Executors 工具类提供的:

public static ThreadFactory defaultThreadFactory() {         return new DefaultThreadFactory(); }
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) {     ////创建的线程以“pool-N-thread-M”命名,N是该工厂的序号,M是线程号   Thread t = new Thread(group, r,          namePrefix + threadNumber.getAndIncrement(),          0);    //设为非后台线程          if (t.isDaemon())    t.setDaemon(false);   //优先级为normal    if (t.getPriority() != Thread.NORM_PRIORITY)    t.setPriority(Thread.NORM_PRIORITY);   return t;  } } 

DefaultThreadFactory 是一个静态内部类

排队策略

前面说到,当线程池中运行的线程等于或多于 corePoolSize ,则 Executor 始终首选将请求加入队列,而不添加新的线程,将任务加入队列有三种策略(具体参见jdk文档)。

被拒绝的任务

两种情况下,新提交的任务将会被拒绝:

  1. Executor 已经关闭

  2. Executor 将有限边界用于最大线程和工作队列容量,且已经饱和

被拒绝的任务, execute 方法都将调用其 RejectedExecutionHandlerRejectedExecutionHandler.rejectedExecution(java.lang.Runnable, java.util.concurrent.ThreadPoolExecutor) 方法。下面提供了四种预定义的处理程序策略:

在默认的 ThreadPoolExecutor.AbortPolicy 中,处理程序遭到拒绝将抛出运行时 RejectedExecutionException

ThreadPoolExecutor.CallerRunsPolicy 中,线程调用运行该任务的 execute 本身。此策略提供简单的反馈控制机制,能够减缓新任务的提交速度。

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

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

定义和使用其他种类的 RejectedExecutionHandler 类也是可能的,但这样做需要非常小心,尤其是当策略仅用于特定容量或排队策略时。

钩子方法

此类提供两个 protected 可重写的 钩子方法:

protected void beforeExecute(Thread t, Runnable r) { } protected void afterExecute(Runnable r, Throwable t) { }

这两种方法分别在执行 每个任务 之前和之后调用。它们可用于操纵执行环境;注意这里是每个任务,即每次运行新任务时都会执行一遍。例如,重新初始化 ThreadLocal 、搜集统计信息或添加日志条目。此外,还可以重写方法 terminated() 来执行 Executor 完全终止后需要完成的所有特殊处理。

如果钩子 (hook) 或回调方法抛出异常,则内部辅助线程将依次失败并突然终止。

jdk文档中提供了一个可以暂停和恢复的线程池例子:

class PausableThreadPoolExecutor extends ThreadPoolExecutor {    private boolean isPaused;    private ReentrantLock pauseLock = new ReentrantLock();    private Condition unpaused = pauseLock.newCondition();     public PausableThreadPoolExecutor(...) { super(...); }    protected void beforeExecute(Thread t, Runnable r) {      super.beforeExecute(t, r);      pauseLock.lock();      try {        while (isPaused) unpaused.await();      } catch(InterruptedException ie) {        t.interrupt();      } finally {        pauseLock.unlock();      }    }    public void pause() {      pauseLock.lock();      try {        isPaused = true;      } finally {        pauseLock.unlock();      }    }    public void resume() {      pauseLock.lock();      try {        isPaused = false;        unpaused.signalAll();      } finally {        pauseLock.unlock();      }    }  } 

ThreadPoolExecutor运行原理

正文到此结束
Loading...