Java并发之线程池的使用浅析

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

线程池有这么几个重要的参数

1. corePoolSize : 线程池里的核心线程数量
2. maximumPoolSize : 线程池里允许有的最大线程数量
3. keepAliveTime :表示线程没有任务执行时最多保持多久时间会终止。 默认情况下，如果 当前线程数量  >  corePoolSize ，多出来的线程会在 keepAliveTime 之后就被释放掉，直到线程池中的线程数不大于corePoolSize。但是如果调用了allowCoreThreadTimeOut(boolean)方法，在线程池中的线程数不大于corePoolSize时，keepAliveTime参数也会起作用，直到线程池中的线程数为0
4. unit : keepAliveTime的时间单位,有7种单位

   • TimeUnit.DAYS;                           //天
     TimeUnit.HOURS;                        //小时
     TimeUnit.MINUTES;                    //分钟
     TimeUnit.SECONDS;                  //秒
     TimeUnit.MILLISECONDS;         //毫秒
     TimeUnit.MICROSECONDS;       //微妙
     TimeUnit.NANOSECONDS;        //月     

5. workQueue : 队列 workQueue的类型为BlockingQueue<Runnable>，通常可以取下面三种类型：
   1. 有界任务队列ArrayBlockingQueue：基于数组的先进先出队列，此队列创建时必须指定大小； 
   2. 无界任务队列LinkedBlockingQueue：基于链表的先进先出队列，如果创建时没有指定此队列大小，则默认为Integer.MAX_VALUE；
   3. 直接提交队列synchronousQueue：这个队列比较特殊，它不会保存提交的任务，而是将直接新建一个线程来执行新来的任务。如果进程数量已经达到最大值，则执行拒绝策略。因此使用该队列需要设置很大的maximumPoolSize，否则很容易执行拒绝策略。
6. threadFactory : 每当需要创建新的线程放入线程池的时候,就是通过这个线程工厂来创建的
7. handler : 就是说当线程,队列都满了,之后采取的策略,比如抛出异常等策略
   • ThreadPoolExecutor.AbortPolicy:丢弃任务并抛出RejectedExecutionException异常。
   • ThreadPoolExecutor.DiscardPolicy：也是丢弃任务，不做任何处理
   • ThreadPoolExecutor.DiscardOldestPolicy：丢弃队列最前面的任务，然后重新尝试执行任务
   • ThreadPoolExecutor.CallerRunsPolicy：由调用线程处理该任务

3. newScheduledThreadPool

计划任务，和其他线程池不同，该线程池并不一定会立即安排任务，主要是起计划任务的作用。他会在指定时间、对任务进行调度。

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

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

• scheduleAtFixedRate:是以固定的频率去执行任务，周期是指每次执行任务成功执行之间的间隔。
• schedultWithFixedDelay:是以固定的延时去执行任务，延时是指上一次执行成功之后和下一次开始执行的之前的时间

4. newSingleThreadExecutor

单个线程线程池，只有一个线程的线程池，阻塞队列使用的是LinkedBlockingQueue,若有多余的任务提交到线程池中，则会被暂存到阻塞队列，待空闲时再去执行。按照先入先出的顺序执行任务。

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

public static ExecutorService newSingleThreadExecutor(ThreadFactory var0) {
       return new Executors.FinalizableDelegatedExecutorService(new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue(), var0));
}

5. newWorkStealingPool

会根据所需的并行层次来动态创建一个拥有足够的线程数目的线程池。通过使用多个队列来降低竞争。并行的层次是和运行的最大线程数目相关。运行过程中实际的线程数目或许会动态地增长和收缩。

其本质是一个一个工作窃取的线程池，所以对于提交的任务不能保证是顺序执行的。底层用的ForkJoinPool来实现的。ForkJoinPool的优势在于，可以充分利用多cpu，多核cpu的优势，

把一个任务拆分成多个“小任务”，把多个“小任务”放到多个处理器核心上并行执行；当多个“小任务”执行完成之后，再将这些执行结果合并起来即可。分治的思想。下面是newWorkStealingPool的构造函数

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

