前置知识

关键类:
Executor <-extends- ExecutorService <-implements-AbstractExecutorService <-extends- ExecutorService
Callable,Future

Executor:

任务的定义和执行分开,只有一个执行Runnable方法:
void execute(Runnable command);

ExecutorService:

除了继承Executor可以执行任务的功能,还完善了整个任务执行器(线程池)的生命周期.比如,一个线程池里面有很多线程,怎么提交任务,执行完任务之后应该怎么处理线程,怎么关闭等等.

Future&Callable

可以看到ExecutorService里面有个方法,提交异步的任务:
Future submit(Callable task);
Callable:和Runnable类似,是一个任务,只不过它执行完后有返回值,有了返回值就可以有各种玩法了.
Future:Callable执行完后有一个返回值,通过Future可以拿到这个结果.
存储了一个会在将来产生的结果.
看个简单的例子:

  1. public static void main(String[] args) throws ExecutionException, InterruptedException {
  2.         ExecutorService service = Executors.newCachedThreadPool();
  3.         Future<String> future = service.submit(() -> {
  4.             // spend some seconds doing something
  5.             TimeUnit.SECONDS.sleep(2);
  6.             return "Hello World!";
  7.         }); //异步
  9.         System.out.println("i will get~");
  10.         System.out.println(future.get());//阻塞
  11.         service.shutdown();
  12.     }

FutureTask(比较常用)

FutureTask implements RunnableFuture,RunnableFuture extend Runnable, Future
FutureTask内部维护了一个Callable成员变量

前面用的Callable只是一个任务,Future只是一个返回值,这个FutureTask就是结合了一下,既是任务又是返回值.(Apple+pen->pineapple! 😃)
线程池WorkStealingPool和ForkJoinPool用到了FutureTask
看一个小例子:

  1. public static void main(String[] args) throws InterruptedException, ExecutionException {
  2.         FutureTask<Integer> task = new FutureTask<>(() -> {
  3.             TimeUnit.MILLISECONDS.sleep(500);
  4.             return 1000;
  5.         }); //new Callable () { Integer call();}
  6.         // 可以是线程,也可以是线程池
  7.         new Thread(task).start();
  8.         System.out.println(task.get()); //阻塞
  9.     }

CompletableFuture(非常灵活)

CompletableFuture implements Future,CompletionStage

