Java 线程池详解

为什么需要线程池

线程的创建是有开销的，如果每次请求或执行一个新任务都新开一个线程的话，服务器将在频繁的创建和销毁线程上花费很多的时间，并消耗很多的系统资源。如果只是创建3、5个线程，其看不出对系统有什么影响，但是如果是几千上万的创建线程必定导致系统性能下降，因此为了保证系统的性能，很多时候需要限制创建线程的数量。

Java 中的线程池

线程池代表一组正在等待的作业并可多次重复使用的工作线程，Java中一般预初始化一组线程，组的大小是固定的，然后当需要执行任务的时候，将任务提交到线程池的中，线程池就分配已经预初始化的线程来执行任务，不需要每个任务都重新创建新线程来执行，可以重复利用线程，避免频繁创建线程和销毁线程大量的消耗系统资源。

自从jdk1.5以后，Java提供了Executor框架，核心的类ThreadPoolExecutor，只需实现Runnable接口然后将它们丢给ThreadPoolExecutor，ThreadPoolExecutor负责执行任务

ThreadPoolExecutor使用线程池提高了性能，当将任务发送给ThreadPoolExecutor时，它会尝试使用池线程来执行此任务，以避免连续产生新的线程。该框架模式如下图

Executor框架.png

怎么创建ThreadPoolExecutor

构造函数

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

参数解析

corePoolSize 核心线程数，线程池中最小活动的线程数，默认的核心线程是不会因为timeout被回收，除非allowCoreThreadTimeOut设置为true,默认为false

maximumPoolSize 线程池中的最大线程数，当核心线忙不过来的时候，且任务队列中有任务要执行的时候，就会再启动新的线程来执行，但是总的线程数不能超出maximumPoolSize的配置

keepAliveTime 线程保活时间，当线程池中线程数超出核心线程数，那么多出的线程的如果空闲时间超出该保活时间，那么线程将被回收

unit 保活时间的单位

workQueue 任务队列，当任务数超过了可运行的线程数，那么后来接收到的任务将被放入到queue中排队等待空闲线程执行

threadFactory 用于创建线程，默认的实现为DefaultThreadFactory

handler 拒绝策略，当任务数比配置的线程数多的时候，要如何处理取决拒绝策略，默认采取AbortPolicy，线程池直接拒绝，抛出异常RejectedExecutionException

线程池原理解析

  • 线程池初始化后，默认的核心线程数为0
  • 当把任务丢给线程池的时候，如果在运行的核心线程少于配置的核心线程数，即使有空闲的线程，也会启动一个新线程来执行任务
  • 如果已启动的线程数大于或等于核心线程数，会将任务放进workQueue中
  • 如果任务队列workQueue满了，且线程数小于maximumPoolSize，则会创建新的线程来执行任务
  • 如果任务队列workQueue满了，且线程数据大于或等于maximumPoolSize，则会执行拒绝策略

任务提交过程源码解析

public class ThreadPoolExecutor extends AbstractExecutorService{

    //...上面省略了N行代码

    public void execute(Runnable command) {
        if (command == null)
            throw new NullPointerException();
        int c = ctl.get();
        //如果worker线程数比核心线程数小，则直接创建新的worker执行任务
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
        //如果worker线程数超过核心线程数，任务放入任务队列
        if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            //这里需要再次检测线程池的状态，因为执行到这里的时候，可能线程池状态发生了变化
            //如果线程池不是运行状态，需要对新加入的任务执行reject操作
            if (! isRunning(recheck) && remove(command))
                reject(command);
            //判断worker线程数是否为0    
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
        //如果线程池没有运行或任务放入队列失败了，则再尝试创建新的workder执行任务，addWorker第二个参数为false表示创建非核心线程
        //如果addWorker返回false,则表示任务执行失败，将执行reject操作
        else if (!addWorker(command, false))
            reject(command);
    }

//...下面省略了N行代码

}

增加worker过程解析

public class ThreadPoolExecutor extends AbstractExecutorService{

    //...上面省略了N行代码

