Java 线程池

前言：啊里巴巴开发手册2021）

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

说明：Executors 返回的线程池对象的弊端如下：

1） FixedThreadPool和 SingleThreadPool： 允许的请求队列长度为 Integer.MAX_VALUE，可能会堆积大量的请求，从而导致 OOM。

2）CachedThreadPool： 允许的创建线程数量为Integer.MAX_VALUE，可能会创建大量的线程，从而导致 OOM。

java自带的线程池工具类（不推荐使用）

newCachedThreadPool

源码

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

创建newCacheThreadPool线程池：

ExecutorService executorService = Executors.newCachedThreadPool();

Executors.newCachedThreadPool()创建出来的newCachedThreadPool实际上是创建一个ThreadPoolExecutor（线程池执行器）类

ThreadPoolExecutor参数说明：

• corePoolSize：核心线程数
• maxnumPoolSize：最大线程数
• keepAliveTime：线程的最大生命周期，这里的生命周期有两个约束条件：==一：该参数针对的是超过corePoolSize数量的线程；二：处于非运行状态的线程。==即非核心线程数存活时间（非核心线程数 = 最大线程数 - 核心线程数）。
• unit：时间单位
• workQueue：任务队列。当线程池中的线程都处于运行状态，而此时任务数量继续增加，则需要一个容器来容纳这些任务，这就是任务队列。

其中，Interger.MAX_VALUE = 2³¹ - 1，即线程池中核心线程为 0 个，非核心线程最大为 2³¹ - 1（而不是初始化就有这么多），空闲的非核心线程生命周期为60s。任务队列为SynchronousQueue，这个队列是无法插入任务的，一有任务立即执行，所以newCachedThreadPool比较适合任务量大但耗时少的任务。

SynchronousQueue特点

• SynchronousQueue没有容量。与其他BlockingQueue不同，SynchronousQueue是一个不存储元素的BlockingQueue。每一个put操作必须要等待一个take操作，否则不能继续添加元素，反之亦然。
• 因为没有容量，所以对应的peek、contains、clear、isEmpty…等方法其实是无效的。例如clear是不执行任何操作的，contains始终返回false，peek始终返回null。
• SynchronousQueue分为公平和非公平，默认情况下采用非公平性访问策略，当然也可以通过构造函数来设置为公平性访问策略（为true即可）。
• SynchronousQueue内部采用了无锁实现（CAS）

执行过程

一旦有任务 ，newCachedThreadPool就会不断创建非核心线程来执行任务，执行任务结束的线程会重新回到线程池等待复用。

优缺点

优点

1. 效率快

缺点

1. 比较消耗cpu
2. 可能存在呢内存泄露（oom）问题，当线程处理业务耗时较长时，对于新的任务，线程池会不断创建新的非核心线程来处理任务，最后导致内存溢出

newFixThreadPool

源码

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

创建newFixedThreadPool线程池：

ExecutorService fixedThreadPool = Executors.newFixedThreadPool(n); // 参数核心n代表线程数量，也代表最大线程数量

newFixedThreadPool的核心线程数量和最大线程数量为传入的n，也就是说非核心线程数量为 0，非运行状态的非核心线程存活时间是0 ms，任务队列是LinkedBlockingQueue（链表实现的有界队列，默认构造方法的队列长度是Integer.MXA_VALUE）

执行过程

newFixedThreadPool线程池执行任务时，如果线程池内存在空闲的核心线程，则用核心线程执行任务，如果核心线程都不空闲，此时还有任务产生，则将任务存储到LinkedBlockingQueue队列中，等待核心线程执行任务后，再从队列中取出任务执行

优缺点

• 执行速度适中
• 当任务过多的时候，可能存在oom问题

newSingleThreadPool

源码

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

创建newSingleThreadPool

ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();

newSingleThreadPool的核心线程数量和最大线程数量为传入的1，也就是说非核心线程数量为 0，非运行状态的非核心线程存活时间是0 ms，任务队列是LinkedBlockingQueue（链表实现的有界队列，默认构造方法的队列长度是Integer.MXA_VALUE）

执行过程

newSingleThreadPool线程池执行任务时，只有唯一的核心线程在执行任务，核心线程执行任务结束，再从队列中取出任务执行

优缺点

• 速度最慢
• 当任务过多的时候，可能存在oom问题

周期性线程池 newScheduledThreadPool

常用于需要延迟执行或周期循环执行任务的场景

源码

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

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

构造方法

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

• threadFactory：执行器创建新线程时使用的工

或者

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

周期执行：scheduleAtFixedRate()

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit);

• command：任务线程
• initialDelay：初始执行延迟时间
• period：执行周期
• unit：时间单位，如果任务执行时间大于执行周期，则任务执行完成后立马开始执行下一轮任务。

demo

package com.zhan.threadPool;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Demo {
    public static void main(String[] args) {
        System.out.println("时间："+new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format( new Date()));
        ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(10);
        scheduledThreadPool.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                try{
                    Thread.sleep(1000L);
                    System.out.println("时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format( new Date()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, 5, 2, TimeUnit.SECONDS);

    }
}

设置任务时长大于执行周期

Thread.sleep(3000L);

延迟执行：schedule()

public ScheduledFuture<?> schedule(Runnable command,
                                   long delay, TimeUnit unit);

demo

scheduledThreadPool.schedule(new Runnable() {
            @Override
            public void run() {
                try{
                    //Thread.sleep(3000L);
                    System.out.println("时间：" + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format( new Date()));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }, 2, TimeUnit.SECONDS);
    }

执行后线程不会结束

延迟执行并返回结果：submit()

public <T> Future<T> submit(Runnable task, T result) {
    return schedule(Executors.callable(task, result), 0, NANOSECONDS);
}

执行后将执行结果放入result，并作为返回值返回

线程池执行过程

提交优先级：核心线程池 > 队列 > 非核心线程池

执行优先级：核心线程池 > 非核心线程池 > 队列

线程异常策略

• ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
• ThreadPoolExecutor.DiscardPolicy：丢弃任务，但是不抛出异常。
• ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新提交被拒绝的任务
• ThreadPoolExecutor.CallerRunsPolicy：由调用线程（提交任务的线程）处理该任务

线程池的5种状态

状态转换