一个典型的应用场景:
有很多个子系统,他们各自有自己的数据库存储系统,可能是MySQL/Oracle/MongoDB等,现在需要统计他们的指标(比如平均请求响应时间)在一张大屏上展示分析.如果串行去查询子系统的数据,那这个分析的API就执行太久了,但是我们使用CompletableFuture,多线程异步执行,那时间就大大缩短.
当然这个场景用普通线程执行Callable也是可以搞定的,只是用CompletableFuture比较方便,相当于JDK已经造好轮子了,我们可以直接用它.
show my code:

  1. public class TestCompletableFuture {
  2.     public static void main(String[] args) throws ExecutionException, InterruptedException {
  3.         normalTest();
  4.         futureTest();
  5. //        test002();
  6.     }
  8.     private static void normalTest() {
  9.         long start = System.currentTimeMillis();
  10.         Map<String, Double> metrics = new HashMap<>(4);
  11.         metrics.put("metricsOfMySQL", metricsOfMySQL());
  12.         metrics.put("metricsOfOracle", metricsOfOracle());
  13.         metrics.put("metricsOfMongoDB", metricsOfMongoDB());
  15.         long end = System.currentTimeMillis();
  16.         System.out.println("use serial method call! " + (end - start));
  17.         System.out.println(metrics + "\n-------------------------------------");
  18.     }
  20.     private static void futureTest() {
  21.         long start = System.currentTimeMillis();
  23.         Map<String, Double> metrics = new HashMap<>(4);
  24.         CompletableFuture<Double> metricsOfMySQL = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfMySQL)
  25.                 .thenApply(value -> metrics.put("metricsOfMySQL", value));
  26.         CompletableFuture<Double> metricsOfOracle = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfOracle)
  27.                 .thenApply(value -> metrics.put("metricsOfOracle", value));
  28.         CompletableFuture<Double> metricsOfMongoDB = CompletableFuture.supplyAsync(TestCompletableFuture::metricsOfMongoDB)
  29.                 .thenApply(value -> metrics.put("metricsOfMongoDB", value));
  31.         System.out.println(metrics);
  32.         CompletableFuture.allOf(metricsOfMySQL, metricsOfOracle, metricsOfMongoDB).join();
  34.         long end = System.currentTimeMillis();
  35.         System.out.println("use completable future! " + (end - start));
  36.         System.out.println(metrics);
  37.     }
  39.     // 其他的用法,灵活的处理结果,有点函数式编程的感觉
  40.     private static void test002() {
  41.         // 异步执行
  42.         CompletableFuture.supplyAsync(() -> metricsOfMongoDB())
  43.                 .thenApply(String::valueOf)
  44.                 .thenApply(str -> "price " + str)
  45.                 .thenAccept(System.out::println);
  46.         // 阻塞住主线程,等待上面执行完
  47.         try {
  48.             System.in.read();
  49.         } catch (IOException e) {
  50.             e.printStackTrace();
  51.         }
  52.     }
  54.     private static double metricsOfMySQL() {
  55.         sleepRandom();
  56.         return 2.00;
  57.     }
  59.     private static double metricsOfOracle() {
  60.         sleepRandom();
  61.         return 3.00;
  62.     }
  64.     private static double metricsOfMongoDB() {
  65.         sleepRandom();
  66.         return 1.00;
  67.     }
  69.     private static void sleepRandom() {
  70.         int time = new Random().nextInt(500);
  71.         try {
  72.             TimeUnit.MILLISECONDS.sleep(time);
  73.         } catch (InterruptedException e) {
  74.             e.printStackTrace();
  75.         }
  76.         System.out.printf("After %s sleep!\n", time);
  77.     }
  78. }
  79. }

认识ThreadPoolExecutor

最开始我们创建一个线程,执行一个Runnable任务,执行完后就销毁线程了.
然而创建一个线程需要跟操作系统申请资源,这个过程是比较耗时的;所以我们最好是让线程复用,即让一个线程去持续执行不同的任务,而不是执行一个任务后就销毁.(可以类比一下泡面桶和陶瓷碗)
线程池里面不仅仅是线程,它维护这两个集合,一个是线程集合,一个是任务集合.

ThreadPoolExecutor的七个重要参数:

  1. int corePoolSize

    核心线程数,最开始线程池里面没有线程,来了任务后会先创建一定数量的核心线程去执行任务;一般没有任务执行时也不会回收核心线程.

  2. int maximumPoolSize

    最大线程数.当任务比较多,核心线程执行不过来时会放入任务队列,任务队列满了后会创建非核心 线 程,maximumPoolSize=临时线程数+核心线程数,主要负责控制临时线程数.非核心线程在空闲一段时间后会被回收.

  3. long keepAliveTime

    生存时间.当一个非核心线程很长时间不执行任务了,就销毁该线程,这个参数就是控制空闲阈值.
    核心线程默认不受此控制,也可以设置参数指定核心线程受此控制(allowCoreThreadTimeOut).

  4. TimeUnit unit 生存时间单位,见名知意

  5. BlockingQueue workQueue 任务队列

  6. ThreadFactory threadFactory

    线程工厂,自定义创建线程的方式.
    有个默认的DefaultThreadFactory,指定了线程名字,daemon=false,priority=5(NORM_PRIORITY).不要小看线程名,多线程环境追踪错误日志时大有用处.

  7. RejectedExecutionHandler handler

拒绝策略当任务很多,任务队列满了,非核心线程数也达到上限后,再来任务的时候的处理策略.拒绝策略可以自定义,JDK提供了四种拒绝策略:
7.1 AbortPolicy:抛异常,这也是默认的拒绝策略
7.2DiscardPolicy:安静的丢掉
7.3DiscardOldestPolicy:丢掉队列中最老的任务,把新的放入队列
做游戏的时候可能会用,比如一个角色的每次移动作为一个操作当如线程池中,正常情况是依次移动;当队列满了就把最老的丢掉,减少影响.
7.4CallerRunsPolicy:提交任务者(调用execute的线程)处理该任务
(实战中这四种一般都不用,而是自定义)

