【Java并发编制程序】7、线程池,java并发线程池

奥门威尼斯网址 1image.png

Java并发 线程池

关注点

ThreadPoolExecutor类,创建线程池的核心类。Executors静态工厂方法,通过这个类来创建线程池。

ExecutorService接口 An Executor that provides methods to manage
termination and methods that can produce a Future for tracking
progress of one or more asynchronous tasks.
Executor接口 An object that executes submitted Runnable tasks.
Executors类 为 Executor, ExecutorService, ScheduledExecutorService,
ThreadFactory, and Callable提供工厂和工具方法,
阻塞队列

【Java并发编程】7、线程池,java并发线程池

Java并发编程:线程池的使用,java并发线程池

在前面的文章中,我们使用线程的时候就去创建一个线程,这样实现起来非常简便,但是就会有一个问题:

如果并发的线程数量很多,并且每个线程都是执行一个时间很短的任务就结束了,这样频繁创建线程就会大大降低系统的效率,因为频繁创建线程和销毁线程需要时间。

那么有没有一种办法使得线程可以复用,就是执行完一个任务,并不被销毁,而是可以继续执行其他的任务?

在Java中可以通过线程池来达到这样的效果。今天我们就来详细讲解一下Java的线程池,首先我们从最核心的ThreadPoolExecutor类中的方法讲起,然后再讲述它的实现原理,接着给出了它的使用示例,最后讨论了一下如何合理配置线程池的大小。

以下是本文的目录大纲:

一.Java中的ThreadPoolExecutor类

二.深入剖析线程池实现原理

三.使用示例

四.如何合理配置线程池的大小

若有不正之处请多多谅解,并欢迎批评指正。

一.Java中的ThreadPoolExecutor类

java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类,因此如果要透彻地了解Java中的线程池,必须先了解这个类。下面我们来看一下ThreadPoolExecutor类的具体实现源码。

在ThreadPoolExecutor类中提供了四个构造方法:

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 public class ThreadPoolExecutor extends AbstractExecutorService {  .....  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);  ... }

从上面的代码可以得知,ThreadPoolExecutor继承了AbstractExecutorService类,并提供了四个构造器,事实上,通过观察每个构造器的源码具体实现,发现前面三个构造器都是调用的第四个构造器进行的初始化工作。

下面解释下一下构造器中各个参数的含义:

  • corePoolSize:核心池的大小,这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后,默认情况下,线程池中并没有任何线程,而是等待有任务到来才创建线程去执行任务,除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法,从这2个方法的名字就可以看出,是预创建线程的意思,即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下,在创建了线程池后,线程池中的线程数为0,当有任务来之后,就会创建一个线程去执行任务,当线程池中的线程数目达到corePoolSize后,就会把到达的任务放到缓存队列当中;
  • maximumPoolSize:线程池最大线程数,这个参数也是一个非常重要的参数,它表示在线程池中最多能创建多少个线程;
  • keepAliveTime:表示线程没有任务执行时最多保持多久时间会终止。默认情况下,只有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起作用,直到线程池中的线程数不大于corePoolSize,即当线程池中的线程数大于corePoolSize时,如果一个线程空闲的时间达到keepAliveTime,则会终止,直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不大于corePoolSize时,keepAliveTime参数也会起作用,直到线程池中的线程数为0;
  • unit:参数keepAliveTime的时间单位,有7种取值,在TimeUnit类中有7种静态属性:
1 2 3 4 5 6 7 TimeUnit.DAYS; //天 TimeUnit.HOURS; //小时 TimeUnit.MINUTES; //分钟 TimeUnit.SECONDS; //秒 TimeUnit.MILLISECONDS; //毫秒 TimeUnit.MICROSECONDS; //微妙 TimeUnit.NANOSECONDS; //纳秒
  • workQueue:一个阻塞队列,用来存储等待执行的任务,这个参数的选择也很重要,会对线程池的运行过程产生重大影响,一般来说,这里的阻塞队列有以下几种选择:
1 2 3 ArrayBlockingQueue; LinkedBlockingQueue; SynchronousQueue;

ArrayBlockingQueue和PriorityBlockingQueue使用较少,一般使用LinkedBlockingQueue和Synchronous。线程池的排队策略与BlockingQueue有关。

  • threadFactory:线程工厂,主要用来创建线程;
  • handler:表示当拒绝处理任务时的策略,有以下四种取值:
1 2 3 4 ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。 ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程) ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务

具体参数的配置与线程池的关系将在下一节讲述。

从上面给出的ThreadPoolExecutor类的代码可以知道,ThreadPoolExecutor继承了AbstractExecutorService,我们来看一下AbstractExecutorService的实现:

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 public abstract class AbstractExecutorService implements ExecutorService {    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };  protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };  public Future<?> submit(Runnable task) {};  public <T> Future<T> submit(Runnable task, T result) { };  public <T> Future<T> submit(Callable<T> task) { };  private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,  boolean timed, long nanos)  throws InterruptedException, ExecutionException, TimeoutException {  };  public <T> T invokeAny(Collection<? extends Callable<T>> tasks)  throws InterruptedException, ExecutionException {  };  public <T> T invokeAny(Collection<? extends Callable<T>> tasks,  long timeout, TimeUnit unit)  throws InterruptedException, ExecutionException, TimeoutException {  };  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)  throws InterruptedException {  };  public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,  long timeout, TimeUnit unit)  throws InterruptedException {  }; }

AbstractExecutorService是一个抽象类,它实现了ExecutorService接口。

我们接着看ExecutorService接口的实现:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 public interface ExecutorService extends Executor {    void shutdown();  boolean isShutdown();  boolean isTerminated();  boolean awaitTermination(long timeout, TimeUnit unit)  throws InterruptedException;  <T> Future<T> submit(Callable<T> task);  <T> Future<T> submit(Runnable task, T result);  Future<?> submit(Runnable task);  <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)  throws InterruptedException;  <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,  long timeout, TimeUnit unit)  throws InterruptedException;    <T> T invokeAny(Collection<? extends Callable<T>> tasks)  throws InterruptedException, ExecutionException;  <T> T invokeAny(Collection<? extends Callable<T>> tasks,  long timeout, TimeUnit unit)  throws InterruptedException, ExecutionException, TimeoutException; }

而ExecutorService又是继承了Executor接口,我们看一下Executor接口的实现:

1 2 3 public interface Executor {  void execute(Runnable command); }

到这里,大家应该明白了ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor几个之间的关系了。

Executor是一个顶层接口,在它里面只声明了一个方法execute(Runnable),返回值为void,参数为Runnable类型,从字面意思可以理解,就是用来执行传进去的任务的;

然后ExecutorService接口继承了Executor接口,并声明了一些方法:submit、invokeAll、invokeAny以及shutDown等;

抽象类AbstractExecutorService实现了ExecutorService接口,基本实现了ExecutorService中声明的所有方法;

然后ThreadPoolExecutor继承了类AbstractExecutorService。

在ThreadPoolExecutor类中有几个非常重要的方法:

1 2 3 4 execute() submit() shutdown() shutdownNow()

execute()方法实际上是Executor中声明的方法,在ThreadPoolExecutor进行了具体的实现,这个方法是ThreadPoolExecutor的核心方法,通过这个方法可以向线程池提交一个任务,交由线程池去执行。

submit()方法是在ExecutorService中声明的方法,在AbstractExecutorService就已经有了具体的实现,在ThreadPoolExecutor中并没有对其进行重写,这个方法也是用来向线程池提交任务的,但是它和execute()方法不同,它能够返回任务执行的结果,去看submit()方法的实现,会发现它实际上还是调用的execute()方法,只不过它利用了Future来获取任务执行结果(Future相关内容将在下一篇讲述)。

shutdown()和shutdownNow()是用来关闭线程池的。

还有很多其他的方法:

比如:getQueue() 、getPoolSize()
、getActiveCount()、getCompletedTaskCount()等获取与线程池相关属性的方法,有兴趣的朋友可以自行查阅API。

二.深入剖析线程池实现原理

在上一节我们从宏观上介绍了ThreadPoolExecutor,下面我们来深入解析一下线程池的具体实现原理,将从下面几个方面讲解:

  1.线程池状态

  2.任务的执行

  3.线程池中的线程初始化

  4.任务缓存队列及排队策略

  5.任务拒绝策略

  6.线程池的关闭

  7.线程池容量的动态调整

 

1.线程池状态

在ThreadPoolExecutor中定义了一个volatile变量,另外定义了几个static
final变量表示线程池的各个状态:

 

1 2 3 4 5 volatile int runState; static final int RUNNING = 0; static final int SHUTDOWN = 1; static final int STOP = 2; static final int TERMINATED = 3;

runState表示当前线程池的状态,它是一个volatile变量用来保证线程之间的可见性;

下面的几个static final变量表示runState可能的几个取值。

当创建线程池后,初始时,线程池处于RUNNING状态;

如果调用了shutdown()方法,则线程池处于SHUTDOWN状态,此时线程池不能够接受新的任务,它会等待所有任务执行完毕;

如果调用了shutdownNow()方法,则线程池处于STOP状态,此时线程池不能接受新的任务,并且会去尝试终止正在执行的任务;

当线程池处于SHUTDOWN或STOP状态,并且所有工作线程已经销毁,任务缓存队列已经清空或执行结束后,线程池被设置为TERMINATED状态。

奥门威尼斯网址,2.任务的执行

在了解将任务提交给线程池到任务执行完毕整个过程之前,我们先来看一下ThreadPoolExecutor类中其他的一些比较重要成员变量:

 

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 private final BlockingQueue<Runnable> workQueue; //任务缓存队列,用来存放等待执行的任务 private final ReentrantLock mainLock = new ReentrantLock(); //线程池的主要状态锁,对线程池状态(比如线程池大小  //、runState等)的改变都要使用这个锁 private final HashSet<Worker> workers = new HashSet<Worker>(); //用来存放工作集   private volatile long keepAliveTime; //线程存货时间 private volatile boolean allowCoreThreadTimeOut; //是否允许为核心线程设置存活时间 private volatile int corePoolSize; //核心池的大小(即线程池中的线程数目大于这个参数时,提交的任务会被放进任务缓存队列) private volatile int maximumPoolSize; //线程池最大能容忍的线程数   private volatile int poolSize; //线程池中当前的线程数   private volatile RejectedExecutionHandler handler; //任务拒绝策略   private volatile ThreadFactory threadFactory; //线程工厂,用来创建线程   private int largestPoolSize; //用来记录线程池中曾经出现过的最大线程数   private long completedTaskCount; //用来记录已经执行完毕的任务个数

每个变量的作用都已经标明出来了,这里要重点解释一下corePoolSize、maximumPoolSize、largestPoolSize三个变量。

corePoolSize在很多地方被翻译成核心池大小,其实我的理解这个就是线程池的大小。举个简单的例子:

假如有一个工厂,工厂里面有10个工人,每个工人同时只能做一件任务。

因此只要当10个工人中有工人是空闲的,来了任务就分配给空闲的工人做;

当10个工人都有任务在做时,如果还来了任务,就把任务进行排队等待;

如果说新任务数目增长的速度远远大于工人做任务的速度,那么此时工厂主管可能会想补救措施,比如重新招4个临时工人进来;

然后就将任务也分配给这4个临时工人做;

如果说着14个工人做任务的速度还是不够,此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。

当这14个工人当中有人空闲时,而新任务增长的速度又比较缓慢,工厂主管可能就考虑辞掉4个临时工了,只保持原来的10个工人,毕竟请额外的工人是要花钱的。

 

这个例子中的corePoolSize就是10,而maximumPoolSize就是14(10+4)。

也就是说corePoolSize就是线程池大小,maximumPoolSize在我看来是线程池的一种补救措施,即任务量突然过大时的一种补救措施。

不过为了方便理解,在本文后面还是将corePoolSize翻译成核心池大小。

largestPoolSize只是一个用来起记录作用的变量,用来记录线程池中曾经有过的最大线程数目,跟线程池的容量没有任何关系。

 

下面我们进入正题,看一下任务从提交到最终执行完毕经历了哪些过程。

在ThreadPoolExecutor类中,最核心的任务提交方法是execute()方法,虽然通过submit也可以提交任务,但是实际上submit方法里面最终调用的还是execute()方法,所以我们只需要研究execute()方法的实现原理即可:

1 2 3 4 5 6 7 8 9 10 11 12 public void execute(Runnable command) {  if (command == null)  throw new NullPointerException();  if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {  if (runState == RUNNING && workQueue.offer(command)) {  if (runState != RUNNING || poolSize == 0)  ensureQueuedTaskHandled(command);  }  else if (!addIfUnderMaximumPoolSize(command))  reject(command); // is shutdown or saturated  } }

上面的代码可能看起来不是那么容易理解,下面我们一句一句解释:

首先,判断提交的任务command是否为null,若是null,则抛出空指针异常;

接着是这句,这句要好好理解一下:

1 if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))

由于是或条件运算符,所以先计算前半部分的值,如果线程池中当前线程数不小于核心池大小,那么就会直接进入下面的if语句块了。

如果线程池中当前线程数小于核心池大小,则接着执行后半部分,也就是执行

1 addIfUnderCorePoolSize(command)

如果执行完addIfUnderCorePoolSize这个方法返回false,则继续执行下面的if语句块,否则整个方法就直接执行完毕了。

如果执行完addIfUnderCorePoolSize这个方法返回false,然后接着判断:

1 if (runState == RUNNING && workQueue.offer(command))

如果当前线程池处于RUNNING状态,则将任务放入任务缓存队列;如果当前线程池不处于RUNNING状态或者任务放入缓存队列失败,则执行:

1 addIfUnderMaximumPoolSize(command)

如果执行addIfUnderMaximumPoolSize方法失败,则执行reject()方法进行任务拒绝处理。

回到前面:

1 if (runState == RUNNING && workQueue.offer(command))

这句的执行,如果说当前线程池处于RUNNING状态且将任务放入任务缓存队列成功,则继续进行判断:

1 if (runState != RUNNING || poolSize == 0)

这句判断是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用shutdown或者shutdownNow方法关闭了线程池的一种应急措施。如果是这样就执行:

1 ensureQueuedTaskHandled(command)

进行应急处理,从名字可以看出是保证 添加到任务缓存队列中的任务得到处理。

