ThreadPoolExecutor线程池解析及Executor创建线程常见四种方式

用户10338856

发布于 2023-02-07 14:05:57

6840

发布于 2023-02-07 14:05:57

new Thread和线程池的比较

每次new Thread是新建了线程对象，并且不能重复使用，为什么不能重复使用？因为new是相当于在内存中独立开辟一个内存来让该线程运行，所以只能释放线程资源和新建线程，性能差。而使用线程池，可以重复使用存在的线程，减少对象的创建及消亡的开销，性能较好。

线程缺乏统一管理，可能无限制的新建线程，相互竞争，有可能占用过多的系统资源导致死机或者抛出OutOfMemoryError。而使用线程池可以有效控制最大并发线程数，提高系统资源利用率，同时避免过多资源竞争，避免阻塞。

同时new Thread，当我们需要定期执行，更多执行，线程中断等等使用Thread操作起来非常的繁琐。线程池则提供定时执行，定期执行，单线程，并发控制等功能，让我们操作线程起来特别方便。

ThreadPoolExecutor如何创建对象

在这里介绍的是JUC包下的ThreadPoolExecutor线程池，这个线程池里有4个构造方法。

public class ThreadPoolExecutor extends AbstractExecutorService{
//第一个构造方法
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
//第二个构造方法
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              ThreadFactory threadFactory) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             threadFactory, defaultHandler);
    }
//第三个构造方法
public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue,
                              RejectedExecutionHandler handler) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), handler);
    }
//第四个也是真正的初始化构造函数
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.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
}

从这四个函数来看，其实是分是否需要默认的线程池工厂和handler。接下来就讲讲这些参数代表什么。

corePoolSize:核心线程数量。当线程数少于corePoolSize的时候，直接创建新的线程，尽管其他线程是空闲的。当线程池中的线程数目达到corePoolSize后，就会把到达的任务放到缓存队列当中;

maximunPoolSize：线程池最大线程数。如果线程数量少于线程最大数且大于核心线程数量的时候，只有当阻塞队列满了才创建新线程。当线程数量大于最大线程数且阻塞队列满了这时候就会执行一些策略来响应该线程。

workQueue：阻塞队列，存储等待执行的任务，会对线程池的运行产生很大的影响。当提交一个新的任务到线程池的时候，线程池会根据当前线程数量来选择不同的处理方式

直接切换队列SynchronousQueue：该队列传递任务到线程而不持有它们。在这一点上，试图向该队列压入一个任务，如果没有可用的线程立刻运行任务，那么就会入列失败，所以一个新的线程就会被创建。当处理那些内部依赖的任务集合时，这个选择可以避免锁住。直接传递通常需要无边界的最大线程数来避免新提交任务被拒绝处理。当任务以平均快于被处理的速度提交到线程池时，它依次地确认无边界线程增长的可能性

使用无界队列LinkedBlockingQueue：使用这个队列的话，没有预先定义容量的无界队列，最大线程数是为corePoolSize，在核心线程都繁忙的时候会使新提交的任务在队列中等待被执行，所以将不会创建更多的线程，这时候，maximunPoolSize最大线程数的值将不起作用。当每个任务之间是相互独立的时比较适合该队列，任务之间不能互相影响执行。

使用有界队列ArrayBlockingQueue：使用这个队列，线程池中的最大线程数量就是maximunPoolSize，能够降低资源消耗，但是却使得线程之间调度变得更加困难，因为队列容量和线程池都规定完了。

如果想降低系统资源消耗，包括CPU使用率，操作系统资源消耗，上下文切换开销等等，可以设置一个较大的队列容量，较小的maximunPoolSize。如果线程经常发生阻塞，那么可以稍微将maximunPoolSize设置大一点。

keepAliveTime：线程没有任务执行最多保持多久时间终止。也就是当线程数量超过核心线程数量的时候，且小于最大线程数量，这一部分的线程在没有任务执行的时候是会保持直到超过keepAliveTime才会销毁

unit：keepAliveTime的时间单位

threadFactory：线程工厂，用来创建线程，当使用默认的线程工厂创建线程的时候，会使得线程具有相同优先级，并且设置了守护性，同时也设置线程名称

handler：拒绝策略。当workQueue满了，并且没有空闲的线程数，即线程达到最大线程数。就会有四种不同策略来处理。