阿里开发手册1.5.0里面一丶(六)也讲到,很多关于线程的规范,下面列举几条:

  1. 线程资源必须通过线程池提供，不允许在应用中自行显式创建线程。

  2. 线程池不允许使用 Executors 去创建，而是通过ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。

    说明：Executors 返回的线程池对象的弊端如下：
    1） FixedThreadPool 和 SingleThreadPool：
    允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求,从而导致 OOM。
    2） CachedThreadPool：
    允许的创建线程数量为 Integer.MAX_VALUE，可能会创建大量的线程,从而导致 OOM。

  3. 创建线程或线程池时请指定有意义的线程名称，方便出错时回溯。

测试小例子:

  1. public class T05_00_HelloThreadPool {
  3.     static class Task implements Runnable {
  4.         private int i;
  6.         public Task(int i) {
  7.             this.i = i;
  8.         }
  10.         @Override
  11.         public void run() {
  12.             // 打印一下当前线程
  13.             System.out.println(Thread.currentThread().getName() + " Task " + i);
  14.             try {
  15.                 // 阻塞住,以便认识不同的拒绝策略
  16.                 TimeUnit.DAYS.sleep(1);
  17.             } catch (InterruptedException e) {
  18.                 e.printStackTrace();
  19.             }
  20.         }
  22.         @Override
  23.         public String toString() {
  24.             return "Task{" +
  25.                     "i=" + i +
  26.                     '}';
  27.         }
  28.     }
  30.     public static void main(String[] args) {
  31.         // 初始化一个线程池,最多同时接纳8个任务
  32.         ThreadPoolExecutor tpe = new ThreadPoolExecutor(2, 4,
  33.                 60, TimeUnit.SECONDS,
  34.                 new ArrayBlockingQueue<Runnable>(4),
  35.                 Executors.defaultThreadFactory(),
  36.                 new ThreadPoolExecutor.CallerRunsPolicy()
  37.         );
  39.         // 把线程池占满
  40.         for (int i = 0; i < 8; i++) {
  41.             tpe.execute(new Task(i));
  42.         }
  44.         // 打印一下当前线程等待队列,是线程2,3,4,5;
  45.         // 因为0和1正被核心线程执行,6和7被非核心线程执行
  46.         System.out.println(tpe.getQueue());
  48.         tpe.execute(new Task(100));
  50.         // 如果是DiscardOldestPolicy,会发现任务2被丢掉了,任务100加入等待队列
  51.         // 如果是CallerRunsPolicy,这句话不会打印,因为新的任务被主线程执行,而任务会阻塞线程;但是会打印main Task 100
  52.         System.out.println("main thread end\n" + tpe.getQueue());
  54.         // 不再接收新任务,等已有任务执行完后关掉线程池
  55.         tpe.shutdown();
  56.         // 尝试马上关掉线程池,不等当前任务结束,而是通过Thread.interrupt打断线程
  57.         // 如果线程没有正确处理InterruptedException,那就永远那不会被终结
  58. //        tpe.shutdownNow();
  59.     }
  60. }

调整线程池的大小

下面是一个建议,也可以说是标准公式吧,但是这个公式中的等待时间和预估时间的比率很难预估出来,工程中还是需要经过各种情况的压力测试,然后取一个相对各方面都照顾的到的值.
一般的等待时间都花在IO上,所以W/C比较高时也称为IO密集型.
image.png

ThreadPoolExecutor源码解析

这块扣起来贼头疼,我们先领会战略精神,具体战术日后再议…
昨天看AQS的源码,扣了半天没搞明白,浪费好多时间,还有很多”上天入地”的任务待完成…
这里补充记录一点,JAVA(不知道其他语言怎么说…)中整数的表示形式,以4位的数来说:
1.正整数和0,就是正常的二进制,1就是0001,2就是0010
2.负整数=对应正整数的反码的补码,-1反码->1110补码->1111,即十进制-1的二进制为1111

