JUC之Executors

悠悠 2023-01-23 14:53 208阅读 0赞

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70][]

在《[JUC之ThreadPoolExecutor（1）总体介绍][JUC_ThreadPoolExecutor_1]》与《[JUC之ThreadPoolExecutor（2）方法介绍][JUC_ThreadPoolExecutor_2]》中介绍了**ThreadPoolExecutor**。使用ThreadPoolExecutor创建线程池，实现线程池的相关功能。

通过继承图可以看到，ThreadPoolExecutor的最顶级父类为接口**Executor**。因此ThreadPoolExecutor也可以转换成Executor类型。

![20210715203848190.png][]

本文介绍**Executors**，Executors可以看做是Executor的工作类（可以对比**Collection**接口与**Collections**工具类）。在Executors中，借助于ThreadPoolExecutor已经实现了几种不同类型的ThreadPoolExecutor，以便我们方便地在开发中直接使用。通过Executors提供的方法可以很快的创建ThreadPoolExecutor对象。

--------------------

# **█ 构造器** #

private Executors() {}

Executors中的构造器访问权限为private，所以在外部无法正常地通过new Executors()来创建Executors对象。在Executors中的方法都是static静态方法，因此可以直接通过Executors.静态方法来调用，比如：Executors.newSingleThreadExecutor();

--------------------

# **█** 不同类型的ExecutorService #

Executors提供了多种获取来创建不同类型的**ExecutorService**。为什么方法返回值不是ThreadPoolExecutor类型，因为ThreadPoolExecutor实现了ExecutorService接口，因此ThreadPoolExecutor也是ExecutorService类型的。而且Executors中创建的线程池不一定是借助了ThreadPoolExecutor，而是依赖于其他的Executor，但他们都有共同的父类ExecutorService。

*  **newFixedThreadPool**

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

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

ThreadPoolExecutor的构造器提供了7大核心参数，通过对参数的设置可以实现不同类型的ThreadPoolExecutor。

通过**new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());**的参数设置可以看出此种类型的ThreadPoolExecutor有以下几种特点：

①核心worker的数量与最大worker的数量相同。

核心worker与最大worker数量相同，而核心worker数量与临时工worker数量相加之和就是最大worker数量。所以，此种类型的ThreadPoolExecutor是只有核心worker，没有临时工worker的。于是当阻塞队列已经满了的时候，是不会创建临时工worker的，而是会直接执行拒绝策略。因为此种类型的ThreadPoolExecutor只创建核心worker，没有临时工worker，而核心worker被创建之后是固定的，不会被回收，因此结合着这个特点，此种类型的ThreadPoolExecutor便是固定大小worker数量的，所以叫newFixedThreadPool，Fixed意思便是固定的。

②空闲时间为0

因为没有临时工worker，所以空闲时间就设置成0即也就相当于和没有设置一样。因为空闲时间是用来设置回收临时工worker的时间，又因为没有临时工worker，因此就不需要回收了，自然这个空闲时间就没有设置意义了。

③**LinkedBlockingQueue**

阻塞队列的类型为LinkedBlockingQueue，LinkedBlockingQueue的特点是使用链表来实现队列的，因为在创建LinkedBlockingQueue的时候并没有指定队列的最大容量（如new LinkedBlockingQueue(100)，指定Queue的最大容量为100），因此这里的LinkedBlockingQueue的没有边界的，可以一直往里面添加元素，会造成内存不够用而发生内存溢出的异常，正是存在这个内存问题，所以阿里巴巴技术手册中建议不要直接使用Executors提供的相关方法来创建线程池，而是通过手动构建ThreadPoolExecutor对象来创建线程池，因为手动构建地时候可以设置阻塞队列的最大容量。*（关于LinkedBlockingQueue的介绍，请戳《[BlockingQueue之LinkedBlockingQueue][BlockingQueue_LinkedBlockingQueue]》）*

*  **newCachedThreadPool**

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

依然从创建ThreadPoolExecutor对象的传参来看该类型的ThreadPoolExecutor的特点。

①核心worker数为0，最大worker数为Integer.MAX\_VALUE。

与newFixedThreadPool刚好相反，此种类型的ThreadPoolExecutor是没有核心worker，只有临时工worker的。因此新提交的任务会先放进阻塞队列中，等阻塞队列满了之后再创建临时worker来执行等待队列中的任务。

②空闲时间为60。

当创建的临时工worker在长达60秒没有执行任务了，线程池便会将该临时工worker回收了。

③**SynchronousQueue**。

