JAVA线程池 之 Executors

Posted 2020-10-22 斌灬小生不才

一、背景

　　线程属于系统稀缺资源，在使用线程时，如果无限制创建线程，达到CPU高负荷时，会导致系统运行缓慢，更有甚者直接宕机。

在这样的基础上我们希望在使用线程时，竟可能使系统线程数处于一个可控范围，尽可能实现线程的重用。

 

二、Executors 分析

Executors 示例  DEMO

　　

/**
 * @author binH
 * @date 2018/01/24
 */
package org.lsnbin.thread;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ThreadAnalyze {
    
    private    static ExecutorService fexecutor = Executors.newFixedThreadPool(20);
    
    public static void main(String[] args) {
        for (int i = 0; i < 20; i++) {
            fexecutor.execute(new ThreadAnalyze().new Task());
        }
    }
    
    private class Task implements Runnable{

        @Override
        public void run() {
            System.out.println("Thread name --> " + Thread.currentThread().getName());
        }
    }
}

　　示例分析：

　　1、使用Executors初始化一个包含10个线程的线程池

　　2、使用execute方法提交20个任务，打印线程名

　　3、负责执行任务的线程的生命周期交由Executors管理。

 

三、Executors 内部分析

　　Executors是线程池的工厂类，内部基于ThreadPoolExecutor实现，对ThreadPoolExecutor进行封装，并提供相对应方法。

　　newFixedThreadPool

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

初始化一定数量的线程池，当corePoolSize=maximumPoolSize时，使用LinkedBlockingQueue作为阻塞队列，不过当线程池不使用时，也不会释放线程。

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
不初始化线程，在需要使用时创建线程，最大允许数为Integer.MAX_VALUE，使用SynchronousQueue作为阻塞队列，当线程空闲时keepAliveTime后释放线程。
容易存在问题：在高并发情况下，一瞬间会创建大量的线程，会出现严重的性能问题。

　　newSingleThreadExecutor

 public static ExecutorService newSingleThreadExecutor() {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
初始化只有一个线程的线程池，如果该线程由于异常结束，会重新创建一个新的线程。线程不过期，使用LinkedBlockingQueue作为阻塞队列。唯一个线程可以保证任务的顺序执行。

　　newScheduledThreadPool

 public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
        return new ScheduledThreadPoolExecutor(corePoolSize);
    }
初始化一定数量线程的线程池，线程池中的任务可以在指定的时间内周期性的执行所提交的任务。可用于定期同步。

　　newWorkStealingPool  -- JDK1.8

 public static ExecutorService newWorkStealingPool(int parallelism) {
        return new ForkJoinPool
            (parallelism,
             ForkJoinPool.defaultForkJoinWorkerThreadFactory,
             null, true);
    }
创建持有足够线程的线程池来支持给定的并行级别，并通过使用多个队列来减少竞争，需要指定parallelism并行参数，如果没指定，则默认CPU数量。

　　ForkJoinPool：支持大任务分解成小任务的线程池，这是Java8新增线程池，通常配合ForkJoinTask接口的子类RecursiveAction或RecursiveTask使用。

 

四、ThreadPoolExecutor 分析

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

  参数分析：

　　1、corePoolSize

　　　　线程池核心线程数。当线程池提交一个任务时，线程池就会创建一个新线程并执行任务，直到线程池内运行状态的线程数等于corePoolSize。

　　　　当继续有任务提交时，会被放入阻塞队列中。使用 prestartCoreThread() 方法可以提前预创建并启动所有核心线程。

　　2、maximumPoolSize

　　　　线程池最大可用线程数。在线程数等于corePoolSize时，如果当前阻塞队列满了，继续提交任务时会创建新的线程并执行任务。前提时线程数小于maximumPoolSize。

　　3、keepAliveTime

　　　　线程池中线程空闲存活时间。默认情况下该情况只会在大于corePoolSize时有用。

　　4、unit

　　　　keepAliveTime存活时间的单位。

　　5、workQueue

　　　　用来保存等待被执行的任务的阻塞队列。JDK实现的队列有以下几种：

　　　　　　ArrayBlockingQueue：基于数组结构的有界阻塞队列，按照FIFO ( 先进先出 ) 排序任务。

 

　　　　　 LinkedBlockingQueue：基于链表结构的阻塞队列，按照FIFO ( 先进先出 ) 排序任务。效果比ArrayBlockingQueue好。

　　　　　　DelayQueue：具有延时执行的无界阻塞队列。

　　　　　　SynchronousQueue：一个不能存储任何元素的阻塞队列，每个插入的操作必须等待另一个相应的删除操作，否则插入则会一直处于阻塞状态。

　　　　　　　　　　　　　　　　第二个线程必须等待前一个线程结束。

　　　　　　PriorityBlockingQueue：具有 优先级的无界阻塞队列。

 

　　6、threadFactory

　　　　创建线程的工厂

　　　　默认工厂实现：

　　

 DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }

 

　　7、handler

　　　　　　线程池的饱和策略，用于线程数饱和与阻塞队列已满的情况。有任务继续提交时触发的处理逻辑。线程池提供了4种策略：

　　　　　　AbortPolicy：直接跑出异常，默认策略。

　　　　　　DiscardPolicy：直接丢弃任务，不处理。

　　　　　　DiscardOldestPolicy：丢弃阻塞队列中靠前的旧的任务，执行新的任务。

　　　　　　CallerRunsPolicy：用调用者的线程来执行任务。

 

 

 

 

 

 

　　

threadFactory