1、常用变量的解释

  1. // 1. `ctl`，可以看做一个int类型的数字，高3位表示线程池状态，低29位表示worker数量
  2. private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
  3. // 2. `COUNT_BITS`，`Integer.SIZE`为32，所以`COUNT_BITS`为29
  4. private static final int COUNT_BITS = Integer.SIZE - 3;
  5. // 3. `CAPACITY`，线程池允许的最大线程数。1左移29位，然后减1，即为 2^29 - 1
  6. private static final int CAPACITY   = (1 << COUNT_BITS) - 1;
  8. // runState is stored in the high-order bits
  9. // 4. 线程池有5种状态，按大小排序如下：RUNNING < SHUTDOWN < STOP < TIDYING < TERMINATED
  10. private static final int RUNNING    = -1 << COUNT_BITS;
  11. private static final int SHUTDOWN   =  0 << COUNT_BITS;
  12. private static final int STOP       =  1 << COUNT_BITS;
  13. private static final int TIDYING    =  2 << COUNT_BITS;
  14. private static final int TERMINATED =  3 << COUNT_BITS;
  16. // Packing and unpacking ctl
  17. // 5. `runStateOf()`，获取线程池状态，通过按位与操作，低29位将全部变成0
  18. private static int runStateOf(int c)     { return c & ~CAPACITY; }
  19. // 6. `workerCountOf()`，获取线程池worker数量，通过按位与操作，高3位将全部变成0
  20. private static int workerCountOf(int c)  { return c & CAPACITY; }
  21. // 7. `ctlOf()`，根据线程池状态和线程池worker数量，生成ctl值
  22. private static int ctlOf(int rs, int wc) { return rs | wc; }
  24. /*
  25.  * Bit field accessors that don't require unpacking ctl.
  26.  * These depend on the bit layout and on workerCount being never negative.
  27.  */
  28. // 8. `runStateLessThan()`，线程池状态小于xx
  29. private static boolean runStateLessThan(int c, int s) {
  30.     return c < s;
  31. }
  32. // 9. `runStateAtLeast()`，线程池状态大于等于xx
  33. private static boolean runStateAtLeast(int c, int s) {
  34.     return c >= s;
  35. }

2、构造方法

  1. public ThreadPoolExecutor(int corePoolSize,
  2.                           int maximumPoolSize,
  3.                           long keepAliveTime,
  4.                           TimeUnit unit,
  5.                           BlockingQueue<Runnable> workQueue,
  6.                           ThreadFactory threadFactory,
  7.                           RejectedExecutionHandler handler) {
  8.     // 基本类型参数校验
  9.     if (corePoolSize < 0 ||
  10.         maximumPoolSize <= 0 ||
  11.         maximumPoolSize < corePoolSize ||
  12.         keepAliveTime < 0)
  13.         throw new IllegalArgumentException();
  14.     // 空指针校验
  15.     if (workQueue == null || threadFactory == null || handler == null)
  16.         throw new NullPointerException();
  17.     this.corePoolSize = corePoolSize;
  18.     this.maximumPoolSize = maximumPoolSize;
  19.     this.workQueue = workQueue;
  20.     // 根据传入参数`unit`和`keepAliveTime`，将存活时间转换为纳秒存到变量`keepAliveTime `中
  21.     this.keepAliveTime = unit.toNanos(keepAliveTime);
  22.     this.threadFactory = threadFactory;
  23.     this.handler = handler;
  24. }

