疯狂java


您现在的位置: 疯狂软件 >> 新闻资讯 >> 正文

Java并发之线程池Executor详解


 

 
 
1.什么是线程池
定义:线程池是指管理一组同构工作线程的资源池
组成部分:
线程管理器(ThreadPool):用于创建并管理线程池。包括创建线程池,销毁线程池,添加新任务
工作线程(PoolWorker):线程池中的线程
任务接口(Task):每个任务必须实现的接口,一共工作线程调度任务的执行
任务队列:用于存放没有处理的任务,提供一种缓冲机制
2.为什么要使用线程池
通过重用现有的线程而不是创建新线程,从而减少了线程创建 和 销毁过程中的巨大开销
当请求到达时,工作线程已经存在,不用再等待线程的创建,从而提高了响应性
可以是处理器保持忙碌状态,提高系统的吞吐量
3.什么时候使用线程池
当一个服务器完成一项任务所需时间为:T1 创建线程的时间,T2 在线程中执行任务的时间,T3 销毁线程时间。若 T1+T3 > T2 时,则一般可以采用线程池。
4.如何使用线程池
线程池的类图
 
其中,Executor 接口中只有一个execute方法
ExecutorService 接口继承于 Executor接口,它提供了如下方法:
shutdown();
shutdownNow();
submit();
invokeAll();
invokeAny();
AbstractExecutorService 抽象类,实现了 ExecutorService 中的部分方法
ThreadPoolExecutor 类,实现了其上级的所有未实现的方法,还提供许多其他方法。其中,由它实现的父级方法有如下:
execute();
shutdown();
shutdownNow();
isTerminated();
其中,需要说明的是 submit 与 execute 方法 区别,submit 方法是在 ExecutorService 中声明的方法,在AbstractExecutorService中的具体实现,而 execute 方法是在 Executor 中声明的方法,在 ThreadPoolExecutor 进行了具体的实现。它们都是向线程池中提交任务,但是不同的是 submit 方法 会返回任务的执行结果,但实际上,它还是调用的 execute方法,只不过它利用了Future来获取任务的执行结果。
shutdown 方法与 shutdownNow 方法的区别在于:shutdown方法将平缓的关闭过程,它不会再接受新的任务,通知等待已提交的任务执行完成,包括那些还未完成的任务。 而shutdownNow 方法,将取消所有运行中的任务,返回还未开始的任务,但不不会返回正在执行的任务
创建线程池
通过调用 Execotors 中的静态工厂方法
newFixedThreadPool:创建一个固定长度的线程池,每当提交一个任务时就创建一个线程,知道达到线程池的最大数量
newCachedThreadPool:创建一个可缓存的线程池,线程池的规模不存在任何限制。有多少任务就会有多少线程。当超过了处理需求时,会回收空闲的线程
newSingleThreadExecutor:创建一个单线程的 Executor,依照任务在队列中的顺序来串行执行,安全的封闭在线程中阿
newScheduledThreadPool: 创建一个固定长度的线程池,而且一延迟或定时的方式来执行任务
通过 ThreadPoolExecutor 的构造方法进行创建
public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,  TimeUnit unit, BlockingQueue<Runnable> workQueue,ThreadFactory threadFactory,RejectedExecutionHandler handler)
corePoolSize:线程池的大小,当创建了线程池后,线程池中没有任何线程,而是等待任务的到来才会创建线程去执行任务。当线程数达到了corePoolSize时,就会把到达的任务放在缓存队列当中
maximumPoolSize:线程池中的最大线程数
keepAliveTime: 表示线程没有任务执行时最多保持多久时间会终止。默认情况下,只有当线程池中的线程数大于corePoolSize时,keepAliveTime才会起作用,直到线程池中的线程数不大于corePoolSize,即当线程池中的线程数大于corePoolSize时,如果一个线程空闲的时间达到keepAliveTime,则会终止,直到线程池中的线程数不超过corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法,在线程池中的线程数不大于corePoolSize时,keepAliveTime参数也会起作用,直到线程池中的线程数为0;
unit : 参数 keepAliveTime 的时间单位
workQueue : 用来存储等待执行的任务
无界队列:LinkedBlockingQueue-可无限制的增加,它是 newFixedThreadPool 和 newSingleThreadExecutor 默认的任务队列
有界队列:ArrayBlockingQueue、有界的 LinkedBlockingQueue、PriorityBlockingQueue-当队列满后,会执行饱和策略
同步移交:在线程之间移交的一种机制,当线程池是无界的或者是可以拒绝任务的就可以使用,newCachedThreadPool就使用的是 SynchronousQueue
threadFactory:线程工厂,主要用来创建线程,可以自定义线程工厂(实现 ThreadFactory接口),可以定制一些行为:如统计信息以及线程被创建或者终止时把调试消息写入日志
handler : 饱和策略— 当任务队列被填满后,会执行的策略。有如下:
ThredPoolExecutor 类中的4个静态内部类:
AbortPolicy: 默认的 Handler,该策略会抛出 未检查的 RegectedExecutionException
CallerRunsPolicy:调用者运行,将某些任务回退到调用者(一般为主线程),由于执行任务需要一定的时间,因此主线程至少在一段时间内不能提交任何任务,从而使得工作者线程有时间来处理正在执行的任务
DiscardPolicy:抛弃新提交的任务
DiscardOldestPolicy:将抛弃下一个将要执行的任务
关于线程池的执行过程
从execute 方法为 入口点
/**
1.当线程池中的数目小于 corePoolSize时,直接new 一个 Thread
2.当线程池数大于corePoolSize 时,直接放入 任务队列中
3.如果队列已经满了且线程池中线程数小于 maximumPoolSize,则新建一个线程
*/
public void execute(Runnable command) {
   if (command == null)
      throw new NullPointerException();
   int c = ctl.get();
  //如果正在运行的线程数小于线程池预设的大小,就尝试addWorker(源码在其后)。若成功,直接返回。若添加失败(可能在添加过程中已达到预设的线程池的数目),重新获取线程池正在运行的线程数
   if (workerCountOf(c) < corePoolSize) {
       if (addWorker(command, true))
       return;
   c = ctl.get();
  }
  //若任务没有线程处理(当达到了线程池的预设大小 corePoolSize),就添加到任务队列。
  if (isRunning(c) && workQueue.offer(command)) {
     int recheck = ctl.get();
     if (! isRunning(recheck) && remove(command)) //若线程池在workeQueue.offer前发生了shutdown,就从任务队列中移除
        reject(command);
    else if (workerCountOf(recheck) == 0) //如果线程池在运行,并且没有可工作的线程,就直接创建一个
       addWorker(null, false);
 }
 else if (!addWorker(command, false)) // 如果任务队列已满,尝试创建一个新的Worker,若失败,说明线程池已经关闭 或者 饱和了
   reject(command); //饱和策略问题
}
  /**
    @param firstTask : 是新线程应该运行的第一个任务,Worker 会被创建 在线程池的数目 比 corePoolSize 小的时候,或者在任务队列已经满的情况下,创建一个线程来代替已死的线程
    @param core : 当为true时,用 corePoolSize作为边界,否则,用 maximumPoolSize 作为边界
  */
    private boolean addWorker(Runnable firstTask, boolean core) {
    retry:
    for (;;) {
        int c = ctl.get(); 
        int rs = runStateOf(c); //获取线程池的状态
 
    // 如果线程池被 shutdown了,一般直接返回false。但是排除 任务队列不为空 但 Workers 为空的情况,在这种情况下,会调用 addWorker(null,false) 来创建一个线程处理队列中的任务
    if (rs >= SHUTDOWN &&
            ! (rs == SHUTDOWN &&
               firstTask == null &&
               ! workQueue.isEmpty()))
            return false;
 
    for (;;) {//如果正在执行的线程数目大于 线程池中预设的线程数,返回false
            int wc = workerCountOf(c);
            if (wc >= CAPACITY ||  wc >= (core ? corePoolSize : maximumPoolSize))
                return false;
            if (compareAndIncrementWorkerCount(c))
                break retry;
            c = ctl.get(); // Re-read ctl
           if (runStateOf(c) != rs)
                continue retry;
      // else CAS failed due to workerCount change; retry inner loop
       }
    }
 
    boolean workerStarted = false;
    boolean workerAdded = false;
    Worker w = null;
    try {
        final ReentrantLock mainLock = this.mainLock;
        w = new Worker(firstTask);//创建 Worker:其中Thread通过调用ThreadFactory 的 newThread 方法构建,所以在此处可以对创建的Thread进行额外的处理
       final Thread t = w.thread;
       if (t != null) {
            mainLock.lock();
            try {
                int c = ctl.get();
                int rs = runStateOf(c);
               //若线程池正在运行 或者 处于 shutdown 但是任务不为空,则把新建的worker添加在workers 中
               if (rs < SHUTDOWN ||  (rs == SHUTDOWN && firstTask == null)) {
                    if (t.isAlive()) // precheck that t is startable
                       throw new IllegalThreadStateException();
                    workers.add(w);
                    int s = workers.size(); //largePoolSize 用于记录曾经出现过得最大的线程数
                    if (s > largestPoolSize)
                        largestPoolSize = s;
                   workerAdded = true;
                }
            } finally {
                mainLock.unlock();
            }
            if (workerAdded) {//启动任务的执行
                t.start();
                workerStarted = true;
            }
        }
    } finally {
        if (! workerStarted)
            addWorkerFailed(w);//若启动失败,则从正在运行的工作集中移除
  }
    return workerStarted;
}
  /**
  当 线程池 被 stop 或者 shutdown 或创建线程失败时,则会调用这个方法
  1.从 workers 中移除 worker
  2.把workerCount -1
  3.尝试终止操作:当 线程池的状态为 shutdown 、线程池的数目 和 任务队列都为空,或者线程池已经 stop 、线程池数目为0
  */