//使用一个无限队列来保存需要执行的任务，可以传入线程的数量，不传入，则默认使用当前计算机中可用的cpu数量，使用分治法来解决问题，使用fork()和join()来进行调用
public ForkJoinPool(int parallelism, ForkJoinWorkerThreadFactory factory, UncaughtExceptionHandler handler,boolean asyncMode) {
        this(checkParallelism(parallelism), checkFactory(factory), handler, asyncMode ? FIFO_QUEUE : LIFO_QUEUE,"ForkJoinPool-" + nextPoolId() + "-worker-");
        checkPermission();
    }

假设有3个线程A、B、C在运行，workStealing可以简单这么认为，每个线程都维护自己的一个队列，线程A的队列里头积累了3个任务，线程B的队列里2个任务，C的队列里1个任务；那么当线程C执行完任务之后，它会去别的线程池所维护的队列里面把任务偷过来继续执行，主动的找活干。

我们看一下newWorkStealingPool的使用案例

/**
 * WorkStealingPool(任务窃取，都是守护线程)
 * 每个线程都有要处理的队列中的任务，如果其中的线程完成自己队列中的任务，
 * 那么它可以去其他线程中获取其他线程的任务去执行
 */
public class TestWorkStealingPool {

    public static void main(String[] args) throws IOException {
        // 根据cpu是几核来开启几个线程
        ExecutorService service = Executors.newWorkStealingPool();
        // 查看当前计算机是几核
        System.out.println(Runtime.getRuntime().availableProcessors());
        service.execute(new R(1000));
        service.execute(new R(3000));
        service.execute(new R(4000));
        service.execute(new R(2000));
        service.execute(new R(3000));
        service.execute(new R(3000));
        service.execute(new R(3000));
        service.execute(new R(3000));

        // WorkStealing是精灵线程(守护线程、后台线程)，主线程不阻塞，看不到输出。
        // 虚拟机不停止，守护线程不停止
        System.in.read();
    }

    static class R implements Runnable {
        int time;

        public R(int time) {
            this.time = time;
        }

        @Override
        public void run() {
            System.out.println(time + ":" + Thread.currentThread().getName() + "执行时间为：" + System.currentTimeMillis());
            try {
                TimeUnit.MILLISECONDS.sleep(time);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

结果输出

可以看到newWorkStealingPool执行时根据cpu核心分配线程数量，这里打印显示cpu核心数为4，明显的可以看出线程将前4 个任务扔给了1-4号线程，后四个任务在排队等待。当1-4号线程中某个线程最先执行完毕后

会自动窃取未执行的任务，这里可以看到1号线程最先执行完毕用时1000ms，然后就窃取第五个任务。第0号线程执行完毕后，又紧接着执行第六个任务....。依次类推，直到所有任务执行完毕。

四、线程池的使用建议

建议根据自己的需要手动创建线程池（ new ThreadPoolExecutor（......） ），这样可以灵活使用线程池，并且可可以加深自己对线程池的理解。阿里巴巴开发手册和 Dubbo线程池的开发手册也是这样建议

Dubbo线程池的开发手册如下

阿里巴巴开发手册建议如下

我们可以看一下Dubbo怎么创建线程池的

@SPI("fixed")
public interface ThreadPool {

    /**
     * 线程池
     * 
     * @param url 线程参数
     * @return 线程池
     */
    @Adaptive({Constants.THREADPOOL_KEY})
    Executor getExecutor(URL url);

}
public class FixedThreadPool implements ThreadPool {

    public Executor getExecutor(URL url) {
        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
        int threads = url.getParameter(Constants.THREADS_KEY, Constants.DEFAULT_THREADS);
        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
        return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, 
                queues == 0 ? new SynchronousQueue<Runnable>() : 
                    (queues < 0 ? new LinkedBlockingQueue<Runnable>() 
                            : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }

}

默认情况下,Dubbo的 FixedThreadPool 中, maximumPoolSize = 200 ,队列是容量很小的 SynchronousQueue .所以当线程超过200的时候,线程池就会抛出异常.