java并发之线程池Executor知识要点 及 核心源码浅析

时间：2021-04-14 18:33:22

1.什么是线程池

  • 定义：线程池是指管理一组同构工作线程的资源池
  • 组成部分：
    • 线程管理器（ThreadPool）：用于创建并管理线程池。包括创建线程池，销毁线程池，添加新任务
    • 工作线程（PoolWorker）：线程池中的线程
    • 任务接口（Task）:每个任务必须实现的接口，一共工作线程调度任务的执行
    • 任务队列：用于存放没有处理的任务，提供一种缓冲机制

2.为什么要使用线程池

  • 通过重用现有的线程而不是创建新线程，从而减少了线程创建 和 销毁过程中的巨大开销
  • 当请求到达时，工作线程已经存在，不用再等待线程的创建，从而提高了响应性
  • 可以是处理器保持忙碌状态,提高系统的吞吐量

3.什么时候使用线程池

当一个服务器完成一项任务所需时间为：T1 创建线程的时间，T2 在线程中执行任务的时间，T3 销毁线程时间。若 T1+T3 > T2 时，则一般可以采用线程池。

4.如何使用线程池

  • 线程池的类图
    java并发之线程池Executor知识要点 及 核心源码浅析java并发之线程池Executor知识要点 及 核心源码浅析
    • 其中，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，该策略会抛出 未检查的 RegectedExecutionExceptionCallerRunsPolicy：调用者运行，将某些任务回退到调用者（一般为主线程），由于执行任务需要一定的时间，因此主线程至少在一段时间内不能提交任何任务，从而使得工作者线程有时间来处理正在执行的任务DiscardPolicy：抛弃新提交的任务DiscardOldestPolicy：将抛弃下一个将要执行的任务
  • 关于线程池的执行过程
    • 从execute 方法为 入口点
      /**1.当线程池中的数目小于 corePoolSize时，直接new 一个 Thread2.当线程池数大于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
标签：

相关文章