五种线程池的对比与使用

Posted 2021-04-06 Java高级架构与思想

今天对五种常见的java内置线程池进行讲解。

线程使用的demo

public static void cache() {        
ExecutorService pool = Executors.newCachedThreadPool();        long start = System.currentTimeMillis();        pool.execute(() -> {            int sum = 0;            for (int i = 0; i < 10; i++) {                sum = (int) Math.sqrt(i * i - 1 + i);                System.out.println(sum);            }        });        System.out.println("cache: " + (System.currentTimeMillis() - start));    }

newCachedThreadPool

• 这是一个可缓存线程池，可以灵活的回收空闲线程，无可回收线程时，新建线程

源码：

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

通过源码可以看出底层调用的是ThreadPoolExecutor方法，传入一个同步的阻塞队列实现缓存。

下面说一下ThreadPoolExecutor

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

通过源码可以看出，我们可以传入线程池的核心线程数(最小线程数)，最大线程数量，保持时间，时间单位，阻塞队列这些参数，最大线程数设置为jvm可用的cpu数量为最佳实践

newWorkStealingPool

• 创建持有足够线程的线程池来并行，通过使用多个队列减少竞争，不传参数，则默认设定为cpu的数量

源码：

public static ExecutorService newWorkStealingPool() {        return new ForkJoinPool      (Runtime.getRuntime().availableProcessors(),        ForkJoinPool.defaultForkJoinWorkerThreadFactory,             null, true);    }

通过源码可以看出底层调用的是ForkJoinPool线程池

下面说一下ForkJoinPool

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();    }

使用一个无限队列来保存需要执行的任务，可以传入线程的数量，不传入，则默认使用当前计算机中可用的cpu数量，使用分治法来解决问题，使用fork()和join()来进行调用

newSingleThreadExecutor

创建一个单线程化的线程池，保证所有任务按照指定的顺序执行(FIFO,LIFO,优先级)，当要求进程限制时，可以进行使用

源码：

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

newFixedThreadPool

创建一个固定线程数量，可重用的线程池

源码：

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

newScheduledThreadPool

创建一个可定期或者延时执行任务的线程池

源码：

return new ScheduledThreadPoolExecutor(corePoolSize);

通过源码可以看出底层调用的是一个ScheduledThreadPoolExecutor，然后传入线程数量

下面来介绍一下ScheduledThreadPoolExecutor

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

通过源码可以看出底层调用了ThreadPoolExecutor，维护了一个延迟队列，可以传入线程数量，传入延时的时间等参数，下面给出一个demo

public static void main(String[] args) {   ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);        
      
      for (int i = 0; i < 15; i = i + 5) {     pool.schedule(() -> System.out.println("我被执行了，
      当前时间" + new Date()), i, TimeUnit.SECONDS);        }        pool.shutdown();
    }

执行结果

我被执行了，当前时间Fri Jan 12 11:20:41 CST 2018 我被执行了，当前时间Fri Jan 12 11:20:46 CST 2018 我被执行了，当前时间Fri Jan 12 11:20:51 CST 2018

有的小伙伴可能会用疑问，为什么使用schedule()而不使用submit()或者execute()呢，下面通过源码来分析

    public void execute(Runnable command) {        schedule(command, 0, NANOSECONDS);    }     public Future<?> submit(Runnable task) {        
               return schedule(task, 0, NANOSECONDS);
    }

通过源码可以发现这两个方法都是调用的schedule(),而且将延时时间设置为了0，所以想要实现延时操作，需要直接调用schedule()

下面我们再来分析一下submit()和execute()的以及shutdown()和shutdownNow()的区别

• submit()，提交一个线程任务，可以接受回调函数的返回值吗，适用于需要处理返回着或者异常的业务场景

• execute()，执行一个任务，没有返回值

• shutdown()，表示不再接受新任务，但不会强行终止已经提交或者正在执行中的任务

• shutdownNow()，对于尚未执行的任务全部取消，正在执行的任务全部发出interrupt()，停止执行

五种线程池的适应场景

1. newCachedThreadPool：用来创建一个可以无限扩大的线程池，适用于服务器负载较轻，执行很多短期异步任务。

2. newFixedThreadPool：创建一个固定大小的线程池，因为采用无界的阻塞队列，所以实际线程数量永远不会变化，适用于可以预测线程数量的业务中，或者服务器负载较重，对当前线程数量进行限制。

3. newSingleThreadExecutor：创建一个单线程的线程池，适用于需要保证顺序执行各个任务，并且在任意时间点，不会有多个线程是活动的场景。

4. newScheduledThreadPool：可以延时启动，定时启动的线程池，适用于需要多个后台线程执行周期任务的场景。

5. newWorkStealingPool：创建一个拥有多个任务队列的线程池，可以减少连接数，创建当前可用cpu数量的线程来并行执行，适用于大耗时的操作，可以并行来执行

以上就是今天所想要分享的内容，由于并发接触并不深入，如有错误，请联系，浩瀚星辰中渺小的我们，却有着改变世界的梦想