    /**
    * 增加工作线程
    * @param firstTask worker线程任务
    * @param core 表示是否是增加核心线程
    */
  private boolean addWorker(Runnable firstTask, boolean core) {
      retry:
      //外层循环
      for (;;) {
          int c = ctl.get();
          //取线程池状态
          int rs = runStateOf(c);
                    
          //如果线程池状态大于SHUTDOWN，则返回false，增加worker失败
          //如果线程池状态等于SHUTDOWN，且firstTask为null，则返回false，增加worker失败
          //如果线程池状态等于SHUTDOWN，且workQueue为空，则返回false，增加worker失败
          //符合上面任意一个条件，worker增加都失败
          if (rs >= SHUTDOWN &&
              ! (rs == SHUTDOWN &&
                 firstTask == null &&
                 ! workQueue.isEmpty()))
              return false;
                    //内层循环
          for (;;) {
                //线程数量
              int wc = workerCountOf(c);
              //如果线程数量超出了CAPACITY容量，则返回false，增加worker失败
              //core参数为true，表示增加核心线程，如果线程数据超出核心线程数，则返回false，增加worker失败
              //core参数为false，表示非核心线程，只要线程数据不超出最大线程，增加成功
              if (wc >= CAPACITY ||
                  wc >= (core ? corePoolSize : maximumPoolSize))
                  return false;
              //尝试增加worker线程，如果增加成功，跳出外层循环
              if (compareAndIncrementWorkerCount(c))
                  break retry;
              c = ctl.get();  // Re-read ctl
              //检测线程池状态，如果状态发生变化，则继续外层循环
              if (runStateOf(c) != rs)
                  continue retry;
              //如果其他情况，则继续尝试内层循环即可
          }
      }

      boolean workerStarted = false;
      boolean workerAdded = false;
      Worker w = null;
      try {
          w = new Worker(firstTask);
          final Thread t = w.thread;
          if (t != null) {
              final ReentrantLock mainLock = this.mainLock;
              //加锁
              mainLock.lock();
              try {
                  //加锁后，需要重新检测线程池状态
                  int rs = runStateOf(ctl.get());
                                    //如果线程池状态小于SHUTDOWN，则添加worker成功
                                    //如果线程池状态等于SHUTDOWN并且firstTask为null，则增加worker成功
                  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();
                      if (s > largestPoolSize)
                            //更新largestPoolSize
                          largestPoolSize = s;
                      workerAdded = true;
                  }
              } finally {
                  mainLock.unlock();
              }
              //如果worker添加成功，则启动worker线程
              if (workerAdded) {
                  t.start();
                  workerStarted = true;
              }
          }
      } finally {
            //如果worker启动失败，则需要执行addWorkerFailed失败操作
          if (! workerStarted)
              addWorkerFailed(w);
      }
      return workerStarted;
  }
  
  //...下面省略了N行代码
  
}

worker执行过程

public class ThreadPoolExecutor extends AbstractExecutorService{

    //...上面省略了N行代码

  final void runWorker(Worker w) {
      Thread wt = Thread.currentThread();
      Runnable task = w.firstTask;
      w.firstTask = null;
      w.unlock(); // allow interrupts
      boolean completedAbruptly = true;
      try {
            //先执行firstTask,如果有的话
            //如果firstTask不为空，则执行firstTask
            //如果firstTask为空，则调用getTask()从队列中获取任务来执行
            //因为这里的队列是阻塞队列，当队列为空的时候，线程将被阻塞，直到有任务返回继续执行
          while (task != null || (task = getTask()) != null) {
                //加锁，保证任务执行是串行的
              w.lock();
              //如果线程正在停止中，则需中断当前线程
              if ((runStateAtLeast(ctl.get(), STOP) ||
                   (Thread.interrupted() &&
                    runStateAtLeast(ctl.get(), STOP))) &&
                  !wt.isInterrupted())
                  wt.interrupt();
              try {
                    //任务执行前，可扩展功能
                  beforeExecute(wt, task);
                  Throwable thrown = null;
                  try {
                        //执行任务
                      task.run();
                  } catch (RuntimeException x) {
                      thrown = x; throw x;
                  } catch (Error x) {
                      thrown = x; throw x;
                  } catch (Throwable x) {
                      thrown = x; throw new Error(x);
                  } finally {
                        //任务执行后，可扩展功能
                      afterExecute(task, thrown);
                  }
              } finally {
                  task = null;
                  //完成任务数+1
                  w.completedTasks++;
                  w.unlock();
              }
          }
          completedAbruptly = false;
      } finally {
          processWorkerExit(w, completedAbruptly);
      }
  }
  
  //...下面省略了N行代码
  
}

Executors工具类

java.util.concurrent.Executors工具类提供了5种常见的创建方式

1、创建固定数据的线程池

创建一个固定数量线程的线程池，不设置队列大小，共享无界队列，任何时候最多有个n个活动线程可以处理任务，如果在所有线程都出于活动状态时提交了新的任务，那么新任务将在队列中等待，知道有空闲线程可用。如下创建了8个线程的线程池，其核心线程数和最大线程数都为8。

ExecutorService executor = Executors.newFixedThreadPool(8);

newFixedThreadPool的源码如下

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

⚠️ 该方法默认的核心线程和最大线程设置的一样，空闲超时为0ms，这个默认只针对非核心线程，任务队列使用的是LinkedBlockingQueue无界队列，这里有时候会踩坑，当任务量很大的时候，线程池忙不过来了，就对堆积到队列里，有可能会把内存撑爆，使用时需要注意，最好自己实例化ThreadPoolExecutor对象使用。