阻塞队列的类型是SynchronousQueue，SynchronousQueue的具有生产者-消费者模型的特点的阻塞队列，每次队列中只能添加一个元素，当添加一个元素到队列后，只有等该元素被从队列中移出后，才能继续向队列中添加元素。

*  **newSingleThreadExecutor**

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

①核心worker与最大worker数量都为1。

此类型的ThreadPoolExecutor始终只有一个核心worker在执行任务。同时也不会创建临时工worker。

②空闲时间为0。

此设置和newFixedThreadPool含义相同。

③LinkedBlockingQueue

创建的ThreadPoolExecutor，其中的BlockingQueue仍然没有指定最大容量，所以也是没有边界的阻塞队列。

**FinalizableDelegatedExecutorService**：

newSingleThreadExecutor没有直接使用ThreadPoolExecutor，而是在外面又封装了一层FinalizableDelegatedExecutorService。FinalizableDelegatedExecutorService继承自**DelegatedExecutorService**。DelegatedExecutorService和ThreadPoolExecutor具有相同的父类继承关系。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1][]

*  **newScheduledThreadPool**

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

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

**ScheduledThreadPoolExecutor**具有定时执行任务功能的线程池。

public class ScheduledThreadPoolExecutor
            extends ThreadPoolExecutor
            implements ScheduledExecutorService

ScheduledThreadPoolExecutor继承了ThreadPoolExecutor，因此其具有普通线程池的功能。

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

super构造器即调用了ThreadPoolExecutor的构造器。从参数设置上面来看，ScheduledThreadPoolExecutor可以指定核心worker的数量，最大worker数量为Integer.MAX\_VALUE，并且阻塞队列类型为**DelayedWorkQueue**。

*（关于ScheduledThreadPoolExecutor的具体介绍，将在后续专门写一篇文章来介绍。在此先对它有个简单的印象。）*

*  **newSingleThreadScheduledExecutor**

与newScheduledThreadPool相同，newSingleThreadScheduledExecutor也是具有定时执行任务功能的。只不过不同于newScheduledThreadPool可以设置多个核心worker，newSingleThreadScheduledExecutor只有一个核心worker。

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1));
    }

public static ScheduledExecutorService newSingleThreadScheduledExecutor(ThreadFactory threadFactory) {
        return new DelegatedScheduledExecutorService
            (new ScheduledThreadPoolExecutor(1, threadFactory));
    }

**DelegatedScheduledExecutorService：**

实现newSingleThreadScheduledExecutor，在外包装了一层**DelegatedScheduledExecutorService**。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2][]

*  **newWorkStealingPool**

可以提供并行运行的线程的线程池。

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

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

newWorkStealingPool的实现借助于**ForkJoinPool**，关于ForkJoinPool的功能介绍将在后续文章中介绍。ForkJoinPool，Fork的意思的分支分叉，Join的意思的汇聚，ForkJoin就是先分叉再汇聚，比如在执行计算结果总和的过程中，对于不需要相互依赖不相关的几个计算过程，可以创建多个线程分别去执行某个计算过程，然后再统一将每个线程执行的结果汇聚（相加）起来。

*  **unconfigurableExecutorService**

将某个ExecutorService包装成另一个ExecutorService对象，这里的另一个对象就是DelegatedExecutorService。

public static ExecutorService unconfigurableExecutorService(ExecutorService executor) {
        if (executor == null)
            throw new NullPointerException();
        return new DelegatedExecutorService(executor);
    }

在调用DelegatedExecutorService的相关方法时，实际还是调用的被包装的ExecutorService。如下面的代码所示，在调用DelegatedExecutorService的submit时候，实际是转发给e.submit方法。

public Future<?> submit(Runnable task) {
        return e.submit(task);
    }

e就是被包装的ExecutorService，在DelegatedExecutorService的构造器中进行初始化。

private final ExecutorService e;
    DelegatedExecutorService(ExecutorService executor) { e = executor; }

*  **unconfigurableScheduledExecutorService**

作用和unconfigurableExecutorService相同，只不过unconfigurableScheduledExecutorService包装的是具有定时执行任务功能的ScheduledExecutorService。

public static ScheduledExecutorService unconfigurableScheduledExecutorService(ScheduledExecutorService executor) {
        if (executor == null)
            throw new NullPointerException();
        return new DelegatedScheduledExecutorService(executor);
    }

--------------------

# **█ 其他方法** #

*  **defaultThreadFactory**

public static ThreadFactory defaultThreadFactory() {
        return new DefaultThreadFactory();
    }

创建默认的线程工厂类，在创建ThreadPoolExecutor线程池中，如果没有指定线程工厂，就会使用这个默认的线程工厂。如下在ThreadPoolExecutor的构造器中通过调用**Executors.defaultThreadFactory()**来获取线程工厂。

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

