技术专家带你彻底掌握线程池

public enum State {
        /**
         * 线程被创建但还未启动
         */
        NEW,

        /**
         * 线程为就绪（可运行）状态，在 jvm 中执行，但是可能需要等待其他操作系统资源执行
         */
        RUNNABLE,

        /**
         * 线程被监控器锁阻塞
         */
        BLOCKED,

        /**
         * 线程处于等待状态，需要被唤醒才能继续执行
         */
        WAITING,

        /**
         * 等待超时，正在等待的线程超过了指定的等待时间。
         */
        TIMED_WAITING,

        /**
         * 线程终止，线程执行完成
         */
        TERMINATED;
    }

public interface Executor {

    void execute(Runnable command);
}

// 提交 Runnable 任务
submit(Callable<T> task);
submit(Runnable task);
submit(Runnable task, T result);
// 请求关闭、发生超时或者当前线程中断，无论哪一个首先发生之后，都将导致阻塞，直到所有任务完成执行
awaitTermination(long timeout, TimeUnit unit);
// 启动一次顺序关闭，执行以前提交的任务，但不接受新任务
shutdown();
// 试图停止所有正在执行的活动任务，暂停处理正在等待的任务，并返回等待执行的任务列表
shutdownNow();
// 批量执行给定的任务
invokeAll(Collection<? extends Callable<T>> tasks);
// 执行单个指定的任务
invokeAny(Collection<? extends Callable<T>> tasks);

// 为给定可运行任务和默认值返回一个 RunnableFuture。
protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value);
// 为给定可运行任务返回一个 RunnableFuture。
protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable)

// 原子整形, 底层采用 CAS 原理控制并发
private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

// runState is stored in the high-order bits
private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

// 计算当前运行状态
private static int runStateOf(int c)     { return c & ~CAPACITY; }
// 计算当前线程数量
private static int workerCountOf(int c)  { return c & CAPACITY; }
// 通过状态和线程数生成 ctl
private static int ctlOf(int rs, int wc) { return rs | wc; }

private boolean addWorker(Runnable firstTask, boolean core) {
       //相当于 goto
        retry:
        for (;;) {
            int c = ctl.get();
            int rs = runStateOf(c);
            // 如果线程池的状态到了 SHUTDOWN 或者之上的状态时候，只有一种情况还需要继续添加线程，
            // 那就是线程池已经 SHUTDOWN，但是队列中还有任务在排队,而且不接受新任务（firstTask 为 null）
            // 这里还继续添加线程的原因是加快执行等待队列中的任务，尽快让线程池关闭
            // Check if queue empty only if necessary.
            if (rs >= SHUTDOWN &&
                ! (rs == SHUTDOWN &&
                   firstTask == null &&
                   ! workQueue.isEmpty()))
                return false;

            for (;;) {
                int wc = workerCountOf(c);
               // 传入的 core 的参数，唯一用到的地方，如果线程数超过理论最大容量，如果 core 是 true 跟最大核心线程数比较，否则跟最大线程数比较
                if (wc >= CAPACITY ||
                    wc >= (core ? corePoolSize : maximumPoolSize))
                    return false;
                // 通过 CAS 自旋，增加线程数+1，增加成功跳出双层循环，继续往下执行
                if (compareAndIncrementWorkerCount(c))
                    break retry;
               // 检测当前线程状态如果发生了变化，则继续回到 retry，重新开始循环
                c = ctl.get();  // Re-read ctl
                if (runStateOf(c) != rs)
                    continue retry;
                // else CAS failed due to workerCount change; retry inner loop
            }
        }
        // 走到这里，说明我们已经成功的将线程数+1 了，但是真正的线程还没有被添加
        boolean workerStarted = false;
        boolean workerAdded = false;
        Worker w = null;
        try {
           // 添加线程，Worker 是继承了 AQS，实现了 Runnable 接口的包装类
            w = new Worker(firstTask);
            final Thread t = w.thread;
            if (t != null) {
               // 加锁
                final ReentrantLock mainLock = this.mainLock;
                mainLock.lock();
                try {
                    // Recheck while holding lock.
                    // Back out on ThreadFactory failure or if
                    // shut down before lock acquired.
                    int rs = runStateOf(ctl.get());
                    // 检查线程状态, 逻辑和之前一样
                    if (rs < SHUTDOWN ||
                        (rs == SHUTDOWN && firstTask == null)) {
                       // 线程只能被 start 一次
                        if (t.isAlive()) // precheck that t is startable
                            throw new IllegalThreadStateException();
                      // workers 是一个 HashSet，添加我们新增的 Worker
                        workers.add(w);
                        int s = workers.size();
                        if (s > largestPoolSize)
                            largestPoolSize = s;
                        workerAdded = true;
                    }
                } finally {
                    mainLock.unlock();
                }
                if (workerAdded) {
                  // 启动 Worker
                    t.start();
                    workerStarted = true;
                }
            }
        } finally {
            if (! workerStarted)
                addWorkerFailed(w);
        }
        return workerStarted;
    }

    /** 使用 Executors 工具类创建固定数量线程池 */
    private ExecutorService executorService = Executors.newFixedThreadPool(3);

    public void start() {
        // 提交一个 Runnable 任务
        executorService.submit(() -> {
            System.out.println("hello word");
        });
    }

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

    /** 使用 Executors 工具类创建 */
    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    public void start() {
        // 提交一个 Runnable 任务
        executorService.submit(() -> {
            System.out.println("hello word");
        });
    }

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

    private ExecutorService executorService = Executors.newCachedThreadPool();

    public void start() {
        executorService.submit(() -> {
            System.out.println("hello word");
        });
    }

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

    /**
     * Creates a {@code SynchronousQueue} with the specified fairness policy.
     *
     * @param fair if true, waiting threads contend in FIFO order for
     *        access; otherwise the order is unspecified.
     */
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

    // 创建一个定时任务线程池, 并指定核心线程数
    private ExecutorService executorService = Executors.newScheduledThreadPool(10);

    public void start() {
        // 提交一个 Runnable 任务
        executorService.submit(() -> {
            System.out.println("hello word");
        });
    }

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

    /**
     * Creates a new {@code ScheduledThreadPoolExecutor} with the
     * given core pool size.
     *
     * @param corePoolSize the number of threads to keep in the pool, even
     *        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @throws IllegalArgumentException if {@code corePoolSize < 0}
     */
    public ScheduledThreadPoolExecutor(int corePoolSize) {
        super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
              new DelayedWorkQueue());
    }