3、提交执行task的过程

  1. public void execute(Runnable command) {
  2.     if (command == null)
  3.         throw new NullPointerException();
  4.     /*
  5.      * Proceed in 3 steps:
  6.      *
  7.      * 1. If fewer than corePoolSize threads are running, try to
  8.      * start a new thread with the given command as its first
  9.      * task.  The call to addWorker atomically checks runState and
  10.      * workerCount, and so prevents false alarms that would add
  11.      * threads when it shouldn't, by returning false.
  12.      *
  13.      * 2. If a task can be successfully queued, then we still need
  14.      * to double-check whether we should have added a thread
  15.      * (because existing ones died since last checking) or that
  16.      * the pool shut down since entry into this method. So we
  17.      * recheck state and if necessary roll back the enqueuing if
  18.      * stopped, or start a new thread if there are none.
  19.      *
  20.      * 3. If we cannot queue task, then we try to add a new
  21.      * thread.  If it fails, we know we are shut down or saturated
  22.      * and so reject the task.
  23.      */
  24.     int c = ctl.get();
  25.     // worker数量比核心线程数小，直接创建worker执行任务
  26.     if (workerCountOf(c) < corePoolSize) {
  27.         if (addWorker(command, true))
  28.             return;
  29.         c = ctl.get();
  30.     }
  31.     // worker数量超过核心线程数，任务直接进入队列
  32.     if (isRunning(c) && workQueue.offer(command)) {
  33.         int recheck = ctl.get();
  34.         // 线程池状态不是RUNNING状态，说明执行过shutdown命令，需要对新加入的任务执行reject()操作。
  35.         // 这儿为什么需要recheck，是因为任务入队列前后，线程池的状态可能会发生变化。
  36.         if (! isRunning(recheck) && remove(command))
  37.             reject(command);
  38.         // 这儿为什么需要判断0值，主要是在线程池构造方法中，核心线程数允许为0
  39.         else if (workerCountOf(recheck) == 0)
  40.             addWorker(null, false);
  41.     }
  42.     // 如果线程池不是运行状态，或者任务进入队列失败，则尝试创建worker执行任务。
  43.     // 这儿有3点需要注意：
  44.     // 1. 线程池不是运行状态时，addWorker内部会判断线程池状态
  45.     // 2. addWorker第2个参数表示是否创建核心线程
  46.     // 3. addWorker返回false，则说明任务执行失败，需要执行reject操作
  47.     else if (!addWorker(command, false))
  48.         reject(command);
  49. }

4、addworker源码解析

  1. private boolean addWorker(Runnable firstTask, boolean core) {
  2.     retry:
  3.     // 外层自旋
  4.     for (;;) {
  5.         int c = ctl.get();
  6.         int rs = runStateOf(c);
  8.         // 这个条件写得比较难懂，我对其进行了调整，和下面的条件等价
  9.         // (rs > SHUTDOWN) || 
  10.         // (rs == SHUTDOWN && firstTask != null) || 
  11.         // (rs == SHUTDOWN && workQueue.isEmpty())
  12.         // 1. 线程池状态大于SHUTDOWN时，直接返回false
  13.         // 2. 线程池状态等于SHUTDOWN，且firstTask不为null，直接返回false
  14.         // 3. 线程池状态等于SHUTDOWN，且队列为空，直接返回false
  15.         // Check if queue empty only if necessary.
  16.         if (rs >= SHUTDOWN &&
  17.             ! (rs == SHUTDOWN &&
  18.                firstTask == null &&
  19.                ! workQueue.isEmpty()))
  20.             return false;
  22.         // 内层自旋
  23.         for (;;) {
  24.             int wc = workerCountOf(c);
  25.             // worker数量超过容量，直接返回false
  26.             if (wc >= CAPACITY ||
  27.                 wc >= (core ? corePoolSize : maximumPoolSize))
  28.                 return false;
  29.             // 使用CAS的方式增加worker数量。
  30.             // 若增加成功，则直接跳出外层循环进入到第二部分
  31.             if (compareAndIncrementWorkerCount(c))
  32.                 break retry;
  33.             c = ctl.get();  // Re-read ctl
  34.             // 线程池状态发生变化，对外层循环进行自旋
  35.             if (runStateOf(c) != rs)
  36.                 continue retry;
  37.             // 其他情况，直接内层循环进行自旋即可
  38.             // else CAS failed due to workerCount change; retry inner loop
  39.         } 
  40.     }
  41.     boolean workerStarted = false;
  42.     boolean workerAdded = false;
  43.     Worker w = null;
  44.     try {
  45.         w = new Worker(firstTask);
  46.         final Thread t = w.thread;
  47.         if (t != null) {
  48.             final ReentrantLock mainLock = this.mainLock;
  49.             // worker的添加必须是串行的，因此需要加锁
  50.             mainLock.lock();
  51.             try {
  52.                 // Recheck while holding lock.
  53.                 // Back out on ThreadFactory failure or if
  54.                 // shut down before lock acquired.
  55.                 // 这儿需要重新检查线程池状态
  56.                 int rs = runStateOf(ctl.get());
  58.                 if (rs < SHUTDOWN ||
  59.                     (rs == SHUTDOWN && firstTask == null)) {
  60.                     // worker已经调用过了start()方法，则不再创建worker
  61.                     if (t.isAlive()) // precheck that t is startable
  62.                         throw new IllegalThreadStateException();
  63.                     // worker创建并添加到workers成功
  64.                     workers.add(w);
  65.                     // 更新`largestPoolSize`变量
  66.                     int s = workers.size();
  67.                     if (s > largestPoolSize)
  68.                         largestPoolSize = s;
  69.                     workerAdded = true;
  70.                 }
  71.             } finally {
  72.                 mainLock.unlock();
  73.             }
  74.             // 启动worker线程
  75.             if (workerAdded) {
  76.                 t.start();
  77.                 workerStarted = true;
  78.             }
  79.         }
  80.     } finally {
  81.         // worker线程启动失败，说明线程池状态发生了变化（关闭操作被执行），需要进行shutdown相关操作
  82.         if (! workerStarted)
  83.             addWorkerFailed(w);
  84.     }
  85.     return workerStarted;
  86. }

