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 方法,将取消所有运行中的任务,返回还未开始的任务,但不不会返回正在执行的任务

  • 创建线程池

    1. 通过调用 Execotors 中的静态工厂方法

      • newFixedThreadPool:创建一个固定长度的线程池,每当提交一个任务时就创建一个线程,知道达到线程池的最大数量
      • newCachedThreadPool:创建一个可缓存的线程池,线程池的规模不存在任何限制。有多少任务就会有多少线程。当超过了处理需求时,会回收空闲的线程
      • newSingleThreadExecutor:创建一个单线程的 Executor,依照任务在队列中的顺序来串行执行,安全的封闭在线程中阿
      • newScheduledThreadPool: 创建一个固定长度的线程池,而且一延迟或定时的方式来执行任务
    2. 通过 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;

  }

    }

}

参考资料:

http://ifeve.com/java-threadpool/

  http://blog.163.com/among_1985/blog/static/275005232012618849266/

http://developer.51cto.com/art/201203/321885.htm

http://www.51itong.net/java-1-7-threadpoolexecutor-19428.html

http://blog.csdn.net/java2000_wl/article/details/22097059

http://blog.csdn.net/xieyuooo/article/details/8718741

时间: 2024-10-12 19:22:44

java并发之线程池Executor 核心源码解析的相关文章

Java并发之——线程池

一. 线程池介绍 1.1 简介 线程池是一种多线程处理形式,处理过程中将任务添加到队列,然后在创建线程后自动启动这些任务.线程池的基本思想还是一种对象池的思想,开辟一块内存空间,里面存放了众多(未死亡)的线程,池中线程执行调度由池管理器来处理.当有线程任务时,从池中取一个,执行完成后线程对象归池,这样可以避免反复创建线程对象所带来的性能开销,节省了系统的资源. 多线程技术主要解决处理器单元内多个线程执行的问题,它可以显著减少处理器单元的闲置时间,增加处理器单元的吞吐能力. 假设一个服务器完成一项

周期性线程池与主要源码解析

之前学习ThreadPool的使用以及源码剖析,并且从面试的角度去介绍知识点的解答.今天小强带来周期性线程池的使用和重点源码剖析. ScheduledThreadPoolExecutor ScheduledThreadPoolExecutor:用来处理延时任务或定时任务 定时线程池类的类结构图 ScheduledThreadPoolExecutor接收ScheduleFutureTask类型的任务,是线程池调度任务的最小单位. 它采用DelayQueue存储等待的任务: 1.DelayQueue

线程池之ThreadPoolExecutor源码解析

1.变量 ThreadPoolExecutor先定义了这几个常量,初看时一脸懵逼,其实它就是用int的二进制高三位来表示线程池的状态, 先回顾一下位运算: <<’左移:右边空出的位置补0,其值相当于乘以2. ‘>>’右移:左边空出的位,如果是正数则补0,若为负数则补0或1,取决于所用的计算机系统OS X中补1.其值相当于除以2. 负数二进制由它的绝对值取反后加1 private final AtomicInteger ctl = new AtomicInteger(ctlOf(RU

Zuul的核心源码解析

在 Zuul中, 整个请求的过程是这样的,首先将请求给zuulservlet处理,zuulservlet中有一个zuulRunner对象,该对象中初始化了RequestContext:作为存储整个请求的一些数据,并被所有的zuulfilter共享.zuulRunner中还有 FilterProcessor,FilterProcessor作为执行所有的zuulfilter的管理器.FilterProcessor从filterloader 中获取zuulfilter,而zuulfilter是被fil

spark内核揭秘-05-SparkContext核心源码解析初体验

SparkContext在获得了一系列的初始化信息后开始创建并启动TaskScheduler实例: 进入createTaskScheduler方法: 我们看一下其Standalone的方式: 在上述代码中首先实例化一个TaskSchedulerImpl: 然后构建出了masterUrls: 接着创建出关键的backend: 进入SparkDeploySchedulerBackend实现: 从以上截图可以看出来,SparkDeploySchedulerBackend核心是为了启动CoarseGra

java线程池和五种常用线程池的策略使用与解析

java线程池和五种常用线程池策略使用与解析 一.线程池 关于为什么要使用线程池久不赘述了,首先看一下java中作为线程池Executor底层实现类的ThredPoolExecutor的构造函数 public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory th

深入浅出 Java Concurrency (30): 线程池 part 3 Executor 生命周期[转]

我们知道线程是有多种执行状态的,同样管理线程的线程池也有多种状态.JVM会在所有线程(非后台daemon线程)全部终止后才退出,为了节省资源和有效释放资源关闭一个线程池就显得很重要.有时候无法正确的关闭线程池,将会阻止JVM的结束. 线程池Executor是异步的执行任务,因此任何时刻不能够直接获取提交的任务的状态.这些任务有可能已经完成,也有可能正在执行或者还在排队等待执行.因此关闭线程池可能出现一下几种情况: 平缓关闭:已经启动的任务全部执行完毕,同时不再接受新的任务 立即关闭:取消所有正在

Java并发——线程池Executor框架

线程池 无限制的创建线程 若采用"为每个任务分配一个线程"的方式会存在一些缺陷,尤其是当需要创建大量线程时: 线程生命周期的开销非常高 资源消耗 稳定性 引入线程池 任务是一组逻辑工作单元,线程则是使任务异步执行的机制.当存在大量并发任务时,创建.销毁线程需要很大的开销,运用线程池可以大大减小开销. Executor框架 说明: Executor 执行器接口,该接口定义执行Runnable任务的方式. ExecutorService 该接口定义提供对Executor的服务. Sched

Java四种线程池newCachedThreadPool,newFixedThreadPool,newScheduledThreadPool,newSingleThreadExecutor

介绍new Thread的弊端及Java四种线程池的使用,对Android同样适用.本文是基础篇,后面会分享下线程池一些高级功能. 1.new Thread的弊端 执行一个异步任务你还只是如下new Thread吗? Java new Thread(new Runnable() { @Override public void run() { // TODO Auto-generated method stub } }).start(); 1 2 3 4 5 6 7 new Thread(new