2、创建单个线程的线程池

创建使用只有一个工作线程的线程池，使用无界队列，只有一个线程执行任务，可以保证任务的顺序执行。如下创建方式，其核心线程等于最大线程，都为1

ExecutorService executor = Executors.newSingleThreadExecutor();

newSingleThreadExecutor的源码如下

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

⚠️ 默认的核心线程和最大线程都为1，空闲超时时间为0ms，这个默认只针对非核心线程，跟newFixedThreadPool一样，使用了无界队列LinkedBlockingQueue，任务量巨大的情况下，有可能会把内存撑爆，使用时需要注意，最好自己实例化ThreadPoolExecutor对象使用。

3、创建按需创建线程的线程池

创建一个线程池，初始化线程为0，最大线程为Intetger.MAX_VALUE,即最大线程数为2^31-1个，每个线程默认最大空闲时间为60s，超过60s将被回收，新任务执行如果没有空闲线程可用，就重新创建新的线程。

ExecutorService executor = Executors.newCachedThreadPool();

newCachedThreadPool()源码如下

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

⚠️ 核心线程为0，最大线程为Integer.MAX_VALUE，也就是2^32个，最大可以创建几十亿个线程，空闲超时时间为60s，内部队列是SynchronousQueue，也是一个阻塞队列，但是内部只能放一个元素，因此newCachedThreadPool每收到一个任务都会新开一个线程去处理，非常消耗系统资源，有可能撑爆内存，使用时需要注意，最好自己实例化ThreadPoolExecutor对象使用。

4、创建可定时执行的线程池

创建一个指定N个核心线程数，最大线程数为Intetger.MAX_VALUE的线程池，超出核心线程数据的线程最大空闲时间为0s，即空闲了即被回收。

ExecutorService executor = Executors.newScheduledThreadPool(8);

newScheduledThreadPool源码如下

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

//...


public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService{
    //...
  
  
  public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }
  
  //...
}

⚠️ 核心线程数为可设置参数，最大线程数为Integer.MAX_VALUE，空闲时间为0，阻塞队列为DelayedWorkQueue，最大可以创建几十亿线程，有可能撑爆内存，使用时需要注意，最好自己实例化ThreadPoolExecutor对象使用。

5、创建任务窃取线程池

这是JDK1.8新增的线程池，该线程池维护足够的线程以支持给定的并行度级别。这里的并行度是指在多处理器机器中在单个时间点将用于执行给定任务的最大线程数。

ExecutorService executor = Executors.newWorkStealingPool();

newWorkStealingPool源代码如下

public static ExecutorService newWorkStealingPool(int parallelism) {
    return new ForkJoinPool
        (parallelism,
         ForkJoinPool.defaultForkJoinWorkerThreadFactory,
         null, true);
}

newWorkStealingPool的实现和前面几个都不一样，是基于ForkJoinPool 类实现，这是一个并行的线程池，参数传入的是并发的线程数量，因此不会保证任务的顺序执行，线程是抢占式执行。

停止线程池

shutdown()

将线程池状态设置为SHUTDOWN状态，并不会立即停止

  • 停止接收submit()提交的新任务

  • 已经提交的正在执行的和正在等待执行的任务会执行完成

  • 没有interrup()操作

shutdownNow()

将线程池状态设置为STOP状态，尝试立即停止，但不一定成功立即停止

  • 停止接收submit()提交的新任务
  • 正在等待执行的任务会被取消
  • 尝试interrup()正在执行的线程，但并不能保证一定能成功的interrupt线程池中的线程
  • 会返回并未终止的线程列表

awaitTermination(timeout,unit)

线程阻塞，接收timeout和unit两个参数，用于设置超时时间以及单位，方法执行完返回bool执行结果，当

  • 在超时时间内所有线程（包括正在执行的和等待执行的）执行完成，返回true
  • 在超时时间内到，没有执行完成，返回false
  • 执行过程线程被中断，则抛出InterruptedException异常
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 219,110评论 6赞 508
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 93,443评论 3赞 395
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 165,474评论 0赞 356
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 58,881评论 1赞 295
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 67,902评论 6赞 392
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 51,698评论 1赞 305
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 40,418评论 3赞 419
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 39,332评论 0赞 276
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 45,796评论 1赞 316
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 37,968评论 3赞 337
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 40,110评论 1赞 351
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 35,792评论 5赞 346
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 41,455评论 3赞 331
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 32,003评论 0赞 22
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 33,130评论 1赞 272
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 48,348评论 3赞 373
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 45,047评论 2赞 355