5、线程池worker任务单元

  1. private final class Worker
  2.     extends AbstractQueuedSynchronizer
  3.     implements Runnable
  4. {
  5.     /**
  6.      * This class will never be serialized, but we provide a
  7.      * serialVersionUID to suppress a javac warning.
  8.      */
  9.     private static final long serialVersionUID = 6138294804551838833L;
  11.     /** Thread this worker is running in.  Null if factory fails. */
  12.     final Thread thread;
  13.     /** Initial task to run.  Possibly null. */
  14.     Runnable firstTask;
  15.     /** Per-thread task counter */
  16.     volatile long completedTasks;
  18.     /**
  19.      * Creates with given first task and thread from ThreadFactory.
  20.      * @param firstTask the first task (null if none)
  21.      */
  22.     Worker(Runnable firstTask) {
  23.         setState(-1); // inhibit interrupts until runWorker
  24.         this.firstTask = firstTask;
  25.         // 这儿是Worker的关键所在，使用了线程工厂创建了一个线程。传入的参数为当前worker
  26.         this.thread = getThreadFactory().newThread(this);
  27.     }
  29.     /** Delegates main run loop to outer runWorker  */
  30.     public void run() {
  31.         runWorker(this);
  32.     }
  34.     // 省略代码...
  35. }