private void addWorkerFailed(Worker w) {
    final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
 try {
        if (w != null)
            workers.remove(w);
  decrementWorkerCount();
  tryTerminate();
  } finally {
        mainLock.unlock();
  }
}
final void tryTerminate() {
    for (;;) {
        int c = ctl.get();
 if (isRunning(c) ||
            runStateAtLeast(c, TIDYING) ||
            (runStateOf(c) == SHUTDOWN && ! workQueue.isEmpty()))
            return;
 if (workerCountOf(c) != 0) { // Eligible to terminate
  interruptIdleWorkers(ONLY_ONE);
 return;  }
 
        final ReentrantLock mainLock = this.mainLock;
  mainLock.lock();
 try {
            if (ctl.compareAndSet(c, ctlOf(TIDYING, 0))) {
                try {
                    terminated();
  } finally {
                    ctl.set(ctlOf(TERMINATED, 0));
  termination.signalAll();
  }
                return;
  }
        } finally {
            mainLock.unlock();
  }
        // else retry on failed CAS
  }
}
  个人觉得还比较重要的一个内部类是 Worker,继承了AbstractQueuedSynchronizer 类,其中最主要的 run 方法
private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{
  Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
  this.firstTask = firstTask;
 this.thread = getThreadFactory().newThread(this);
  }
 
 
  public void run() {
        runWorker(this);
  }
 
  protected boolean isHeldExclusively() {
        return getState() != 0;
  }
 
    protected boolean tryAcquire(int unused) {
        if (compareAndSetState(0, 1)) {
            setExclusiveOwnerThread(Thread.currentThread());
 return true;  }
        return false;
  }
 
    protected boolean tryRelease(int unused) {
        setExclusiveOwnerThread(null);
  setState(0);
 return true;  }
    // 参数 1 就是把锁设置为了独占锁,在获取到一个任务后,准备执行前首先要获取这个锁。
    public void lock()        { acquire(1); }
    public boolean tryLock()  { return tryAcquire(1); }
    public void unlock()      { release(1); }
    public boolean isLocked() { return isHeldExclusively(); }
 
    void interruptIfStarted() {
        Thread t;
 if (getState() >= 0 && (t = thread) != null && !t.isInterrupted()) {
            try {
                t.interrupt();
  } catch (SecurityException ignore) {
            }
        }
    }
}
  final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    // 不从 任务队列 中获取第一个任务,而是执行刚提交的任务
  Runnable task = w.firstTask;
  w.firstTask = null;
  w.unlock(); // allow interrupts
  boolean completedAbruptly = true;
 try {
        while (task != null || (task = getTask()) != null) {
            w.lock();
  // 如果线程池是 STOP 状态,则需要保证线程时被中断了
 // if not, ensure thread is not interrupted.  This // requires a recheck in second case to deal with // shutdownNow race while clearing interrupt  
 if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
    try {
                beforeExecute(wt, task); // 执行前的勾子函数,可以通过重写的方式对这个函数进行扩展功能。当发生了异常时,是不会执行任务的,afterExecute 也不会执行
                Throwable thrown = null;
                try {
                    task.run(); // 执行任务
               } catch (RuntimeException x) { //当发生运行时异常 或 ERROR 时,会原样抛出
                    thrown = x; throw x;
               } catch (Error x) { 
                    thrown = x; throw x;
               } catch (Throwable x) { // 如果是一个 Throwable ,则会包装成一个 Error 抛出
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown); // 执行后的勾子函数,同样可以进行扩展
                }
    } finally {
       task = null;
       w.completedTasks++;
       w.unlock();
      }
   }
   completedAbruptly = false;
} finally {
        processWorkerExit(w, completedAbruptly);
  }
}
  ~
  /**
  从任务队列中获取一个任务,在以下情况会退出:
  1.线程数 大于 maximumPoolSize,我也不知道为什么会大于??创建的时候就会判断啊??
  2.线程池 已经停止
  3.线程池 shutdown 并且 队列 也为空
  4.获取一个task,超时
 
  @return : 返回一个 task ,或者worker 退出,workerCount -1 
  */
private Runnable getTask() {
    boolean timedOut = false; // Did the last poll() time out?
 
  retry:
    for (;;) {
        int c = ctl.get();
 int rs = runStateOf(c);
 
  // 如果当前线程池已经 shutdown 或者 stop && 任务队列为空,则workerCount - 1 
  if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
            decrementWorkerCount();
 return null;  }
 
        boolean timed; // Are workers subject to culling?
 
  for (;;) {
            int wc = workerCountOf(c);
  timed = allowCoreThreadTimeOut || wc > corePoolSize;
 
 if (wc <= maximumPoolSize && ! (timedOut && timed)) //若没有超时
                break;
 if (compareAndDecrementWorkerCount(c))
                return null;
  c = ctl.get(); // Re-read ctl
  if (runStateOf(c) != rs)
                continue retry;
  // else CAS failed due to workerCount change; retry inner loop
  }
 
        try {
            Runnable r = timed ?
                workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                workQueue.take();
 if (r != null)
                return r;
  timedOut = true;  //若没有获取到 task ,超时
  } catch (InterruptedException retry) { // 若被中断了,不能算作超时
            timedOut = false;
  }
    }
}