我们接着看2个关键方法的实现:addIfUnderCorePoolSize和addIfUnderMaximumPoolSize:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 private boolean addIfUnderCorePoolSize(Runnable firstTask) {  Thread t = null;  final ReentrantLock mainLock = this.mainLock;  mainLock.lock();  try {  if (poolSize < corePoolSize && runState == RUNNING)  t = addThread(firstTask); //创建线程去执行firstTask任务  } finally {  mainLock.unlock();  }  if (t == null)  return false;  t.start();  return true; }

这个是addIfUnderCorePoolSize方法的具体实现,从名字可以看出它的意图就是当低于核心吃大小时执行的方法。下面看其具体实现,首先获取到锁,因为这地方涉及到线程池状态的变化,先通过if语句判断当前线程池中的线程数目是否小于核心池大小,有朋友也许会有疑问:前面在execute()方法中不是已经判断过了吗,只有线程池当前线程数目小于核心池大小才会执行addIfUnderCorePoolSize方法的,为何这地方还要继续判断?原因很简单,前面的判断过程中并没有加锁,因此可能在execute方法判断的时候poolSize小于corePoolSize,而判断完之后,在其他线程中又向线程池提交了任务,就可能导致poolSize不小于corePoolSize了,所以需要在这个地方继续判断。然后接着判断线程池的状态是否为RUNNING,原因也很简单,因为有可能在其他线程中调用了shutdown或者shutdownNow方法。然后就是执行

1 t = addThread(firstTask);

这个方法也非常关键,传进去的参数为提交的任务,返回值为Thread类型。然后接着在下面判断t是否为空,为空则表明创建线程失败(即poolSize>=corePoolSize或者runState不等于RUNNING),否则调用t.start()方法启动线程。

我们来看一下addThread方法的实现:

1 2 3 4 5 6 7 8 9 10 11 12 private Thread addThread(Runnable firstTask) {  Worker w = new Worker(firstTask);  Thread t = threadFactory.newThread(w); //创建一个线程,执行任务  if (t != null) {  w.thread = t; //将创建的线程的引用赋值为w的成员变量  workers.add(w);  int nt = ++poolSize; //当前线程数加1  if (nt > largestPoolSize)  largestPoolSize = nt;  }  return t; }

在addThread方法中,首先用提交的任务创建了一个Worker对象,然后调用线程工厂threadFactory创建了一个新的线程t,然后将线程t的引用赋值给了Worker对象的成员变量thread,接着通过workers.add(w)将Worker对象添加到工作集当中。

下面我们看一下Worker类的实现:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 private final class Worker implements Runnable {  private final ReentrantLock runLock = new ReentrantLock();  private Runnable firstTask;  volatile long completedTasks;  Thread thread;  Worker(Runnable firstTask) {  this.firstTask = firstTask;  }  boolean isActive() {  return runLock.isLocked();  }  void interruptIfIdle() {  final ReentrantLock runLock = this.runLock;  if (runLock.tryLock()) {  try {  if (thread != Thread.currentThread())  thread.interrupt();  } finally {  runLock.unlock();  }  }  }  void interruptNow() {  thread.interrupt();  }    private void runTask(Runnable task) {  final ReentrantLock runLock = this.runLock;  runLock.lock();  try {  if (runState < STOP &&  Thread.interrupted() &&  runState >= STOP)  boolean ran = false;  beforeExecute(thread, task); //beforeExecute方法是ThreadPoolExecutor类的一个方法,没有具体实现,用户可以根据  //自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息,比如某个任务的执行时间等  try {  task.run();  ran = true;  afterExecute(task, null);  ++completedTasks;  } catch (RuntimeException ex) {  if (!ran)  afterExecute(task, ex);  throw ex;  }  } finally {  runLock.unlock();  }  }    public void run() {  try {  Runnable task = firstTask;  firstTask = null;  while (task != null || (task = getTask()) != null) {  runTask(task);  task = null;  }  } finally {  workerDone(this); //当任务队列中没有任务时,进行清理工作  }  } }

它实际上实现了Runnable接口,因此上面的Thread t =
threadFactory.newThread(w);效果跟下面这句的效果基本一样:

1 Thread t = new Thread(w);

相当于传进去了一个Runnable任务,在线程t中执行这个Runnable。

既然Worker实现了Runnable接口,那么自然最核心的方法便是run()方法了:

1 2 3 4 5 6 7 8 9 10 11 12 public void run() {  try {  Runnable task = firstTask;  firstTask = null;  while (task != null || (task = getTask()) != null) {  runTask(task);  task = null;  }  } finally {  workerDone(this);  } }

从run方法的实现可以看出,它首先执行的是通过构造器传进来的任务firstTask,在调用runTask()执行完firstTask之后,在while循环里面不断通过getTask()去取新的任务来执行,那么去哪里取呢?自然是从任务缓存队列里面去取,getTask是ThreadPoolExecutor类中的方法,并不是Worker类中的方法,下面是getTask方法的实现:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 Runnable getTask() {  for (;;) {  try {  int state = runState;  if (state > SHUTDOWN)  return null;  Runnable r;  if (state == SHUTDOWN) // Help drain queue  r = workQueue.poll();  else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间,  //则通过poll取任务,若等待一定的时间取不到任务,则返回null  r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);  else  r = workQueue.take();  if (r != null)  return r;  if (workerCanExit()) { //如果没取到任务,即r为null,则判断当前的worker是否可以退出  if (runState >= SHUTDOWN) // Wake up others  interruptIdleWorkers(); //中断处于空闲状态的worker  return null;  }  // Else retry  } catch (InterruptedException ie) {  // On interruption, re-check runState  }  } }

在getTask中,先判断当前线程池状态,如果runState大于SHUTDOWN(即为STOP或者TERMINATED),则直接返回null。

如果runState为SHUTDOWN或者RUNNING,则从任务缓存队列取任务。

如果当前线程池的线程数大于核心池大小corePoolSize或者允许为核心池中的线程设置空闲存活时间,则调用poll(time,timeUnit)来取任务,这个方法会等待一定的时间,如果取不到任务就返回null。

然后判断取到的任务r是否为null,为null则通过调用workerCanExit()方法来判断当前worker是否可以退出,我们看一下workerCanExit()的实现:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 private boolean workerCanExit() {  final ReentrantLock mainLock = this.mainLock;  mainLock.lock();  boolean canExit;  //如果runState大于等于STOP,或者任务缓存队列为空了  //或者 允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1  try {  canExit = runState >= STOP ||  workQueue.isEmpty() ||  (allowCoreThreadTimeOut &&  poolSize > Math.max(1, corePoolSize));  } finally {  mainLock.unlock();  }  return canExit; }

也就是说如果线程池处于STOP状态、或者任务队列已为空或者允许为核心池线程设置空闲存活时间并且线程数大于1时,允许worker退出。如果允许worker退出,则调用interruptIdleWorkers()中断处于空闲状态的worker,我们看一下interruptIdleWorkers()的实现:

1 2 3 4 5 6 7 8 9 10 void interruptIdleWorkers() {  final ReentrantLock mainLock = this.mainLock;  mainLock.lock();  try {  for (Worker w : workers) //实际上调用的是worker的interruptIfIdle()方法  w.interruptIfIdle();  } finally {  mainLock.unlock();  } }

从实现可以看出,它实际上调用的是worker的interruptIfIdle()方法,在worker的interruptIfIdle()方法中:

1 2 3 4 5 6 7 8 9 10 11 12 void interruptIfIdle() {  final ReentrantLock runLock = this.runLock;  if (runLock.tryLock()) { //注意这里,是调用tryLock()来获取锁的,因为如果当前worker正在执行任务,锁已经被获取了,是无法获取到锁的  //如果成功获取了锁,说明当前worker处于空闲状态  try {  if (thread != Thread.currentThread())  thread.interrupt();  } finally {  runLock.unlock();  }  } }

这里有一个非常巧妙的设计方式,假如我们来设计线程池,可能会有一个任务分派线程,当发现有线程空闲时,就从任务缓存队列中取一个任务交给空闲线程执行。但是在这里,并没有采用这样的方式,因为这样会要额外地对任务分派线程进行管理,无形地会增加难度和复杂度,这里直接让执行完任务的线程去任务缓存队列里面取任务来执行。

我们再看addIfUnderMaximumPoolSize方法的实现,这个方法的实现思想和addIfUnderCorePoolSize方法的实现思想非常相似,唯一的区别在于addIfUnderMaximumPoolSize方法是在线程池中的线程数达到了核心池大小并且往任务队列中添加任务失败的情况下执行的:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {  Thread t = null;  final ReentrantLock mainLock = this.mainLock;  mainLock.lock();  try {  if (poolSize < maximumPoolSize && runState == RUNNING)  t = addThread(firstTask);  } finally {  mainLock.unlock();  }  if (t == null)  return false;  t.start();  return true; }

看到没有,其实它和addIfUnderCorePoolSize方法的实现基本一模一样,只是if语句判断条件中的poolSize
< maximumPoolSize不同而已。

到这里,大部分朋友应该对任务提交给线程池之后到被执行的整个过程有了一个基本的了解,下面总结一下:

1)首先,要清楚corePoolSize和maximumPoolSize的含义;

2)其次,要知道Worker是用来起到什么作用的;

3)要知道任务提交给线程池之后的处理策略,这里总结一下主要有4点:

如果当前线程池中的线程数目小于corePoolSize,则每来一个任务,就会创建一个线程去执行这个任务;
如果当前线程池中的线程数目>=corePoolSize,则每来一个任务,会尝试将其添加到任务缓存队列当中,若添加成功,则该任务会等待空闲线程将其取出去执行;若添加失败(一般来说是任务缓存队列已满),则会尝试创建新的线程去执行这个任务;
如果当前线程池中的线程数目达到maximumPoolSize,则会采取任务拒绝策略进行处理;
如果线程池中的线程数量大于
corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止,直至线程池中的线程数目不大于corePoolSize;如果允许为核心池中的线程设置存活时间,那么核心池中的线程空闲时间超过keepAliveTime,线程也会被终止。
3.线程池中的线程初始化

默认情况下,创建线程池之后,线程池中是没有线程的,需要提交任务之后才会创建线程。

在实际中如果需要线程池创建之后立即创建线程,可以通过以下两个方法办到:

  • prestartCoreThread():初始化一个核心线程;
  • prestartAllCoreThreads():初始化所有核心线程

下面是这2个方法的实现:

1 2 3 4 5 6 7 8 9 10 public boolean prestartCoreThread() {  return addIfUnderCorePoolSize(null); //注意传进去的参数是null }   public int prestartAllCoreThreads() {  int n = 0;  while (addIfUnderCorePoolSize(null))//注意传进去的参数是null  ++n;  return n; }

注意上面传进去的参数是null,根据第2小节的分析可知如果传进去的参数为null,则最后执行线程会阻塞在getTask方法中的

1 r = workQueue.take();

即等待任务队列中有任务。

4.任务缓存队列及排队策略

在前面我们多次提到了任务缓存队列,即workQueue,它用来存放等待执行的任务。

workQueue的类型为BlockingQueue<Runnable>,通常可以取下面三种类型:

1)ArrayBlockingQueue:基于数组的先进先出队列,此队列创建时必须指定大小;

2)LinkedBlockingQueue:基于链表的先进先出队列,如果创建时没有指定此队列大小,则默认为Integer.MAX_VALUE;

3)synchronousQueue:这个队列比较特殊,它不会保存提交的任务,而是将直接新建一个线程来执行新来的任务。

5.任务拒绝策略

当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略,通常有以下四种策略:

1 2 3 4 ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。 ThreadPoolExecutor.DiscardPolicy:也是丢弃任务,但是不抛出异常。 ThreadPoolExecutor.DiscardOldestPolicy:丢弃队列最前面的任务,然后重新尝试执行任务(重复此过程) ThreadPoolExecutor.CallerRunsPolicy:由调用线程处理该任务

6.线程池的关闭

ThreadPoolExecutor提供了两个方法,用于线程池的关闭,分别是shutdown()和shutdownNow(),其中:

  • shutdown():不会立即终止线程池,而是要等所有任务缓存队列中的任务都执行完后才终止,但再也不会接受新的任务
  • shutdownNow():立即终止线程池,并尝试打断正在执行的任务,并且清空任务缓存队列,返回尚未执行的任务

7.线程池容量的动态调整

ThreadPoolExecutor提供了动态调整线程池容量大小的方法:setCorePoolSize()和setMaximumPoolSize(),

  • setCorePoolSize:设置核心池大小
  • setMaximumPoolSize:设置线程池最大能创建的线程数目大小

当上述参数从小变大时,ThreadPoolExecutor进行线程赋值,还可能立即创建新的线程来执行任务。

三.使用示例

前面我们讨论了关于线程池的实现原理,这一节我们来看一下它的具体使用:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 public class Test {  public static void main(String[] args) {  ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,  new ArrayBlockingQueue<Runnable>(5));    for(int i=0;i<15;i++){  MyTask myTask = new MyTask(i);  executor.execute(myTask);  System.out.println("线程池中线程数目:"+executor.getPoolSize()+",队列中等待执行的任务数目:"+  executor.getQueue().size()+",已执行玩别的任务数目:"+executor.getCompletedTaskCount());  }  executor.shutdown();  } }   class MyTask implements Runnable {  private int taskNum;    public MyTask(int num) {  this.taskNum = num;  }    @Override  public void run() {  System.out.println("正在执行task "+taskNum);  try {  Thread.currentThread().sleep(4000);  } catch (InterruptedException e) {  e.printStackTrace();  }  System.out.println("task "+taskNum+"执行完毕");  } }

执行结果:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 正在执行task 0 线程池中线程数目:1,队列中等待执行的任务数目:0,已执行玩别的任务数目:0 线程池中线程数目:2,队列中等待执行的任务数目:0,已执行玩别的任务数目:0 正在执行task 1 线程池中线程数目:3,队列中等待执行的任务数目:0,已执行玩别的任务数目:0 正在执行task 2 线程池中线程数目:4,队列中等待执行的任务数目:0,已执行玩别的任务数目:0 正在执行task 3 线程池中线程数目:5,队列中等待执行的任务数目:0,已执行玩别的任务数目:0 正在执行task 4 线程池中线程数目:5,队列中等待执行的任务数目:1,已执行玩别的任务数目:0 线程池中线程数目:5,队列中等待执行的任务数目:2,已执行玩别的任务数目:0 线程池中线程数目:5,队列中等待执行的任务数目:3,已执行玩别的任务数目:0 线程池中线程数目:5,队列中等待执行的任务数目:4,已执行玩别的任务数目:0 线程池中线程数目:5,队列中等待执行的任务数目:5,已执行玩别的任务数目:0 线程池中线程数目:6,队列中等待执行的任务数目:5,已执行玩别的任务数目:0 正在执行task 10 线程池中线程数目:7,队列中等待执行的任务数目:5,已执行玩别的任务数目:0 正在执行task 11 线程池中线程数目:8,队列中等待执行的任务数目:5,已执行玩别的任务数目:0 正在执行task 12 线程池中线程数目:9,队列中等待执行的任务数目:5,已执行玩别的任务数目:0 正在执行task 13 线程池中线程数目:10,队列中等待执行的任务数目:5,已执行玩别的任务数目:0 正在执行task 14 task 3执行完毕 task 0执行完毕 task 2执行完毕 task 1执行完毕 正在执行task 8 正在执行task 7 正在执行task 6 正在执行task 5 task 4执行完毕 task 10执行完毕 task 11执行完毕 task 13执行完毕 task 12执行完毕 正在执行task 9 task 14执行完毕 task 8执行完毕 task 5执行完毕 task 7执行完毕 task 6执行完毕 task 9执行完毕

从执行结果可以看出,当线程池中线程的数目大于5时,便将任务放入任务缓存队列里面,当任务缓存队列满了之后,便创建新的线程。如果上面程序中,将for循环中改成执行20个任务,就会抛出任务拒绝异常了。

不过在java
doc中,并不提倡我们直接使用ThreadPoolExecutor,而是使用Executors类中提供的几个静态方法来创建线程池:

1 2 3 Executors.newCachedThreadPool(); //创建一个缓冲池,缓冲池容量大小为Integer.MAX_VALUE Executors.newSingleThreadExecutor(); //创建容量为1的缓冲池 Executors.newFixedThreadPool(int); //创建固定容量大小的缓冲池

下面是这三个静态方法的具体实现;

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 public static ExecutorService newFixedThreadPool(int nThreads) {  return new ThreadPoolExecutor(nThreads, nThreads,  0L, TimeUnit.MILLISECONDS,  new LinkedBlockingQueue<Runnable>()); } public static ExecutorService newSingleThreadExecutor() {  return new FinalizableDelegatedExecutorService  (new ThreadPoolExecutor(1, 1,  0L, TimeUnit.MILLISECONDS,  new LinkedBlockingQueue<Runnable>())); } public static ExecutorService newCachedThreadPool() {  return new ThreadPoolExecutor(0, Integer.MAX_VALUE,  60L, TimeUnit.SECONDS,  new SynchronousQueue<Runnable>()); }

从它们的具体实现来看,它们实际上也是调用了ThreadPoolExecutor,只不过参数都已配置好了。

newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的,它使用的LinkedBlockingQueue;

newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1,也使用的LinkedBlockingQueue;

newCachedThreadPool将corePoolSize设置为0,将maximumPoolSize设置为Integer.MAX_VALUE,使用的SynchronousQueue,也就是说来了任务就创建线程运行,当线程空闲超过60秒,就销毁线程。

实际中,如果Executors提供的三个静态方法能满足要求,就尽量使用它提供的三个方法,因为自己去手动配置ThreadPoolExecutor的参数有点麻烦,要根据实际任务的类型和数量来进行配置。

另外,如果ThreadPoolExecutor达不到要求,可以自己继承ThreadPoolExecutor类进行重写。

四.如何合理配置线程池的大小

本节来讨论一个比较重要的话题:如何合理配置线程池大小,仅供参考。

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

如果是CPU密集型任务,就需要尽量压榨CPU,参考值可以设为 NCPU+1

如果是IO密集型任务,参考值可以设置为2*NCPU

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

参考资料:

《JDK API 1.6》

问啊-一键呼叫程序员答题神器,牛人一对一服务,开发者编程必备官方网站:www.wenaaa.com

QQ群290551701
聚集很多互联网精英,技术总监,架构师,项目经理!开源技术研究,欢迎业内人士,大牛及新手有志于从事IT行业人员进入!

在前面的文章中,我们使用线程的时候就去创建一个线程,这样实现起来非常简便,但是就会…

  转载自:

java.util.concurrent.ExecutorService
接口表示一个异步执行机制,使我们能够在后台执行任务。因此一个
ExecutorService 很类似于一个线程池。实际上,存在于 java.util.concurrent
包里的 ExecutorService 实现就是一个线程池实现。

ThreadPoolExecutor类

java.util.concurrent.TheadPoolExecutor 继承自
java.util.cocurrent.AbstractExecutorService
implements 接口 Executor ExecutorService

线程池解决两个问题:解决大量异步工作中,减少每个线程的创建开销,提供了一种执行一组任务集合的管理资源的方法,资源包括
线程,消耗。

构造函数,

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

corePoolSize 核心池的大小
maximumPoolSize 最大线程数
keepAliveTime 没有任务执行时 最多保持多久会 终止
TimeUnit unit 七种时间单位 天,小时,分钟,秒
BlockingQueue

  • 线程池大小
    • 核心线程池大小。
    • 最大线程数。

在创建了线程池后,默认情况下,线程池中并没有任何线程,等到有任务来才创建线程去执行任务,
除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法
当创建的线程数等于 corePoolSize 时,会加入设置的阻塞队列。
当队列满时,会创建线程执行任务直到线程池中的数量等于maximumPoolSize。

  • 阻塞队列

ArrayBlockingQueue : 由数组结构组成的有界阻塞队列。
LinkedBlockingQueue : 由链表结构组成的有界阻塞队列。
PriorityBlockingQueue :支持优先级排序的无界阻塞队列。
DelayQueue: 使用优先级队列实现的无界阻塞队列。
SynchronousQueue: 不存储元素的阻塞队列。
LinkedTransferQueue: 由链表结构组成的无界阻塞队列。
LinkedBlockingDeque: 由链表结构组成的双向阻塞队列。

  • 拒绝策略 –

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

1. 为什么使用线程池

诸如 Web 服务器、数据库服务器、文件服务器或邮件服务器之类的许多服务器应用程序都面向处理来自某些远程来源的大量短小的任务。请求以某种方式到达服务器,这种方式可能是通过网络协议(例如 HTTP、FTP 或 POP)、通过 JMS 队列或者可能通过轮询数据库。不管请求如何到达,服务器应用程序中经常出现的情况是:单个任务处理的时间很短而请求的数目却是巨大的。

构建服务器应用程序的一个简单模型是:每当一个请求到达就创建一个新线程,然后在新线程中为请求服务。实际上对于原型开发这种方法工作得很好,但如果试图部署以这种方式运行的服务器应用程序,那么这种方法的严重不足就很明显。每个请求对应一个线程(thread-per-request)方法的不足之一是:为每个请求创建一个新线程的开销很大;为每个请求创建新线程的服务器在创建和销毁线程上花费的时间和消耗的系统资源要比花在处理实际的用户请求的时间和资源更多。

除了创建和销毁线程的开销之外,活动的线程也消耗系统资源。在一个 JVM 里创建太多的线程可能会导致系统由于过度消耗内存而用完内存或“切换过度”。为了防止资源不足,服务器应用程序需要一些办法来限制任何给定时刻处理的请求数目。

线程池为线程生命周期开销问题和资源不足问题提供了解决方案。通过对多个任务重用线程,线程创建的开销被分摊到了多个任务上。其好处是,因为在请求到达时线程已经存在,所以无意中也消除了线程创建所带来的延迟。这样,就可以立即为请求服务,使应用程序响应更快。而且,通过适当地调整线程池中的线程数目,也就是当请求的数目超过某个阈值时,就强制其它任何新到的请求一直等待,直到获得一个线程来处理为止,从而可以防止资源不足。

  在前面的文章中,我们使用线程的时候就去创建一个线程,这样实现起来非常简便,但是就会有一个问题:

以下是一个简单的 ExecutorService 例子:

Executors

Executors弊端:

1)newFixedThreadPool 和 newSingleThreadExecutor:
主要问题是堆积的请求处理队列可能会耗费非常大的内存,甚至 OOM。
2)newCachedThreadPool 和 newScheduledThreadPool:
主要问题是线程数最大数是
Integer.MAX_VALUE,可能会创建数量非常多的线程,甚至 OOM。

2. 使用线程池的风险

虽然线程池是构建多线程应用程序的强大机制,但使用它并不是没有风险的。用线程池构建的应用程序容易遭受任何其它多线程应用程序容易遭受的所有并发风险,诸如同步错误和死锁,它还容易遭受特定于线程池的少数其它风险,诸如与池有关的死锁、资源不足和线程泄漏。

  如果并发的线程数量很多,并且每个线程都是执行一个时间很短的任务就结束了,这样频繁创建线程就会大大降低系统的效率,因为频繁创建线程和销毁线程需要时间。

ExecutorService executorService = Executors.newFixedThreadPool; executorService.execute(new Runnable() { public void run() { System.out.println("Asynchronous task"); } }); executorService.shutdown(); 

Executor接口

public interface Executor {

    void execute(Runnable command);
}

2.1 死锁

任何多线程应用程序都有死锁风险。当一组进程或线程中的每一个都在等待一个只有该组中另一个进程才能引起的事件时,我们就说这组进程或线程 死锁了。死锁的最简单情形是:线程 A 持有对象 X 的独占锁,并且在等待对象 Y 的锁,而线程 B 持有对象 Y 的独占锁,却在等待对象 X 的锁。除非有某种方法来打破对锁的等待(Java 锁定不支持这种方法),否则死锁的线程将永远等下去。

虽然任何多线程程序中都有死锁的风险,但线程池却引入了另一种死锁可能,在那种情况下,所有池线程都在执行已阻塞的等待队列中另一任务的执行结果的任务,但这一任务却因为没有未被占用的线程而不能运行。当线程池被用来实现涉及许多交互对象的模拟,被模拟的对象可以相互发送查询,这些查询接下来作为排队的任务执行,查询对象又同步等待着响应时,会发生这种情况。

  那么有没有一种办法使得线程可以复用,就是执行完一个任务,并不被销毁,而是可以继续执行其他的任务?

首先使用 newFixedThreadPool() 工厂方法创建一个
ExecutorService。这里创建了一个十个线程执行任务的线程池。然后,将一个
Runnable 接口的匿名实现类传递给 execute() 方法。这将导致 ExecutorService
中的某个线程执行该 Runnable。

ExecutorService接口

public interface ExecutorService extends Executor {

    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

    <T> Future<T> submit(Callable<T> task);

    <T> Future<T> submit(Runnable task, T result);

    Future<?> submit(Runnable task);

    invokeAll(Collection<? extends Callable<T>> tasks)

An Executor that provides methods to manage termination and methods that
can produce a Future for tracking progress of one or more asynchronous
tasks.

ExecutorService 可以被停止,会导致拒绝新任务task。
两种shutdown方法。
shutdown(),允许之前提交的任务继续执行完毕,但不再接受新任务。
shutdownNow(); 所有任务 立即停止,返回结果。

public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        /*
         *三步
         * 1
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.

任务从提交到执行的步骤:

public void execute(Runnable command){
    //...
    //三步: 
    1.如果少于corePoolSize在运行,尝试启动一个新线程,并将command给他。
    call addWorker
    addIfUnderCorePoolSize(command)
=
         * 1. If fewer than corePoolSize threads are running, try to
         * start a new thread with the given command as its first
         * task.  The call to addWorker atomically checks runState and
         * workerCount, and so prevents false alarms that would add
         * threads when it shouldn't, by returning false.
         *
         * 2. If a task can be successfully queued, then we still need
         * to double-check whether we should have added a thread
         * (because existing ones died since last checking) or that
         * the pool shut down since entry into this method. So we
         * recheck state and if necessary roll back the enqueuing if
         * stopped, or start a new thread if there are none.
         *
         * 3. If we cannot queue task, then we try to add a new
         * thread.  If it fails, we know we are shut down or saturated
         * and so reject the task.


}

2.2 资源不足

线程池的一个优点在于:相对于其它替代调度机制(有些我们已经讨论过)而言,它们通常执行得很好。但只有恰当地调整了线程池大小时才是这样的。线程消耗包括内存和其它系统资源在内的大量资源。除了 Thread 对象所需的内存之外,每个线程都需要两个可能很大的执行调用堆栈。除此以外,JVM 可能会为每个 Java 线程创建一个本机线程,这些本机线程将消耗额外的系统资源。最后,虽然线程之间切换的调度开销很小,但如果有很多线程,环境切换也可能严重地影响程序的性能。

如果线程池太大,那么被那些线程消耗的资源可能严重地影响系统性能。在线程之间进行切换将会浪费时间,而且使用超出比您实际需要的线程可能会引起资源匮乏问题,因为池线程正在消耗一些资源,而这些资源可能会被其它任务更有效地利用。除了线程自身所使用的资源以外,服务请求时所做的工作可能需要其它资源,例如 JDBC 连接、套接字或文件。这些也都是有限资源,有太多的并发请求也可能引起失效,例如不能分配 JDBC 连接。

  在Java中可以通过线程池来达到这样的效果。今天我们就来详细讲解一下Java的线程池,首先我们从最核心的ThreadPoolExecutor类中的方法讲起,然后再讲述它的实现原理,接着给出了它的使用示例,最后讨论了一下如何合理配置线程池的大小。

下图说明了一个线程是如何将一个任务委托给一个 ExecutorService
去异步执行的:

参考资料:

从阿里Java开发手册学习线程池的正确创建方法
线程池的原理以及java的线程池框架

线程池的原理以及java的线程池框架

2.3 并发错误

线程池和其它排队机制依靠使用 wait() 和 notify() 方法,这两个方法都难于使用。如果编码不正确,那么可能丢失通知,导致线程保持空闲状态,尽管队列中有工作要处理。使用这些方法时,必须格外小心。而最好使用现有的、已经知道能工作的实现,例如 util.concurrent 包。

  以下是本文的目录大纲:

奥门威尼斯网址 2image.png

2.4 线程泄漏

各种类型的线程池中一个严重的风险是线程泄漏,当从池中除去一个线程以执行一项任务,而在任务完成后该线程却没有返回池时,会发生这种情况。发生线程泄漏的一种情形出现在任务抛出一个 RuntimeException 或一个 Error 时。如果池类没有捕捉到它们,那么线程只会退出而线程池的大小将会永久减少一个。当这种情况发生的次数足够多时,线程池最终就为空,而且系统将停止,因为没有可用的线程来处理任务。

有些任务可能会永远等待某些资源或来自用户的输入,而这些资源又不能保证变得可用,用户可能也已经回家了,诸如此类的任务会永久停止,而这些停止的任务也会引起和线程泄漏同样的问题。如果某个线程被这样一个任务永久地消耗着,那么它实际上就被从池除去了。对于这样的任务,应该要么只给予它们自己的线程,要么只让它们等待有限的时间。

  一、Java中的ThreadPoolExecutor类

一个线程将一个任务委派给一个 ExecutorService 去异步执行。

2.5 请求过载

仅仅是请求就压垮了服务器,这种情况是可能的。在这种情形下,我们可能不想将每个到来的请求都排队到我们的工作队列,因为排在队列中等待执行的任务可能会消耗太多的系统资源并引起资源缺乏。在这种情形下决定如何做取决于您自己;在某些情况下,您可以简单地抛弃请求,依靠更高级别的协议稍后重试请求,您也可以用一个指出服务器暂时很忙的响应来拒绝请求。

  二、深入剖析线程池实现原理

一旦该线程将任务委派给
ExecutorService,该线程将继续它自己的执行,独立于该任务的执行。

3. 有效使用线程池的准则

只要您遵循几条简单的准则,线程池可以成为构建服务器应用程序的极其有效的方法:

不要对那些同步等待其它任务结果的任务排队。这可能会导致上面所描述的那种形式的死锁,在那种死锁中,所有线程都被一些任务所占用,这些任务依次等待排队任务的结果,而这些任务又无法执行,因为所有的线程都很忙。

在为时间可能很长的操作使用合用的线程时要小心。如果程序必须等待诸如 I/O 完成这样的某个资源,那么请指定最长的等待时间,以及随后是失效还是将任务重新排队以便稍后执行。这样做保证了:通过将某个线程释放给某个可能成功完成的任务,从而将最终取得某些进展。

理解任务。要有效地调整线程池大小,您需要理解正在排队的任务以及它们正在做什么。它们是 CPU 限制的(CPU-bound)吗?它们是I/O 限制的(I/O-bound)吗?您的答案将影响您如何调整应用程序。如果您有不同的任务类,这些类有着截然不同的特征,那么为不同任务类设置多个工作队列可能会有意义,这样可以相应地调整每个池。

  三、使用示例

既然 ExecutorService
是个接口,如果你想用它的话就得去使用它的实现类之一。

4. 线程池的大小设置

调整线程池的大小基本上就是避免两类错误:线程太少或线程太多。幸运的是,对于大多数应用程序来说,太多和太少之间的余地相当宽。

请回忆:在应用程序中使用线程有两个主要优点,尽管在等待诸如 I/O 的慢操作,但允许继续进行处理,并且可以利用多处理器。在运行于具有 N 个处理器机器上的计算限制的应用程序中,在线程数目接近 N 时添加额外的线程可能会改善总处理能力,而在线程数目超过 N时添加额外的线程将不起作用。事实上,太多的线程甚至会降低性能,因为它会导致额外的环境切换开销。

线程池的最佳大小取决于可用处理器的数目以及工作队列中的任务的性质。若在一个具有 N 个处理器的系统上只有一个工作队列,其中全部是计算性质的任务,在线程池具有 N 或 N+1 个线程时一般会获得最大的 CPU 利用率。

对于那些可能需要等待 I/O 完成的任务(例如,从套接字读取 HTTP 请求的任务),需要让池的大小超过可用处理器的数目,因为并不是所有线程都一直在工作。通过使用概要分析,您可以估计某个典型请求的等待时间(WT)与服务时间(ST)之间的比例。如果我们将这一比例称之为 WT/ST,那么对于一个具有 N 个处理器的系统,需要设置大约 N*(1+WT/ST) 个线程来保持处理器得到充分利用。

处理器利用率不是调整线程池大小过程中的唯一考虑事项。随着线程池的增长,您可能会碰到调度程序、可用内存方面的限制,或者其它系统资源方面的限制,例如套接字、打开的文件句柄或数据库连接等的数目。

  四、如何合理配置线程池的大小  

java.util.concurrent 包提供了 ExecutorService 接口的以下实现类:

5. 常用的几种线程池

一、Java中的ThreadPoolExecutor类

  java.uitl.concurrent.ThreadPoolExecutor类是线程池中最核心的一个类,因此如果要透彻地了解Java中的线程池,必须先了解这个类。下面我们来看一下ThreadPoolExecutor类的具体实现源码。

  在ThreadPoolExecutor类中提供了四个构造方法:

public class ThreadPoolExecutor extends AbstractExecutorService {
    .....
    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);
    ...
}

   从上面的代码可以得知,ThreadPoolExecutor继承了AbstractExecutorService类,并提供了四个构造器,事实上,通过观察每个构造器的源码具体实现,发现前面三个构造器都是调用的第四个构造器进行的初始化工作。

   下面解释下一下构造器中各个参数的含义:

  • corePoolSize:核心池的大小,这个参数跟后面讲述的线程池的实现原理有非常大的关系。在创建了线程池后,默认情况下,线程池中并没有任何线程,而是等待有任务到来才创建线程去执行任务,除非调用了prestartAllCoreThreads()或者prestartCoreThread()方法,从这2个方法的名字就可以看出,是预创建线程的意思,即在没有任务到来之前就创建corePoolSize个线程或者一个线程。默认情况下,在创建了线程池后,线程池中的线程数为0,当有任务来之后,就会创建一个线程去执行任务,当线程池中的线程数目达到corePoolSize后,就会把到达的任务放到缓存队列当中;
  • maximumPoolSize:线程池最大线程数,这个参数也是一个非常重要的参数,它表示在线程池中最多能创建多少个线程;
  • keepAliveTime:表示线程没有任务执行时最多保持多久时间会终止。默认情况下,只有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起作用,直到线程池中的线程数不大于corePoolSize,即当线程池中的线程数大于corePoolSize时,如果一个线程空闲的时间达到keepAliveTime,则会终止,直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不大于corePoolSize时,keepAliveTime参数也会起作用,直到线程池中的线程数为0;
  • unit:参数keepAliveTime的时间单位,有7种取值,在TimeUnit类中有7种静态属性:

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

  • workQueue:一个阻塞队列,用来存储等待执行的任务,这个参数的选择也很重要,会对线程池的运行过程产生重大影响,一般来说,这里的阻塞队列有以下几种选择:

    ArrayBlockingQueue;
    LinkedBlockingQueue;
    SynchronousQueue;

  ArrayBlockingQueue和PriorityBlockingQueue使用较少,一般使用LinkedBlockingQueue和Synchronous。线程池的排队策略与BlockingQueue有关。

  • threadFactory:线程工厂,主要用来创建线程;
  • handler:表示当拒绝处理任务时的策略,有以下四种取值:

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

   具体参数的配置与线程池的关系将在下一节讲述。

  从上面给出的ThreadPoolExecutor类的代码可以知道,ThreadPoolExecutor继承了AbstractExecutorService,我们来看一下AbstractExecutorService的实现:

public abstract class AbstractExecutorService implements ExecutorService {


    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) { };
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) { };
    public Future<?> submit(Runnable task) {};
    public <T> Future<T> submit(Runnable task, T result) { };
    public <T> Future<T> submit(Callable<T> task) { };
    private <T> T doInvokeAny(Collection<? extends Callable<T>> tasks,
                            boolean timed, long nanos)
        throws InterruptedException, ExecutionException, TimeoutException {
    };
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException {
    };
    public <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                           long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException {
    };
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException {
    };
    public <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                         long timeout, TimeUnit unit)
        throws InterruptedException {
    };
}

   AbstractExecutorService是一个抽象类,它实现了ExecutorService接口。

  我们接着看ExecutorService接口的实现:

public interface ExecutorService extends Executor {

    void shutdown();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit)
        throws InterruptedException;
    <T> Future<T> submit(Callable<T> task);
    <T> Future<T> submit(Runnable task, T result);
    Future<?> submit(Runnable task);
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
        throws InterruptedException;
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                  long timeout, TimeUnit unit)
        throws InterruptedException;

    <T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;
    <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                    long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

   而ExecutorService又是继承了Executor接口,我们看一下Executor接口的实现:

public interface Executor {
    void execute(Runnable command);
}

   到这里,大家应该明白了ThreadPoolExecutor、AbstractExecutorService、ExecutorService和Executor几个之间的关系了。

  Executor是一个顶层接口,在它里面只声明了一个方法execute(Runnable),返回值为void,参数为Runnable类型,从字面意思可以理解,就是用来执行传进去的任务的;

  然后ExecutorService接口继承了Executor接口,并声明了一些方法:submit、invokeAll、invokeAny以及shutDown等;

  抽象类AbstractExecutorService实现了ExecutorService接口,基本实现了ExecutorService中声明的所有方法;

  然后ThreadPoolExecutor继承了类AbstractExecutorService。

  在ThreadPoolExecutor类中有几个非常重要的方法:

execute()
submit()
shutdown()
shutdownNow()

   execute()方法实际上是Executor中声明的方法,在ThreadPoolExecutor进行了具体的实现,这个方法是ThreadPoolExecutor的核心方法,通过这个方法可以向线程池提交一个任务,交由线程池去执行。

  submit()方法是在ExecutorService中声明的方法,在AbstractExecutorService就已经有了具体的实现,在ThreadPoolExecutor中并没有对其进行重写,这个方法也是用来向线程池提交任务的,但是它和execute()方法不同,它能够返回任务执行的结果,去看submit()方法的实现,会发现它实际上还是调用的execute()方法,只不过它利用了Future来获取任务执行结果(Future相关内容将在下一篇讲述)。

  shutdown()和shutdownNow()是用来关闭线程池的。

  还有很多其他的方法:

  比如:getQueue() 、getPoolSize()
、getActiveCount()、getCompletedTaskCount()等获取与线程池相关属性的方法,有兴趣的朋友可以自行查阅API。

  • ThreadPoolExecutor
  • ScheduledThreadPoolExecutor

5.1 newCachedThreadPool

创建一个可缓存线程池,如果线程池长度超过处理需要,可灵活回收空闲线程,若无可回收,则新建线程。

这种类型的线程池特点是:

  • 工作线程的创建数量几乎没有限制(其实也有限制的,数目为Interger.
    MAX_VALUE), 这样可灵活的往线程池中添加线程。
  • 如果长时间没有往线程池中提交任务,即如果工作线程空闲了指定的时间(默认为1分钟),则该工作线程将自动终止。终止后,如果你又提交了新的任务,则线程池重新创建一个工作线程。
  • 在使用CachedThreadPool时,一定要注意控制任务的数量,否则,由于大量线程同时运行,很有会造成系统瘫痪。

示例代码如下:

奥门威尼斯网址 3

 1 package test;
 2 import java.util.concurrent.ExecutorService;
 3 import java.util.concurrent.Executors;
 4 public class ThreadPoolExecutorTest {
 5  public static void main(String[] args) {
 6   ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
 7   for (int i = 0; i < 10; i++) {
 8    final int index = i;
 9    try {
10     Thread.sleep(index * 1000);
11    } catch (InterruptedException e) {
12     e.printStackTrace();
13    }
14    cachedThreadPool.execute(new Runnable() {
15     public void run() {
16      System.out.println(index);
17     }
18    });
19   }
20  }
21 }

奥门威尼斯网址 4

二、深入剖析线程池实现原理

  在上一节我们从宏观上介绍了ThreadPoolExecutor,下面我们来深入解析一下线程池的具体实现原理,将从下面几个方面讲解:

  1.线程池状态

  2.任务的执行

  3.线程池中的线程初始化

  4.任务缓存队列及排队策略

  5.任务拒绝策略

  6.线程池的关闭

  7.线程池容量的动态调整

ExecutorService 的创建依赖于你使用的具体实现。但是你也可以使用 Executors
工厂类来创建 ExecutorService 实例。

5.1 newFixedThreadPool

创建一个指定工作线程数量的线程池。每当提交一个任务就创建一个工作线程,如果工作线程数量达到线程池初始的最大数,则将提交的任务存入到池队列中。

FixedThreadPool是一个典型且优秀的线程池,它具有线程池提高程序效率和节省创建线程时所耗的开销的优点。但是,在线程池空闲时,即线程池中没有可运行任务时,它不会释放工作线程,还会占用一定的系统资源。

示例代码如下:

 

奥门威尼斯网址 5

package test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest {
 public static void main(String[] args) {
  ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
  for (int i = 0; i < 10; i++) {
   final int index = i;
   fixedThreadPool.execute(new Runnable() {
    public void run() {
     try {
      System.out.println(index);
      Thread.sleep(2000);
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }
   });
  }
 }
}

奥门威尼斯网址 6

 

因为线程池大小为3,每个任务输出index后sleep
2秒,所以每两秒打印3个数字。
定长线程池的大小最好根据系统资源进行设置如Runtime.getRuntime().availableProcessors()。

 

1)线程池状态

  在ThreadPoolExecutor中定义了一个volatile变量,另外定义了几个static
final变量表示线程池的各个状态:

volatile int runState;
static final int RUNNING    = 0;
static final int SHUTDOWN   = 1;
static final int STOP       = 2;
static final int TERMINATED = 3;

   runState表示当前线程池的状态,它是一个volatile变量用来保证线程之间的可见性;

  下面的几个static final变量表示runState可能的几个取值。

  当创建线程池后,初始时,线程池处于RUNNING状态;

  如果调用了shutdown()方法,则线程池处于SHUTDOWN状态,此时线程池不能够接受新的任务,它会等待所有任务执行完毕;

  如果调用了shutdownNow()方法,则线程池处于STOP状态,此时线程池不能接受新的任务,并且会去尝试终止正在执行的任务;

  当线程池处于SHUTDOWN或STOP状态,并且所有工作线程已经销毁,任务缓存队列已经清空或执行结束后,线程池被设置为TERMINATED状态。

以下是几个创建 ExecutorService 实例的例子:

5.1 newSingleThreadExecutor

创建一个单线程化的Executor,即只创建唯一的工作者线程来执行任务,它只会用唯一的工作线程来执行任务,保证所有任务按照指定顺序(FIFO,
LIFO, 优先级)执行。如果这个线程异常结束,会有另一个取代它,保证顺序执行。单工作线程最大的特点是可保证顺序地执行各个任务,并且在任意给定的时间不会有多个线程是活动的。

示例代码如下:

 

奥门威尼斯网址 7

package test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExecutorTest {
 public static void main(String[] args) {
  ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
  for (int i = 0; i < 10; i++) {
   final int index = i;
   singleThreadExecutor.execute(new Runnable() {
    public void run() {
     try {
      System.out.println(index);
      Thread.sleep(2000);
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
    }
   });
  }
 }
} 

奥门威尼斯网址 8

 

2)任务的执行

  在了解将任务提交给线程池到任务执行完毕整个过程之前,我们先来看一下ThreadPoolExecutor类中其他的一些比较重要成员变量:

private final BlockingQueue<Runnable> workQueue;              //任务缓存队列,用来存放等待执行的任务
private final ReentrantLock mainLock = new ReentrantLock();   //线程池的主要状态锁,对线程池状态(比如线程池大小
                                                              //、runState等)的改变都要使用这个锁
private final HashSet<Worker> workers = new HashSet<Worker>();  //用来存放工作集

private volatile long  keepAliveTime;    //线程存货时间   
private volatile boolean allowCoreThreadTimeOut;   //是否允许为核心线程设置存活时间
private volatile int   corePoolSize;     //核心池的大小(即线程池中的线程数目大于这个参数时,提交的任务会被放进任务缓存队列)
private volatile int   maximumPoolSize;   //线程池最大能容忍的线程数

private volatile int   poolSize;       //线程池中当前的线程数

private volatile RejectedExecutionHandler handler; //任务拒绝策略

private volatile ThreadFactory threadFactory;   //线程工厂,用来创建线程

private int largestPoolSize;   //用来记录线程池中曾经出现过的最大线程数

private long completedTaskCount;   //用来记录已经执行完毕的任务个数

   每个变量的作用都已经标明出来了,这里要重点解释一下corePoolSize、maximumPoolSize、largestPoolSize三个变量。

  corePoolSize在很多地方被翻译成核心池大小,其实我的理解这个就是线程池的大小。举个简单的例子:

  假如有一个工厂,工厂里面有10个工人,每个工人同时只能做一件任务。

  因此只要当10个工人中有工人是空闲的,来了任务就分配给空闲的工人做;

  当10个工人都有任务在做时,如果还来了任务,就把任务进行排队等待;

  如果说新任务数目增长的速度远远大于工人做任务的速度,那么此时工厂主管可能会想补救措施,比如重新招4个临时工人进来;

  然后就将任务也分配给这4个临时工人做;

  如果说着14个工人做任务的速度还是不够,此时工厂主管可能就要考虑不再接收新的任务或者抛弃前面的一些任务了。

  当这14个工人当中有人空闲时,而新任务增长的速度又比较缓慢,工厂主管可能就考虑辞掉4个临时工了,只保持原来的10个工人,毕竟请额外的工人是要花钱的。

 

  这个例子中的corePoolSize就是10,而maximumPoolSize就是14(10+4)。

  也就是说corePoolSize就是线程池大小,maximumPoolSize在我看来是线程池的一种补救措施,即任务量突然过大时的一种补救措施。

  不过为了方便理解,在本文后面还是将corePoolSize翻译成核心池大小。

  largestPoolSize只是一个用来起记录作用的变量,用来记录线程池中曾经有过的最大线程数目,跟线程池的容量没有任何关系。

 

  下面我们进入正题,看一下任务从提交到最终执行完毕经历了哪些过程。

  在ThreadPoolExecutor类中,最核心的任务提交方法是execute()方法,虽然通过submit也可以提交任务,但是实际上submit方法里面最终调用的还是execute()方法,所以我们只需要研究execute()方法的实现原理即可:

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {
        if (runState == RUNNING && workQueue.offer(command)) {
            if (runState != RUNNING || poolSize == 0)
                ensureQueuedTaskHandled(command);
        }
        else if (!addIfUnderMaximumPoolSize(command))
            reject(command); // is shutdown or saturated
    }
}

   上面的代码可能看起来不是那么容易理解,下面我们一句一句解释:

  首先,判断提交的任务command是否为null,若是null,则抛出空指针异常;

  接着是这句,这句要好好理解一下:

if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command))

   由于是或条件运算符,所以先计算前半部分的值,如果线程池中当前线程数不小于核心池大小,那么就会直接进入下面的if语句块了。

  如果线程池中当前线程数小于核心池大小,则接着执行后半部分,也就是执行

addIfUnderCorePoolSize(command)

  如果执行完addIfUnderCorePoolSize这个方法返回false,则继续执行下面的if语句块,否则整个方法就直接执行完毕了。

  如果执行完addIfUnderCorePoolSize这个方法返回false,然后接着判断:

if (runState == RUNNING && workQueue.offer(command))

   如果当前线程池处于RUNNING状态,则将任务放入任务缓存队列;如果当前线程池不处于RUNNING状态或者任务放入缓存队列失败,则执行:

addIfUnderMaximumPoolSize(command)

  如果执行addIfUnderMaximumPoolSize方法失败,则执行reject()方法进行任务拒绝处理。

  回到前面:

if (runState == RUNNING && workQueue.offer(command))

   这句的执行,如果说当前线程池处于RUNNING状态且将任务放入任务缓存队列成功,则继续进行判断:

if (runState != RUNNING || poolSize == 0)

   这句判断是为了防止在将此任务添加进任务缓存队列的同时其他线程突然调用shutdown或者shutdownNow方法关闭了线程池的一种应急措施。如果是这样就执行:

ensureQueuedTaskHandled(command)

   进行应急处理,从名字可以看出是保证
添加到任务缓存队列中的任务得到处理。

  我们接着看2个关键方法的实现:addIfUnderCorePoolSize和addIfUnderMaximumPoolSize:

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
    Thread t = null;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (poolSize < corePoolSize && runState == RUNNING)
            t = addThread(firstTask);        //创建线程去执行firstTask任务   
        } finally {
        mainLock.unlock();
    }
    if (t == null)
        return false;
    t.start();
    return true;
}

   这个是addIfUnderCorePoolSize方法的具体实现,从名字可以看出它的意图就是当低于核心吃大小时执行的方法。下面看其具体实现,首先获取到锁,因为这地方涉及到线程池状态的变化,先通过if语句判断当前线程池中的线程数目是否小于核心池大小,有朋友也许会有疑问:前面在execute()方法中不是已经判断过了吗,只有线程池当前线程数目小于核心池大小才会执行addIfUnderCorePoolSize方法的,为何这地方还要继续判断?原因很简单,前面的判断过程中并没有加锁,因此可能在execute方法判断的时候poolSize小于corePoolSize,而判断完之后,在其他线程中又向线程池提交了任务,就可能导致poolSize不小于corePoolSize了,所以需要在这个地方继续判断。然后接着判断线程池的状态是否为RUNNING,原因也很简单,因为有可能在其他线程中调用了shutdown或者shutdownNow方法。然后就是执行

t = addThread(firstTask);

   这个方法也非常关键,传进去的参数为提交的任务,返回值为Thread类型。然后接着在下面判断t是否为空,为空则表明创建线程失败(即poolSize>=corePoolSize或者runState不等于RUNNING),否则调用t.start()方法启动线程。

  我们来看一下addThread方法的实现:

private Thread addThread(Runnable firstTask) {
    Worker w = new Worker(firstTask);
    Thread t = threadFactory.newThread(w);  //创建一个线程,执行任务   
    if (t != null) {
        w.thread = t;            //将创建的线程的引用赋值为w的成员变量       
        workers.add(w);
        int nt = ++poolSize;     //当前线程数加1       
        if (nt > largestPoolSize)
            largestPoolSize = nt;
    }
    return t;
}

   在addThread方法中,首先用提交的任务创建了一个Worker对象,然后调用线程工厂threadFactory创建了一个新的线程t,然后将线程t的引用赋值给了Worker对象的成员变量thread,接着通过workers.add(w)将Worker对象添加到工作集当中。

  下面我们看一下Worker类的实现:

private final class Worker implements Runnable {
    private final ReentrantLock runLock = new ReentrantLock();
    private Runnable firstTask;
    volatile long completedTasks;
    Thread thread;
    Worker(Runnable firstTask) {
        this.firstTask = firstTask;
    }
    boolean isActive() {
        return runLock.isLocked();
    }
    void interruptIfIdle() {
        final ReentrantLock runLock = this.runLock;
        if (runLock.tryLock()) {
            try {
        if (thread != Thread.currentThread())
        thread.interrupt();
            } finally {
                runLock.unlock();
            }
        }
    }
    void interruptNow() {
        thread.interrupt();
    }

    private void runTask(Runnable task) {
        final ReentrantLock runLock = this.runLock;
        runLock.lock();
        try {
            if (runState < STOP &&
                Thread.interrupted() &&
                runState >= STOP)
            boolean ran = false;
            beforeExecute(thread, task);   //beforeExecute方法是ThreadPoolExecutor类的一个方法,没有具体实现,用户可以根据
            //自己需要重载这个方法和后面的afterExecute方法来进行一些统计信息,比如某个任务的执行时间等            
            try {
                task.run();
                ran = true;
                afterExecute(task, null);
                ++completedTasks;
            } catch (RuntimeException ex) {
                if (!ran)
                    afterExecute(task, ex);
                throw ex;
            }
        } finally {
            runLock.unlock();
        }
    }

    public void run() {
        try {
            Runnable task = firstTask;
            firstTask = null;
            while (task != null || (task = getTask()) != null) {
                runTask(task);
                task = null;
            }
        } finally {
            workerDone(this);   //当任务队列中没有任务时,进行清理工作       
        }
    }
}

   它实际上实现了Runnable接口,因此上面的Thread t =
threadFactory.newThread(w);效果跟下面这句的效果基本一样:

Thread t = new Thread(w);

   相当于传进去了一个Runnable任务,在线程t中执行这个Runnable。

  既然Worker实现了Runnable接口,那么自然最核心的方法便是run()方法了:

public void run() {
    try {
        Runnable task = firstTask;
        firstTask = null;
        while (task != null || (task = getTask()) != null) {
            runTask(task);
            task = null;
        }
    } finally {
        workerDone(this);
    }
}

   从run方法的实现可以看出,它首先执行的是通过构造器传进来的任务firstTask,在调用runTask()执行完firstTask之后,在while循环里面不断通过getTask()去取新的任务来执行,那么去哪里取呢?自然是从任务缓存队列里面去取,getTask是ThreadPoolExecutor类中的方法,并不是Worker类中的方法,下面是getTask方法的实现:

Runnable getTask() {
    for (;;) {
        try {
            int state = runState;
            if (state > SHUTDOWN)
                return null;
            Runnable r;
            if (state == SHUTDOWN)  // Help drain queue
                r = workQueue.poll();
            else if (poolSize > corePoolSize || allowCoreThreadTimeOut) //如果线程数大于核心池大小或者允许为核心池线程设置空闲时间,
                //则通过poll取任务,若等待一定的时间取不到任务,则返回null
                r = workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS);
            else
                r = workQueue.take();
            if (r != null)
                return r;
            if (workerCanExit()) {    //如果没取到任务,即r为null,则判断当前的worker是否可以退出
                if (runState >= SHUTDOWN) // Wake up others
                    interruptIdleWorkers();   //中断处于空闲状态的worker
                return null;
            }
            // Else retry
        } catch (InterruptedException ie) {
            // On interruption, re-check runState
        }
    }
}

   在getTask中,先判断当前线程池状态,如果runState大于SHUTDOWN(即为STOP或者TERMINATED),则直接返回null。

  如果runState为SHUTDOWN或者RUNNING,则从任务缓存队列取任务。

  如果当前线程池的线程数大于核心池大小corePoolSize或者允许为核心池中的线程设置空闲存活时间,则调用poll(time,timeUnit)来取任务,这个方法会等待一定的时间,如果取不到任务就返回null。

  然后判断取到的任务r是否为null,为null则通过调用workerCanExit()方法来判断当前worker是否可以退出,我们看一下workerCanExit()的实现:

private boolean workerCanExit() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    boolean canExit;
    //如果runState大于等于STOP,或者任务缓存队列为空了
    //或者  允许为核心池线程设置空闲存活时间并且线程池中的线程数目大于1
    try {
        canExit = runState >= STOP ||
            workQueue.isEmpty() ||
            (allowCoreThreadTimeOut &&
             poolSize > Math.max(1, corePoolSize));
    } finally {
        mainLock.unlock();
    }
    return canExit;
}

   也就是说如果线程池处于STOP状态、或者任务队列已为空或者允许为核心池线程设置空闲存活时间并且线程数大于1时,允许worker退出。如果允许worker退出,则调用interruptIdleWorkers()中断处于空闲状态的worker,我们看一下interruptIdleWorkers()的实现:

void interruptIdleWorkers() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        for (Worker w : workers)  //实际上调用的是worker的interruptIfIdle()方法
            w.interruptIfIdle();
    } finally {
        mainLock.unlock();
    }
}

   从实现可以看出,它实际上调用的是worker的interruptIfIdle()方法,在worker的interruptIfIdle()方法中:

void interruptIfIdle() {
    final ReentrantLock runLock = this.runLock;
    if (runLock.tryLock()) {    //注意这里,是调用tryLock()来获取锁的,因为如果当前worker正在执行任务,锁已经被获取了,是无法获取到锁的
                                //如果成功获取了锁,说明当前worker处于空闲状态
        try {
    if (thread != Thread.currentThread())  
    thread.interrupt();
        } finally {
            runLock.unlock();
        }
    }
}

    这里有一个非常巧妙的设计方式,假如我们来设计线程池,可能会有一个任务分派线程,当发现有线程空闲时,就从任务缓存队列中取一个任务交给空闲线程执行。但是在这里,并没有采用这样的方式,因为这样会要额外地对任务分派线程进行管理,无形地会增加难度和复杂度,这里直接让执行完任务的线程去任务缓存队列里面取任务来执行。

   我们再看addIfUnderMaximumPoolSize方法的实现,这个方法的实现思想和addIfUnderCorePoolSize方法的实现思想非常相似,唯一的区别在于addIfUnderMaximumPoolSize方法是在线程池中的线程数达到了核心池大小并且往任务队列中添加任务失败的情况下执行的:

private boolean addIfUnderMaximumPoolSize(Runnable firstTask) {
    Thread t = null;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        if (poolSize < maximumPoolSize && runState == RUNNING)
            t = addThread(firstTask);
    } finally {
        mainLock.unlock();
    }
    if (t == null)
        return false;
    t.start();
    return true;
}

   看到没有,其实它和addIfUnderCorePoolSize方法的实现基本一模一样,只是if语句判断条件中的poolSize
< maximumPoolSize不同而已。

  到这里,大部分朋友应该对任务提交给线程池之后到被执行的整个过程有了一个基本的了解,下面总结一下:

  1)首先,要清楚corePoolSize和maximumPoolSize的含义;

  2)其次,要知道Worker是用来起到什么作用的;

  3)要知道任务提交给线程池之后的处理策略,这里总结一下主要有4点:

  • 如果当前线程池中的线程数目小于corePoolSize,则每来一个任务,就会创建一个线程去执行这个任务;
  • 如果当前线程池中的线程数目>=corePoolSize,则每来一个任务,会尝试将其添加到任务缓存队列当中,若添加成功,则该任务会等待空闲线程将其取出去执行;若添加失败(一般来说是任务缓存队列已满),则会尝试创建新的线程去执行这个任务;
  • 如果当前线程池中的线程数目达到maximumPoolSize,则会采取任务拒绝策略进行处理;
  • 如果线程池中的线程数量大于
    corePoolSize时,如果某线程空闲时间超过keepAliveTime,线程将被终止,直至线程池中的线程数目不大于corePoolSize;如果允许为核心池中的线程设置存活时间,那么核心池中的线程空闲时间超过keepAliveTime,线程也会被终止。
ExecutorService executorService1 = Executors.newSingleThreadExecutor(); ExecutorService executorService2 = Executors.newFixedThreadPool; ExecutorService executorService3 = Executors.newScheduledThreadPool; 

5.1 newScheduleThreadPool

创建一个定长的线程池,而且支持定时的以及周期性的任务执行,支持定时及周期性任务执行。

延迟3秒执行,延迟执行示例代码如下:

 

奥门威尼斯网址 9

package test;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExecutorTest {
 public static void main(String[] args) {
  ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  scheduledThreadPool.schedule(new Runnable() {
   public void run() {
    System.out.println("delay 3 seconds");
   }
  }, 3, TimeUnit.SECONDS);
 }
}

奥门威尼斯网址 10

 

表示延迟1秒后每3秒执行一次,定期执行示例代码如下:

 

奥门威尼斯网址 11

package test;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class ThreadPoolExecutorTest {
 public static void main(String[] args) {
  ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(5);
  scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
   public void run() {
    System.out.println("delay 1 seconds, and excute every 3 seconds");
   }
  }, 1, 3, TimeUnit.SECONDS);
 }
}

奥门威尼斯网址 12

1.
为什么使用线程池
诸如Web服务器、数据库服务器、文件服务器或邮件服务器之类的许多服务…

3)线程池中的线程初始化

  默认情况下,创建线程池之后,线程池中是没有线程的,需要提交任务之后才会创建线程。

  在实际中如果需要线程池创建之后立即创建线程,可以通过以下两个方法办到:

  • prestartCoreThread():初始化一个核心线程;
  • prestartAllCoreThreads():初始化所有核心线程

  下面是这2个方法的实现:

public boolean prestartCoreThread() {
    return addIfUnderCorePoolSize(null); //注意传进去的参数是null
}

public int prestartAllCoreThreads() {
    int n = 0;
    while (addIfUnderCorePoolSize(null))//注意传进去的参数是null
        ++n;
    return n;
}

   注意上面传进去的参数是null,根据第2小节的分析可知如果传进去的参数为null,则最后执行线程会阻塞在getTask方法中的

r = workQueue.take();

   即等待任务队列中有任务。

有几种不同的方式来将任务委托给 ExecutorService 去执行:

4)任务缓存队列及排队策略

  在前面我们多次提到了任务缓存队列,即workQueue,它用来存放等待执行的任务。

  workQueue的类型为BlockingQueue<Runnable>,通常可以取下面三种类型:

  1)ArrayBlockingQueue:基于数组的先进先出队列,此队列创建时必须指定大小;

  2)LinkedBlockingQueue:基于链表的先进先出队列,如果创建时没有指定此队列大小,则默认为Integer.MAX_VALUE;

  3)synchronousQueue:这个队列比较特殊,它不会保存提交的任务,而是将直接新建一个线程来执行新来的任务。

  • execute
  • submit
  • submit
  • invokeAny
  • invokeAll

5)任务拒绝策略

  当线程池的任务缓存队列已满并且线程池中的线程数目达到maximumPoolSize,如果还有任务到来就会采取任务拒绝策略,通常有以下四种策略:

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

接下来我们挨个看一下这些方法。

6)线程池的关闭

  ThreadPoolExecutor提供了两个方法,用于线程池的关闭,分别是shutdown()和shutdownNow(),其中:

  • shutdown():不会立即终止线程池,而是要等所有任务缓存队列中的任务都执行完后才终止,但再也不会接受新的任务
  • shutdownNow():立即终止线程池,并尝试打断正在执行的任务,并且清空任务缓存队列,返回尚未执行的任务

execute 方法要求一个 java.lang.Runnable
对象,然后对它进行异步执行。以下是使用 ExecutorService 执行一个 Runnable
的示例:

7)线程池容量的动态调整

  ThreadPoolExecutor提供了动态调整线程池容量大小的方法:setCorePoolSize()和setMaximumPoolSize(),

  • setCorePoolSize:设置核心池大小
  • setMaximumPoolSize:设置线程池最大能创建的线程数目大小

  当上述参数从小变大时,ThreadPoolExecutor进行线程赋值,还可能立即创建新的线程来执行任务。

ExecutorService executorService = Executors.newSingleThreadExecutor(); executorService.execute(new Runnable() { public void run() { System.out.println("Asynchronous task"); } }); executorService.shutdown(); 

三、使用示例

  前面我们讨论了关于线程池的实现原理,这一节我们来看一下它的具体使用:

public class Test {
     public static void main(String[] args) {   
         ThreadPoolExecutor executor = new ThreadPoolExecutor(5, 10, 200, TimeUnit.MILLISECONDS,
                 new ArrayBlockingQueue<Runnable>(5));

         for(int i=0;i<15;i++){
             MyTask myTask = new MyTask(i);
             executor.execute(myTask);
             System.out.println("线程池中线程数目:"+executor.getPoolSize()+",队列中等待执行的任务数目:"+
             executor.getQueue().size()+",已执行玩别的任务数目:"+executor.getCompletedTaskCount());
         }
         executor.shutdown();
     }
}


class MyTask implements Runnable {
    private int taskNum;

    public MyTask(int num) {
        this.taskNum = num;
    }

    @Override
    public void run() {
        System.out.println("正在执行task "+taskNum);
        try {
            Thread.currentThread().sleep(4000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("task "+taskNum+"执行完毕");
    }
}

   执行结果:

正在执行task 0
线程池中线程数目:1,队列中等待执行的任务数目:0,已执行玩别的任务数目:0
线程池中线程数目:2,队列中等待执行的任务数目:0,已执行玩别的任务数目:0
正在执行task 1
线程池中线程数目:3,队列中等待执行的任务数目:0,已执行玩别的任务数目:0
正在执行task 2
线程池中线程数目:4,队列中等待执行的任务数目:0,已执行玩别的任务数目:0
正在执行task 3
线程池中线程数目:5,队列中等待执行的任务数目:0,已执行玩别的任务数目:0
正在执行task 4
线程池中线程数目:5,队列中等待执行的任务数目:1,已执行玩别的任务数目:0
线程池中线程数目:5,队列中等待执行的任务数目:2,已执行玩别的任务数目:0
线程池中线程数目:5,队列中等待执行的任务数目:3,已执行玩别的任务数目:0
线程池中线程数目:5,队列中等待执行的任务数目:4,已执行玩别的任务数目:0
线程池中线程数目:5,队列中等待执行的任务数目:5,已执行玩别的任务数目:0
线程池中线程数目:6,队列中等待执行的任务数目:5,已执行玩别的任务数目:0
正在执行task 10
线程池中线程数目:7,队列中等待执行的任务数目:5,已执行玩别的任务数目:0
正在执行task 11
线程池中线程数目:8,队列中等待执行的任务数目:5,已执行玩别的任务数目:0
正在执行task 12
线程池中线程数目:9,队列中等待执行的任务数目:5,已执行玩别的任务数目:0
正在执行task 13
线程池中线程数目:10,队列中等待执行的任务数目:5,已执行玩别的任务数目:0
正在执行task 14
task 3执行完毕
task 0执行完毕
task 2执行完毕
task 1执行完毕
正在执行task 8
正在执行task 7
正在执行task 6
正在执行task 5
task 4执行完毕
task 10执行完毕
task 11执行完毕
task 13执行完毕
task 12执行完毕
正在执行task 9
task 14执行完毕
task 8执行完毕
task 5执行完毕
task 7执行完毕
task 6执行完毕
task 9执行完毕

  从执行结果可以看出,当线程池中线程的数目大于5时,便将任务放入任务缓存队列里面,当任务缓存队列满了之后,便创建新的线程。如果上面程序中,将for循环中改成执行20个任务,就会抛出任务拒绝异常了。

  不过在java
doc中,并不提倡我们直接使用ThreadPoolExecutor,而是使用Executors类中提供的几个静态方法来创建线程池:

Executors.newCachedThreadPool();        //创建一个缓冲池,缓冲池容量大小为Integer.MAX_VALUE
Executors.newSingleThreadExecutor();   //创建容量为1的缓冲池
Executors.newFixedThreadPool(int);    //创建固定容量大小的缓冲池

   下面是这三个静态方法的具体实现;

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}
public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}
public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

  从它们的具体实现来看,它们实际上也是调用了ThreadPoolExecutor,只不过参数都已配置好了。

  newFixedThreadPool创建的线程池corePoolSize和maximumPoolSize值是相等的,它使用的LinkedBlockingQueue;

  newSingleThreadExecutor将corePoolSize和maximumPoolSize都设置为1,也使用的LinkedBlockingQueue;

  newCachedThreadPool将corePoolSize设置为0,将maximumPoolSize设置为Integer.MAX_VALUE,使用的SynchronousQueue,也就是说来了任务就创建线程运行,当线程空闲超过60秒,就销毁线程。

  实际中,如果Executors提供的三个静态方法能满足要求,就尽量使用它提供的三个方法,因为自己去手动配置ThreadPoolExecutor的参数有点麻烦,要根据实际任务的类型和数量来进行配置。

  另外,如果ThreadPoolExecutor达不到要求,可以自己继承ThreadPoolExecutor类进行重写。

没有办法得知被执行的 Runnable 的执行结果。如果有需要的话你得使用一个
Callable。

四、如何合理配置线程池的大小

  本节来讨论一个比较重要的话题:如何合理配置线程池大小,仅供参考。

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

  如果是CPU密集型任务,就需要尽量压榨CPU,参考值可以设为
NCPU+1

  如果是IO密集型任务,参考值可以设置为2*NCPU

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

  参考资料:

  

  

  

  

  

  

  《JDK API 1.6》

submit 方法也要求一个 Runnable 实现类,但它返回一个 Future 对象。这个
Future 对象可以用来检查 Runnable 是否已经执行完毕。以下是
ExecutorService submit() 示例:

Future future = executorService.submit(new Runnable() { public void run() { System.out.println("Asynchronous task"); } }); future.get(); //returns null if the task has finished correctly. 

submit 方法类似于 submit 方法,除了它所要求的参数类型之外。Callable
实例除了它的 call() 方法能够返回一个结果之外和一个 Runnable
很相像。Runnable.run() 不能够返回一个结果。Callable 的结果可以通过
submit 方法返回的 Future 对象进行获取。

以下是一个 ExecutorService Callable 示例:

Future future = executorService.submit(new Callable(){ public Object call() throws Exception { System.out.println("Asynchronous Callable"); return "Callable Result"; } }); System.out.println("future.get() = " + future.get; 

以上代码输出:Asynchronous Callablefuture.get() = Callable Result

invokeAny() 方法要求一系列的 Callable
或者其子接口的实例对象。调用这个方法并不会返回一个
Future,但它返回其中一个 Callable 对象的结果。无法保证返回的是哪个
Callable 的结果 – 只能表明其中一个已执行结束。

如果其中一个任务执行结束,其他 Callable 将被取消。以下是示例代码:

ExecutorService executorService = Executors.newSingleThreadExecutor(); Set<Callable<String>> callables = new HashSet<Callable<String>>(); callables.add(new Callable<String>() { public String call() throws Exception { return "Task 1"; } }); callables.add(new Callable<String>() { public String call() throws Exception { return "Task 2"; } }); callables.add(new Callable<String>() { public String call() throws Exception { return "Task 3"; } }); String result = executorService.invokeAny(callables); System.out.println("result = " + result); executorService.shutdown(); 

上述代码将会打印出给定 Callable
集合中的一个的执行结果。我自己试着执行了它几次,结果始终在变。有时是
“Task 1″,有时是 “Task 2” 等等。

invokeAll() 方法将调用你在集合中传给 ExecutorService 的所有 Callable
对象。invokeAll() 返回一系列的 Future 对象,通过它们你可以获取每个
Callable
的执行结果。记住,一个任务可能会由于一个异常而结束,因此它可能没有
“成功”。

无法通过一个 Future
对象来告知我们是两种结束中的哪一种。以下是一个代码示例:

ExecutorService executorService = Executors.newSingleThreadExecutor(); Set<Callable<String>> callables = new HashSet<Callable<String>>(); callables.add(new Callable<String>() { public String call() throws Exception { return "Task 1"; } }); callables.add(new Callable<String>() { public String call() throws Exception { return "Task 2"; } }); callables.add(new Callable<String>() { public String call() throws Exception { return "Task 3"; } }); List<Future<String>> futures = executorService.invokeAll(callables); for(Future<String> future : futures){ System.out.println("future.get = " + future.get; } executorService.shutdown(); 

使用完 ExecutorService 之后你应该将其关闭,以使其中的线程不再运行。

比如,如果你的应用是通过一个 main() 方法启动的,之后 main
方法退出了你的应用,如果你的应用有一个活动的 ExexutorService
它将还会保持运行。ExecutorService 里的活动线程阻止了 JVM 的关闭。

要终止 ExecutorService 里的线程你需要调用 ExecutorService 的 shutdown()
方法。ExecutorService
并不会立即关闭,但它将不再接受新的任务,而且一旦所有线程都完成了当前任务的时候,ExecutorService
将会关闭。在 shutdown() 被调用之前所有提交给 ExecutorService
的任务都被执行。如果你想要立即关闭 ExecutorService,你可以调用
shutdownNow()
方法。这样会立即尝试停止所有执行中的任务,并忽略掉那些已提交但尚未开始处理的任务。无法担保执行任务的正确执行。可能它们被停止了,也可能已经执行结束。

java.util.concurrent.ThreadPoolExecutor 是 ExecutorService
接口的一个实现。ThreadPoolExecutor
使用其内部池中的线程执行给定任务(Callable 或者 Runnable)。

ThreadPoolExecutor
包含的线程池能够包含不同数量的线程。池中线程的数量由以下变量决定:

  • corePoolSize
  • maximumPoolSize

当一个任务委托给线程池时,如果池中线程数量低于
corePoolSize,一个新的线程将被创建,即使池中可能尚有空闲线程。如果内部任务队列已满,而且有至少
corePoolSize 正在运行,但是运行线程的数量低于
maximumPoolSize,一个新的线程将被创建去执行该任务。

ThreadPoolExecutor 图解:

奥门威尼斯网址 13image.png

** 一个 ThreadPoolExecutor **

ThreadPoolExecutor 有若干个可用构造子。比如:

int corePoolSize = 5; int maxPoolSize = 10; long keepAliveTime = 5000; ExecutorService threadPoolExecutor = new ThreadPoolExecutor( corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>; 

但是,除非你确实需要显式为 ThreadPoolExecutor 定义所有参数,使用
java.util.concurrent.Executors 类中的工厂方法之一会更加方便,正如
ExecutorService 小节所述。

java.util.concurrent.ScheduledExecutorService 是一个 ExecutorService,
它能够将任务延后执行,或者间隔固定时间多次执行。
任务由一个工作者线程异步执行,而不是由提交任务给
ScheduledExecutorService 的那个线程执行。

以下是一个简单的 ScheduledExecutorService 示例:

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool; ScheduledFuture scheduledFuture = scheduledExecutorService.schedule(new Callable() { public Object call() throws Exception { System.out.println("Executed!"); return "Called!"; } }, 5, TimeUnit.SECONDS); 

首先一个内置 5 个线程的 ScheduledExecutorService 被创建。之后一个
Callable 接口的匿名类示例被创建然后传递给 schedule()
方法。后边的俩参数定义了 Callable 将在 5 秒钟之后被执行。

既然 ScheduledExecutorService 是一个接口,你要用它的话就得使用
java.util.concurrent 包里对它的某个实现类。ScheduledExecutorService
具有以下实现类:ScheduledThreadPoolExecutor

创建一个 ScheduledExecutorService如何创建一个 ScheduledExecutorService
取决于你采用的它的实现类。但是你也可以使用 Executors 工厂类来创建一个
ScheduledExecutorService 实例。比如:

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool; 

一旦你创建了一个 ScheduledExecutorService,你可以通过调用它的以下方法:

  • schedule (Callable task, long delay, TimeUnit timeunit)
  • schedule (Runnable task, long delay, TimeUnit timeunit)
  • scheduleAtFixedRate (Runnable, long initialDelay, long period,
    TimeUnit timeunit)
  • scheduleWithFixedDelay (Runnable, long initialDelay, long period,
    TimeUnit timeunit)

下面我们就简单看一下这些方法。

这个方法计划指定的 Callable 在给定的延迟之后执行。这个方法返回一个
ScheduledFuture,通过它你可以在它被执行之前对它进行取消,或者在它执行之后获取结果。以下是一个示例:

ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool; ScheduledFuture scheduledFuture = scheduledExecutorService.schedule(new Callable() { public Object call() throws Exception { System.out.println("Executed!"); return "Called!"; } }, 5, TimeUnit.SECONDS); System.out.println("result = " + scheduledFuture.get; scheduledExecutorService.shutdown(); 

示例输出结果:Executed!result = Called!

除了 Runnable 无法返回一个结果之外,这一方法工作起来就像以一个 Callable
作为一个参数的那个版本的方法一样,因此 ScheduledFuture.get()
在任务执行结束之后返回 null。

这一方法规划一个任务将被定期执行。该任务将会在首个 initialDelay
之后得到执行,然后每个 period
时间之后重复执行。如果给定任务的执行抛出了异常,该任务将不再执行。如果没有任何异常的话,这个任务将会持续循环执行到
ScheduledExecutorService
被关闭。如果一个任务占用了比计划的时间间隔更长的时候,下一次执行将在当前执行结束执行才开始。计划任务在同一时间不会有多个线程同时执行。

除了 period 有不同的解释之外这个方法和 scheduleAtFixedRate() 非常像。

scheduleAtFixedRate() 方法中,period
被解释为前一个执行的开始和下一个执行的开始之间的间隔时间。而在本方法中,period
则被解释为前一个执行的结束和下一个执行的结束之间的间隔。因此这个延迟是执行结束之间的间隔,而不是执行开始之间的间隔。

正如 ExecutorService,在你使用结束之后你需要把 ScheduledExecutorService
关闭掉。否则他将导致 JVM 继续运行,即使所有其他线程已经全被关闭。

你可以使用从 ExecutorService 接口继承来的 shutdown() 或 shutdownNow()
方法将 ScheduledExecutorService 关闭。参见 ExecutorService
关闭部分以获取更多信息。

ForkJoinPool 在 Java 7 中被引入。它和 ExecutorService
很相似,除了一点不同。ForkJoinPool
让我们可以很方便地把任务分裂成几个更小的任务,这些分裂出来的任务也将会提交给
ForkJoinPool。任务可以继续分割成更小的子任务,只要它还能分割。可能听起来有些抽象,因此本节中我们将会解释
ForkJoinPool 是如何工作的,还有任务分割是如何进行的。

在我们开始看 ForkJoinPool
之前我们先来简要解释一下分叉和合并的原理。分叉和合并原理包含两个递归进行的步骤。两个步骤分别是分叉步骤和合并步骤。

一个使用了分叉和合并原理的任务可以将自己分叉为更小的子任务,这些子任务可以被并发执行。如下图所示:

奥门威尼斯网址 14image.png

通过把自己分割成多个子任务,每个子任务可以由不同的 CPU
并行执行,或者被同一个 CPU
上的不同线程执行。只有当给的任务过大,把它分割成几个子任务才有意义。把任务分割成子任务有一定开销,因此对于小型任务,这个分割的消耗可能比每个子任务并发执行的消耗还要大。

什么时候把一个任务分割成子任务是有意义的,这个界限也称作一个阀值。这要看每个任务对有意义阀值的决定。很大程度上取决于它要做的工作的种类。

当一个任务将自己分割成若干子任务之后,该任务将进入等待所有子任务的结束之中。一旦子任务执行结束,该任务可以把所有结果合并到同一个结果。图示如下:

奥门威尼斯网址 15image.png

当然,并非所有类型的任务都会返回一个结果。如果这个任务并不返回一个结果,它只需等待所有子任务执行完毕。也就不需要结果的合并啦。

ForkJoinPool 是一个特殊的线程池,它的设计是为了更好的配合 分叉-和-合并
任务分割的工作。ForkJoinPool 也在 java.util.concurrent
包中,其完整类名为 java.util.concurrent.ForkJoinPool。

你可以通过其构造子创建一个 ForkJoinPool。作为传递给 ForkJoinPool
构造子的一个参数,你可以定义你期望的并行级别。并行级别表示你想要传递给
ForkJoinPool 的任务所需的线程或 CPU 数量。以下是一个 ForkJoinPool 示例:

ForkJoinPool forkJoinPool = new ForkJoinPool; 

这个示例创建了一个并行级别为 4 的 ForkJoinPool。

就像提交任务到 ExecutorService 那样,把任务提交到
ForkJoinPool。你可以提交两种类型的任务。一种是没有任何返回值的,另一种是有返回值的。这两种类型分别由
RecursiveAction 和 RecursiveTask
表示。接下来介绍如何使用这两种类型的任务,以及如何对它们进行提交。

RecursiveAction
是一种没有任何返回值的任务。它只是做一些工作,比如写数据到磁盘,然后就退出了。一个
RecursiveAction
可以把自己的工作分割成更小的几块,这样它们可以由独立的线程或者 CPU
执行。你可以通过继承来实现一个 RecursiveAction。示例如下:

import java.util.ArrayList; import java.util.List; import java.util.concurrent.RecursiveAction; public class MyRecursiveAction extends RecursiveAction { private long workLoad = 0; public MyRecursiveAction(long workLoad) { this.workLoad = workLoad; } @Override protected void compute() { //if work is above threshold, break tasks up into smaller tasks if(this.workLoad > 16) { System.out.println("Splitting workLoad : " + this.workLoad); List<MyRecursiveAction> subtasks = new ArrayList<MyRecursiveAction>(); subtasks.addAll(createSubtasks; for(RecursiveAction subtask : subtasks){ subtask.fork(); } } else { System.out.println("Doing workLoad myself: " + this.workLoad); } } private List<MyRecursiveAction> createSubtasks() { List<MyRecursiveAction> subtasks = new ArrayList<MyRecursiveAction>(); MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / 2); MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / 2); subtasks.add; subtasks.add; return subtasks; } } 

例子很简单。MyRecursiveAction 将一个虚构的 workLoad
作为参数传给自己的构造子。如果 workLoad
高于一个特定阀值,该工作将被分割为几个子工作,子工作继续分割。如果
workLoad 低于特定阀值,该工作将由 MyRecursiveAction
自己执行。你可以这样规划一个 MyRecursiveAction 的执行:

MyRecursiveAction myRecursiveAction = new MyRecursiveAction; forkJoinPool.invoke(myRecursiveAction); 

RecursiveTask
是一种会返回结果的任务。它可以将自己的工作分割为若干更小任务,并将这些子任务的执行结果合并到一个集体结果。可以有几个水平的分割和合并。以下是一个
RecursiveTask 示例:

import java.util.ArrayList; import java.util.List; import java.util.concurrent.RecursiveTask; public class MyRecursiveTask extends RecursiveTask<Long> { private long workLoad = 0; public MyRecursiveTask(long workLoad) { this.workLoad = workLoad; } protected Long compute() { //if work is above threshold, break tasks up into smaller tasks if(this.workLoad > 16) { System.out.println("Splitting workLoad : " + this.workLoad); List<MyRecursiveTask> subtasks = new ArrayList<MyRecursiveTask>(); subtasks.addAll(createSubtasks; for(MyRecursiveTask subtask : subtasks){ subtask.fork(); } long result = 0; for(MyRecursiveTask subtask : subtasks) { result += subtask.join(); } return result; } else { System.out.println("Doing workLoad myself: " + this.workLoad); return workLoad * 3; } } private List<MyRecursiveTask> createSubtasks() { List<MyRecursiveTask> subtasks = new ArrayList<MyRecursiveTask>(); MyRecursiveTask subtask1 = new MyRecursiveTask(this.workLoad / 2); MyRecursiveTask subtask2 = new MyRecursiveTask(this.workLoad / 2); subtasks.add; subtasks.add; return subtasks; } } 

除了有一个结果返回之外,这个示例和 RecursiveAction
的例子很像。MyRecursiveTask 类继承自
RecursiveTask<Long>,这也就意味着它将返回一个 Long 类型的结果。

MyRecursiveTask 示例也会将工作分割为子任务,并通过 fork()
方法对这些子任务计划执行。

此外,本示例还通过调用每个子任务的 join()
方法收集它们返回的结果。子任务的结果随后被合并到一个更大的结果,并最终将其返回。对于不同级别的递归,这种子任务的结果合并可能会发生递归。

你可以这样规划一个 RecursiveTask:

MyRecursiveTask myRecursiveTask = new MyRecursiveTask; long mergedResult = forkJoinPool.invoke(myRecursiveTask); System.out.println("mergedResult = " + mergedResult); 

注意是如何通过 ForkJoinPool.invoke() 方法的调用来获取最终执行结果的。

貌似并非每个人都对 Java 7 里的 ForkJoinPool 满意:《一个 Java 分叉-合并
带来的灾祸》。

在你计划在自己的项目里使用 ForkJoinPool 之前最好读一下该篇文章。

个人介绍:

** 高广超**
:多年一线互联网研发与架构设计经验,擅长设计与落地高可用、高性能互联网架构。目前就职于美团网,负责核心业务研发工作。

本文首发在 高广超的简书博客 转载请注明!

奥门威尼斯网址 16image.png

发表评论

电子邮件地址不会被公开。 必填项已用*标注