6、核心线程执行逻辑-runworker

  1. final void runWorker(Worker w) {
  2.     Thread wt = Thread.currentThread();
  3.     Runnable task = w.firstTask;
  4.     w.firstTask = null;
  5.     // 调用unlock()是为了让外部可以中断
  6.     w.unlock(); // allow interrupts
  7.     // 这个变量用于判断是否进入过自旋（while循环）
  8.     boolean completedAbruptly = true;
  9.     try {
  10.         // 这儿是自旋
  11.         // 1. 如果firstTask不为null，则执行firstTask；
  12.         // 2. 如果firstTask为null，则调用getTask()从队列获取任务。
  13.         // 3. 阻塞队列的特性就是：当队列为空时，当前线程会被阻塞等待
  14.         while (task != null || (task = getTask()) != null) {
  15.             // 这儿对worker进行加锁，是为了达到下面的目的
  16.             // 1. 降低锁范围，提升性能
  17.             // 2. 保证每个worker执行的任务是串行的
  18.             w.lock();
  19.             // If pool is stopping, ensure thread is interrupted;
  20.             // if not, ensure thread is not interrupted.  This
  21.             // requires a recheck in second case to deal with
  22.             // shutdownNow race while clearing interrupt
  23.             // 如果线程池正在停止，则对当前线程进行中断操作
  24.             if ((runStateAtLeast(ctl.get(), STOP) ||
  25.                  (Thread.interrupted() &&
  26.                   runStateAtLeast(ctl.get(), STOP))) &&
  27.                 !wt.isInterrupted())
  28.                 wt.interrupt();
  29.             // 执行任务，且在执行前后通过`beforeExecute()`和`afterExecute()`来扩展其功能。
  30.             // 这两个方法在当前类里面为空实现。
  31.             try {
  32.                 beforeExecute(wt, task);
  33.                 Throwable thrown = null;
  34.                 try {
  35.                     task.run();
  36.                 } catch (RuntimeException x) {
  37.                     thrown = x; throw x;
  38.                 } catch (Error x) {
  39.                     thrown = x; throw x;
  40.                 } catch (Throwable x) {
  41.                     thrown = x; throw new Error(x);
  42.                 } finally {
  43.                     afterExecute(task, thrown);
  44.                 }
  45.             } finally {
  46.                 // 帮助gc
  47.                 task = null;
  48.                 // 已完成任务数加一 
  49.                 w.completedTasks++;
  50.                 w.unlock();
  51.             }
  52.         }
  53.         completedAbruptly = false;
  54.     } finally {
  55.         // 自旋操作被退出，说明线程池正在结束
  56.         processWorkerExit(w, completedAbruptly);
  57.     }
  58. }

Executors线程工厂

1. newSingleThreadExecutor()

不建议使用,LinkedBlockingQueue可能会堆积Integer.MAX_VALUE个任务

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

2. newCachedThreadPool()

不建议使用,可能会创建Integer.MAX_VALUE个线程
这里面用了SynchronousQueue作为等待队列

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

3. newFixedThreadPool()

不建议使用,LinkedBlockingQueue可能会堆积Integer.MAX_VALUE个任务

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

4. newScheduledThreadPool

特点:任务队列是内部类DelayedWorkQueue
一般也不咋用,第一是因为maxPoolSize是Integer.MAX_VALUE,第二个是我们有专门的定时任务调度框架(比如quartz)

  1.   public ScheduledThreadPoolExecutor(int corePoolSize) {
  2.         super(corePoolSize, Integer.MAX_VALUE,
  3.               DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
  4.               new DelayedWorkQueue());
  5.     }

5. newWorkStealingPool(其实是ForkJoinPool)

其实就是一个ForkJoinPool,具体见下面的ForkJoinPool

  1.   public static ExecutorService newWorkStealingPool() {
  2.         return new ForkJoinPool
  3.             (Runtime.getRuntime().availableProcessors(),
  4.              ForkJoinPool.defaultForkJoinWorkerThreadFactory,
  5.              null, true);
  6.     }

任务在执行的时候,如果改任务不是ForkJoinTask,则会被转换成ForkJoinTask.RunnableExecuteAction(task)

  1. public void execute(Runnable task) {
  2.         if (task == null)
  3.             throw new NullPointerException();
  4.         ForkJoinTask<?> job;
  5.         if (task instanceof ForkJoinTask<?>) // avoid re-wrap
  6.             job = (ForkJoinTask<?>) task;
  7.         else
  8.             job = new ForkJoinTask.RunnableExecuteAction(task);
  9.         externalSubmit(job);
  10.     }

ForkJoinPool

ForkJoinPool是和ThreadPoolExecutor同级别的一个类,我觉得这个Pool很厉害.

ForkJoin的思想就和MapReduce很像,会把一个大任务切分成若干个小任务执行,这个过程就是Fork(分叉);小任务执行完后再汇总起来,得到一个整体的结果,这个过程就是Join(汇总).

ThreadPoolExecutor是多个worker线程共用一个任务队列,从里面取任务执行;而ForkJoinPool是每个线程worker(ForkJoinWorkerThread)有自己的一个任务队列;
当一个线程的任务执行完啦,会从其他线程的队列中偷一个加到自己的队列中来执行.

