线程池

Posted 2021-01-02 duanjiapingjy

　　　　　　　　Executor

　　　　　　　　ExecutorService

AbstractExecutorService　　ScheduledExecutorService

ThreadPoolExecutor

　　　　　　　　ScheduledTreadPollExecutor

 

ThreadPoolExecutor

参数

private volatile long keepAliveTime;　　空闲线程存活时间（毫秒数），当线程数量大于corePoolSize或者allowCoreThreadTimeOut为true时，此参数才会起作用

private volatile boolean allowCoreThreadTimeOut;　　是否允许核心线程超市等待。默认false，核心线程始终保持存活状态，即使核心线程空闲。true，核心线程在空闲时，存活指定的时间keepAliveTime

private volatile int corePoolSize;　　线程池中活动线程的最小数量，如果allowCoreThreadTimeOut（允许核心线程等待超时）参数设置为true，corePoolSize为0

private volatile int maximumPoolSize;　　线程池维护线程的最大数量

private volatile ThreadFactory threadFactory;　　线程工厂，所有的线程的创建都要经过线程工厂，Executors.defaultThreadFactory()  （默认使用的线程工厂）

private final ReentrantLock mainLock = new ReentrantLock();

private final HashSet<Worker> workers = new HashSet<Worker>();　　包含线程池中所有活动线程，直到持有mainLock锁时，线程才开始执行

private final BlockingQueue<Runnable> workQueue;　　任务队列，缓存未执行的任务

private static final RejectedExecutionHandler defaultHandler = new AbortPolicy();　　拒绝策略，当线程池和任务队列处于饱和状态时，使用拒绝策略处理新任务（默认直接丢弃任务并抛出异常）

 

线程的创建时机

线程池中线程的个数：由corePoolSize和maximumPoolSize决定

线程池中线程的创建时机：由corePoolSize和workQueue决定

在线程池中没有空闲线程时，如何处理新任务？

（1）线程池中的线程数量小于corePoolSize，创建一个新的线程处理新任务

（2）线程池中的线程数量大于等于corePoolSize，但是workQueue任务队列未满，则缓存新任务

（3）线程池中的线程数量大于等于corePoolSize，workQueue任务队列已满，并且线程数量小于maximumPoolSize，创建一个新的线程处理新任务

（4）线程池中的线程数量大于等于maximumPoolSize，并且workQueue任务队列已满，使用拒绝策略处理新任务

线程的回收策略

参数keepAliveTime和allowCoreThreadTimeOut

对于超出corePoolSize数量的线程，当线程空闲的时间超过keepAliveTime，该空闲线程会被回收

对于corePoolSize数量之内的线程，是否回收取决于allowCoreThreadTimeOut是否为true

缓存策略

当线程池中现成的个数大于等于corePoolSize，并且workQueue未满时，将新任务缓存在队列中

BlockingQueue<Runnable>

public interface BlockingQueue<E> extends Queue<E>

BlockingQueue<E>接口的四个实现类：

　　ArrayBlockingQueue<E>（有界、数组、先进先出） 

　　LinkedBlockingQueue<E>（有界、链表、先进先出） 

　　 PriorityBlockingQueue<E>（优先级） 

　　SynchronousQueue<E>（同步、该队列不存储元素，每一个插入操作必须等待另一个线程调用移除操作，否则插入一直阻塞）

拒绝策略

当线程池中的线程数量大于等于maximumPoolSize，并且workQueue队列已满，根据拒绝策略处理新任务，默认的拒绝策略是AbortPolicy，可以通过公共的setXxx方法执行拒绝策略

RejectedExecutionHandler 

RejectedExecutionHandler 接口的实现类：

　　AbortPolicy　　丢弃新任务，并抛出RejectedExecutionException

　　CallerRunsPolicy　　有调用线程执行新任务

　　DiscardOldestPolicy　　丢弃最旧的任务，即队列队首的任务，执行新任务

　　DiscardPolicy　　丢弃新任务，不做其他任何操作

创建线程和线程服用

必须使用线程工厂ThreadFactory创建线程池，默认使用Executors.defaultThreadFactory()线程工厂（构造器中的代码），可以通过公共的setXxx方法指定要使用的线程工厂

ThreadFactory接口

　　实现类DefaultThreadFactory

线程池中，线程可以复用。一个线程在执行完一个任务之后，不会立即退出，而是到任务队列的队头获取一个新的任务继续执行，如果任务队列为空，并且线程不满足回收条件（keepAliveTime和allowCoreThreadTimeOut参数决定线程回收的条件），会一直阻塞。

 线程池中的线程放在private final HashSet<Worker> workers = new HashSet<Worker>(); 中，Worker是ThreadPoolExecutor的内部类

private final class Worker
extends AbstractQueuedSynchronizer
implements Runnable

public void run() {
runWorker(this);
}

线程执行时，调用run方法，然后调用runWorker方法，在该方法中，循环从任务队列中获取任务，并执行任务

 

 提交任务

通过调用线程池的submit方法提交任务，在submit方法中调用execute方法

提交任务有可能立即被执行，有可能放到任务队列中，有可能被拒绝

关闭线程池

　　shutdown 将线程池的状态置为SHUTDOWN 中断空闲线程　　advanceRunState(SHUTDOWN);　　interruptIdleWorkers();

　　shutdownNow 将线程池的状态置为STOP 中断所有线程 将未执行的任务全部返回advanceRunState(STOP);　　interruptWorkers();　　return tasks;

 

 

Executors（工具类）

Executors工具类的作用就是创建线程池，一般情况下，不直接调用ThreadPoolExecutor类的构造器创建线程池

（1）调用Executors工具类的相关方法，创建一个线程池，返回一个ExecutorService 或者 ScheduledExecutorService对象

（2）调用ExecutorService对象的submit方法向线程池中提交任务，在submit方法中调用ThreadPoolExecutor的execute方法，在execute方法中判断当前任务是创建新线程执行、加入任务队列、或者调用拒绝策略

　　或者调用ScheduledExecutorService对象的submit方法提交任务，在submit方法中调用ScheduledThreadPoolExecutor的schedule方法

 

    public static ExecutorService newFixedThreadPool(int nThreads) {　　//构建包括固定线程数量的线程池，默认情况下，空闲线程不会被回收
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    public static ExecutorService newSingleThreadExecutor() {　　　　//构建一个线程数量为1的线程池，等价于newFixedThreadPool(1)
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    public static ExecutorService newCachedThreadPool() {　　　　//构建线程数量不确定的线程池，线程数量随任务量变动，默认线程存活时间超过60秒会被回收
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    public static ScheduledExecutorService newSingleThreadScheduledExecutor() {　　//等价于newScheduledThreadPool（1）
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }
    public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {　　//构建线程数量为corePoolSize的线程池，可执行定时任务

        return new ScheduledThreadPoolExecutor(corePoolSize);
    }