DiscardPolicy()：丢弃掉该任务但是不抛出异常，不推荐这种(导致使用者没觉察情况发生)
DiscardOldestPolicy()：丢弃队列中等待最久的任务，然后把当前任务加入队列中。
AbortPolicy()：丢弃任务并抛出 RejectedExecutionException 异常(默认)。
CallerRunsPolicy()：由主线程负责调用任务的run()方法从而绕过线程池直接执行，既不抛弃任务也不抛出异常(当最大线程数满了，任务队列中也满了，再来一个任务的时候，由主线程执行)

注： 1. 临时线程什么时候创建？

：新任务提交时发现核心线程都在忙，任务队列也满了，并且还可以创建临时线程，

此时才会创建临时线程；

2. 什么时候会开始拒绝任务？

：核心线程和临时线程都在忙，任务队列也满了，新的任务过来的时候才会开始拒绝任务；

# execute()方法解析(JDK1.8)

　　execute()：提交任务交给线程池运行

public class ThreadPoolExecutor extends AbstractExecutorService {
    public void execute(Runnable command) {
        //任务为空则报空异常
        if (command == null)
            throw new NullPointerException();
        /*
         * 1. 如果正在运行的线程数少于corePoolSize。那么就尝试去开始一个新线程并用传入的command作为它
         * 第一个任务，然后让addworker去原子性的检查线程池的运行状态和线程数量，以至于能提前知道是否能
         * 添加线程进去。如果成功则线程运行，不成功就会去到下一步，并且再获取一次ctl的值（ctl是用来获
         * 取线程状态和线程数的）
         */
        /*
         * 2.如果一个任务能被成功的排队，那么仍然需要双重检查是否我们需要添加一个新的线程（因为有可能
         * 会第一次检查完后有一个线程销毁，所以需要双重检查）或者进入此方法后线程关闭。所以很有必要重
         * 新检查一遍状态是否需要回滚排队，是否停止或开始一个新线程或是否不存在
         */
        /* 
         * 3.如果我们无法将任务进行排队（即进入队列中），那么我们尝试添加一个新线程，如果失败我们就使用
         * 拒绝策略拒绝这个任务
         */
        int c = ctl.get();
        if (workerCountOf(c) < corePoolSize) {
            if (addWorker(command, true))
                return;
            c = ctl.get();
        }
 　　　　if (isRunning(c) && workQueue.offer(command)) {
            int recheck = ctl.get();
            if (! isRunning(recheck) && remove(command))
                reject(command);
            else if (workerCountOf(recheck) == 0)
                addWorker(null, false);
        }
         
        else if (!addWorker(command, false))
            reject(command);
    }
}

Executor线程池创建的四种线程

newFixedThreadPool：创建的是定长的线程池，可以控制线程最大并发数，超出的线程会在线程中等待，使用的是无界队列，核心线程数和最大线程数一样，当线程池中的线程没有任务时候立刻销毁,使用默认线程工厂。
newSingleThreadExecutor：创建的是单线程化的线程池，只会用唯一一个工作线程执行任务，可以指定按照是否是先入先出，还是优先级来执行任务。同样使用无界队列，核心线程数和最大线程数都是1个，同样keepAliveTime为0，可选择是否使用默认线程工厂。
newCachedThreadPool：设定一个可缓存的线程池，当线程池长度超过处理的需要，可以灵活回收空闲线程，如果没有可以回收的才新建线程。没有核心线程数，当线程没有任务60s之后就会回收空闲线程，使用有界队列。同样可以选择是否使用默认线程工厂。
newScheduledThreadPool：支持线程定时操作和周期性操作。

下面是方法的源码：

public class Executors {
    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 ExecutorService newSingleThreadExecutor(ThreadFactory threadFactory) {
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>(),
                                    threadFactory));
    }

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

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

线程池如何关闭？

线程池关闭，可以使用 shutdown() 或 shutdownNow() 方法，它们的区别是：

shutdown()：不会立即终止线程池，而是要等所有任务队列中的任务都执行完后才会终止。执行完 shutdown 方法之后，线程池就不会再接受新任务了。

shutdownNow()：执行该方法，线程池的状态立刻变成 STOP 状态，并试图停止所有正在执行的线程，不再处理还在池队列中等待的任务，执行此方法会返回未执行的任务。

shutdownNow() 和 shutdown() 两个方法有什么区别？

shutdownNow() 和 shutdown() 都是用来终止线程池的，它们的区别是，使用 shutdown() 程序不会报错，也不会立即终止线程，它会等待线程池中的缓存任务执行完之后再退出，执行了 shutdown() 之后就不能给线程池添加新任务了；shutdownNow() 会试图立马停止任务，如果线程池中还有缓存任务正在执行，则会抛出 java.lang.InterruptedException: sleep interrupted 异常。