ForkJoinPool底层使用了work-stealing算法,all threads in the pool attempt to find and execute tasks submitted to the pool and/or created by other active tasks (eventually blocking waiting for work if none exist).
里面所有的工作线程在初始化的时候都被设置成守护线程isDaemon=true.守护线程的特点是:当只有守护线程时,JVM会退出.我觉得应该是担心很大的任务,或者有些死循环导致程序不能退出.
什么是守护线程?看这里(https://www.cnblogs.com/quanxiaoha/p/10731361.html)

当然肯定不是所有的任务都可以去分叉拆分,所以ForkJoinPool只接收ForkJoinTask(的实现者).
但是ForkJoinTask比较原始,实现起来比较麻烦,一般我们自定义实现这两个类:
RecursiveAction不带返回值
RecursiveTask带返回值

  1. public class T12_ForkJoinPool {
  2.     static int[] nums = new int[1000000];
  3.     static final int MAX_NUM = 50000;
  4.     static Random r = new Random();
  6.     static {
  7.         long start = System.currentTimeMillis();
  8.         for (int i = 0; i < nums.length; i++) {
  9.             nums[i] = r.nextInt(100);
  10.         }
  11.         long end = System.currentTimeMillis();
  12.         System.out.println("---" + Arrays.stream(nums).sum() + "|" + (end - start)); //stream api
  13.     }
  15.     // 没有返回值,可能用于处理一些后台任务
  16.     static class AddTask extends RecursiveAction {
  17.         int start, end;
  19.         AddTask(int s, int e) {
  20.             start = s;
  21.             end = e;
  22.         }
  24.         @Override
  25.         protected void compute() {
  26.             if (end - start <= MAX_NUM) {
  27.                 long sum = 0L;
  28.                 for (int i = start; i < end; i++) {
  29.                     sum += nums[i];
  30.                 }
  31.                 System.out.println("from:" + start + " to:" + end + " = " + sum + "|" + Thread.currentThread().isDaemon());
  32.             } else {
  33.                 int middle = start + (end - start) / 2;
  34.                 AddTask subTask1 = new AddTask(start, middle);
  35.                 AddTask subTask2 = new AddTask(middle, end);
  36.                 subTask1.fork();
  37.                 subTask2.fork();
  38.             }
  39.         }
  40.     }
  42.     // 有返回值
  43.     static class AddTaskRet extends RecursiveTask<Long> {
  44.         private static final long serialVersionUID = 1L;
  45.         int start, end;
  47.         AddTaskRet(int s, int e) {
  48.             start = s;
  49.             end = e;
  50.         }
  52.         @Override
  53.         protected Long compute() {
  54.             if (end - start <= MAX_NUM) {
  55.                 long sum = 0L;
  56.                 for (int i = start; i < end; i++) {
  57.                     sum += nums[i];
  58.                 }
  59. //                System.out.println("from:" + start + " to:" + end + " = " + sum + "|" + Thread.currentThread().isDaemon());
  60.                 return sum;
  61.             }
  62.             int middle = start + (end - start) / 2;
  63.             AddTaskRet subTask1 = new AddTaskRet(start, middle);
  64.             AddTaskRet subTask2 = new AddTaskRet(middle, end);
  65.             subTask1.fork();
  66.             subTask2.fork();
  67.             return subTask1.join() + subTask2.join();
  68.         }
  69.     }
  71.     public static void main(String[] args) throws IOException {
  72. //        ForkJoinPool fjp001 = new ForkJoinPool();
  73. //        AddTask task001 = new AddTask(0, nums.length);
  74. //        fjp001.execute(task001);
  75. //        System.in.read();
  77.         ForkJoinPool fjp = new ForkJoinPool();
  78.         long start = System.currentTimeMillis();
  79.         AddTaskRet task = new AddTaskRet(0, nums.length);
  80.         fjp.execute(task);
  81.         // 这个join会阻塞等待结果
  82.         long result = task.join();
  83.         long end = System.currentTimeMillis();
  84.         System.out.println("join result :" + result + "|" + (end - start));
  85.     }
  86. }