**DefaultThreadFactory：**

static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;
    
        DefaultThreadFactory() {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :
                                  Thread.currentThread().getThreadGroup();
            namePrefix = "pool-" +
                          poolNumber.getAndIncrement() +
                         "-thread-";
        }
    
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r,
                                  namePrefix + threadNumber.getAndIncrement(),
                                  0);
            if (t.isDaemon())
                t.setDaemon(false);
            if (t.getPriority() != Thread.NORM_PRIORITY)
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

*  **privilegedThreadFactory**

获取带有权限控制功能的线程工厂。

public static ThreadFactory privilegedThreadFactory() {
        return new PrivilegedThreadFactory();
    }

**PrivilegedThreadFactory：**

static class PrivilegedThreadFactory extends DefaultThreadFactory {
        private final AccessControlContext acc;
        private final ClassLoader ccl;
    
        PrivilegedThreadFactory() {
            super();
            SecurityManager sm = System.getSecurityManager();
            if (sm != null) {
                // Calls to getContextClassLoader from this class
                // never trigger a security check, but we check
                // whether our callers have this permission anyways.
                sm.checkPermission(SecurityConstants.GET_CLASSLOADER_PERMISSION);
    
                // Fail fast
                sm.checkPermission(new RuntimePermission("setContextClassLoader"));
            }
            this.acc = AccessController.getContext();
            this.ccl = Thread.currentThread().getContextClassLoader();
        }
    
        public Thread newThread(final Runnable r) {
            return super.newThread(new Runnable() {
                public void run() {
                    AccessController.doPrivileged(new PrivilegedAction<Void>() {
                        public Void run() {
                            Thread.currentThread().setContextClassLoader(ccl);
                            r.run();
                            return null;
                        }
                    }, acc);
                }
            });
        }
    }

*  **callable**

提供了一个方法将**Runnable**转换成**Callable**类型。

public static <T> Callable<T> callable(Runnable task, T result) {
        if (task == null)
            throw new NullPointerException();
        return new RunnableAdapter<T>(task, result);
    }

Runnable是不带有返回值结果的，而Callable是带有返回值结果的。所以可以通过第二个参数T result来指定，当Runnable类型的task执行完成之后，Callable的call方法返回的结果。

**RunnableAdapter：**

static final class RunnableAdapter<T> implements Callable<T> {
        final Runnable task;
        final T result;
        RunnableAdapter(Runnable task, T result) {
            this.task = task;
            this.result = result;
        }
        public T call() {
            task.run();
            return result;
        }
    }

callable方法的重载：

public static Callable<Object> callable(Runnable task) {
        if (task == null)
            throw new NullPointerException();
        return new RunnableAdapter<Object>(task, null);
    }

在此方法中，设置的T result参数为null，所以当Runnable task执行完成后，Callable的call方法返回值null。

*  **privilegedCallable**

增加了权限控制功能的转换。对参数传进来的Callable增加权限控制功能的调用。

public static <T> Callable<T> privilegedCallable(Callable<T> callable) {
        if (callable == null)
            throw new NullPointerException();
        return new PrivilegedCallable<T>(callable);
    }

**PrivilegedCallable：**

static final class PrivilegedCallable<T> implements Callable<T> {
        private final Callable<T> task;
        private final AccessControlContext acc;
    
        PrivilegedCallable(Callable<T> task) {
            this.task = task;
            this.acc = AccessController.getContext();
        }
    
        public T call() throws Exception {
            try {
                return AccessController.doPrivileged(
                    new PrivilegedExceptionAction<T>() {
                        public T run() throws Exception {
                            return task.call();
                        }
                    }, acc);
            } catch (PrivilegedActionException e) {
                throw e.getException();
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70]: /images/20221004/312a2079db9c4dc39c989f794a2e92f2.png
[JUC_ThreadPoolExecutor_1]: https://blog.csdn.net/weixin_50518271/article/details/118684239?spm=1001.2014.3001.5501
[JUC_ThreadPoolExecutor_2]: https://blog.csdn.net/weixin_50518271/article/details/118735011?spm=1001.2014.3001.5501
[20210715203848190.png]: /images/20221004/d3554b1693fc4b85a5aeadb2365b392d.png
[BlockingQueue_LinkedBlockingQueue]: https://blog.csdn.net/weixin_50518271/article/details/118195710?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 1]: /images/20221004/a3f4a3b366be4e2dae135f56d77d7ac7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl81MDUxODI3MQ_size_16_color_FFFFFF_t_70 2]: /images/20221004/2a9fd0fef85b46f68ee5cbda65a2818c.png