public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory,
                              RejectedExecutionHandler handler) {
        if (corePoolSize < 0 ||
            maximumPoolSize <= 0 ||
            maximumPoolSize < corePoolSize ||
            keepAliveTime < 0)
            throw new IllegalArgumentException();
        if (workQueue == null || threadFactory == null || handler == null)
            throw new NullPointerException();
        this.acc = System.getSecurityManager() == null ?
                null :
                AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

   static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        DefaultThreadFactory() {
            // 从安全管理器中拿到线程组
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            // 指定线程的名字
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            // 设置用户进程
            if (t.isDaemon())
                t.setDaemon(false);
            // 设置优先级
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

    /**
     * 源码中默认使用的是 AbortPolicy 策略 
     * The default rejected execution handler
     */
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();

        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-context</artifactId>
            <version>4.3.12.RELEASE</version>
        </dependency>

/**
 * @author 九月长安
 * @version $Id: MyThreadPoolConfig.java, v 0.1 2021-08-03 18:41 九月长安 Exp $$
 */
@Configuration
public class MyThreadPoolConfig {

    // 指定注入的 bean 名称
    @Bean(name = "executorService")
    public ExecutorService getThreadPool() {
        return new ThreadPoolExecutor(2,
            20,
            30L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue(),
            Executors.defaultThreadFactory(),
            new AbortPolicy());
    }
}

/**
 * @author 九月长安
 * @version $Id: UserService.java, v 0.1 2021-08-03 11:16 九月长安 Exp $$
 */
@Service
public class UserService {

    @Autowired
    private UserDao userDao;

    // 注入线程池
    @Autowired
    private ExecutorService executorService;

    public void getUserInfo() throws ExecutionException, InterruptedException {
        // 提交有返回值的任务
        Future<Person> future = executorService.submit(new Callable<Person>() {
            @Override
            public Person call() throws Exception {
                return userDao.getPerson();
            }
        });
        // 获取返回结果
        Person p = future.get();
        // 打印
        System.out.println(p);
    }

}

@Configuration
public class MyThreadPoolConfig {

    // 指定注入的 bean 名称
    @Bean(name = "executorService")
    public ExecutorService getThreadPool() {
        return new ThreadPoolExecutor(2,
            20,
            30L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue(),
            Executors.defaultThreadFactory(),
            new AbortPolicy());
    }
}