JUC系列（九）| ThreadPool 线程池

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

/**
     * 可缓存线程池
     * @return
     */
public static ExecutorService newCachedThreadPool() {
    /**
        * corePoolSize 线程池的核心线程数
        * maximumPoolSize 能容纳的最大线程数
        * keepAliveTime 空闲线程存活时间
        * unit 存活的时间单位
        * workQueue 存放提交但未执行任务的队列
        * threadFactory 创建线程的工厂类:可以省略
        * handler 等待队列满后的拒绝策略:可以省略
        */
    return new ThreadPoolExecutor(0,
                                  Integer.MAX_VALUE,
                                  60L,
                                  TimeUnit.SECONDS,
                                  new SynchronousQueue<>(),
                                  Executors.defaultThreadFactory(),
                                  new ThreadPoolExecutor.AbortPolicy());
}

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

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

public static ScheduledExecutorService newScheduledThreadPool(
    int corePoolSize, ThreadFactory threadFactory) {
    return new ScheduledThreadPoolExecutor(corePoolSize, threadFactory);
}
    
public ScheduledThreadPoolExecutor(int corePoolSize,
                                   ThreadFactory threadFactory) {
    super(corePoolSize, Integer.MAX_VALUE,
          DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
          new DelayedWorkQueue(), threadFactory);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         threadFactory, defaultHandler);
}

import java.util.concurrent.*;

/**
 * 入门案例
 */
public class ThreadPoolDemo1 {


    /**
     * 火车站 3 个售票口, 10 个用户买票
     *
     * @param args
     */
    public static void main(String[] args) {
        //定时线程次:线程数量为 3---窗口数为 3
        ExecutorService threadService = new ThreadPoolExecutor(3,
                3,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.DiscardOldestPolicy());
        try {
            //10 个人买票
            for (int i = 1; i <= 10; i++) {
                threadService.execute(() -> {
                    try {
                        System.out.println(Thread.currentThread().getName() + " 窗口,开始卖票");
                        Thread.sleep(500);
                        System.out.println(Thread.currentThread().getName() + " 窗口买票结束");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                });
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //完成后结束
            threadService.shutdown();
        }
    }
}
/**
 pool-1-thread-3 窗口,开始卖票
 pool-1-thread-2 窗口,开始卖票
 pool-1-thread-1 窗口,开始卖票
 pool-1-thread-3 窗口买票结束
 pool-1-thread-2 窗口买票结束
 pool-1-thread-1 窗口买票结束
 ....
 */

public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();
    int c = ctl.get();
   //如果线程池的线程个数少于corePoolSize则创建新线程执行当前任务
    if (workerCountOf(c) < corePoolSize) {
        //执行addworker，创建一个核心线程，创建失败重新获取ctl
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
//如果工作线程数大于核心线程数，判断线程池的状态是否为running，并且可以添加进队列
//如果线程池不是running状态，则执行拒绝策略，（还是会调用一次addworker）
    if (isRunning(c) && workQueue.offer(command)) {
         //再次获取ctl，进行双重检索
        int recheck = ctl.get();
        //如果线程池是不是处于RUNNING的状态，那么就会将任务从队列中移除， 
        //如果移除失败，则会判断工作线程是否为0 ，如果过为0 就创建一个非核心线程 
        //如果移除成功，就执行拒绝策略，因为线程池已经不可用了；
        if (! isRunning(recheck) && remove(command))
            //为给定的命令调用被拒绝的执行处理程序
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
   //检查是否可以根据当前池状态和给定界限（核心或最大值）添加新的工作线程。 
   //如果是这样，则相应地调整工作人员数量，并且如果可能，将创建并启动一个新工作人员，将 firstTask 作为其第一个任务运行。 
   //如果池已停止或有资格关闭，则此方法返回 false。 
   //如果线程工厂在询问时未能创建线程，它也会返回 false。 
   //如果线程创建失败，要么是由于线程工厂返回 null，要么是由于异常（通常是 Thread.start() 中的 OutOfMemoryError），我们会干净利落地回滚。
    else if (!addWorker(command, false))
        // 为给定的命令调用被拒绝的执行处理程序
        reject(command);
}