【Java并发】- 10.对ThreadPoolExecutor线程池的简单解析及创建线程池的参数的分析

墨蓝 2023-02-20 03:16 17阅读 0赞

### 文章目录 ###

*   *   *  1 从Executors开始讲起
         *  2.对ThreadPoolExecutor的分析
         *   *  2.1 BlockingQueue workQueue阻塞队列
             *   *  LinkedBlockingQueue：基于链表的阻塞队列
                 *  ArrayBlockingQueue：基于数组的阻塞队列
                 *  SynchronousQueue：同步阻塞队列
             *  2.2 ThreadFactory threadFactory：线程工厂
             *   *  DefaultThreadFactory:默认线程工厂，使用最多的线程工厂
             *  2.3 RejectedExecutionHandler handler：拒绝策略
             *   *  AbortPolicy：放弃策略
                 *  DiscardPolicy
                 *  DiscardOldestPolicy
                 *  CallerRunsPolicy
         *  3 对Executors的再次分析
         *   *  3.1 newFixedThreadPool创建一个大小固定的线程池
             *  3.2 newSingleThreadExecutor：单线程线程池
             *  3.3 newCachedThreadPool：无界线程池
             *  Executors的局限

### 1 从Executors开始讲起 ###

相信我们在学习有关线程池的时候最先接触的很大概率是这个类。这个类提供了一些简单的线程池的实现方法，所以本篇文章从Executors类开始一步步的深挖线程池的具体实现。

下面是一个简单的线程池的所以例子

public class PoolDemo { 
    
        public static void main(String[] args) { 
            ExecutorService threadPool = Executors.newFixedThreadPool(3);
    
            for (int i = 0; i < 3; i++) { 
                threadPool.submit(() -> { 
                    for (int j = 0; j < 3; j++) { 
                        System.out.println(Thread.currentThread().getName() + ":" + j);
                    }
                });
            }
        }
    }

执行结果：

pool-1-thread-1:0
    pool-1-thread-1:1
    pool-1-thread-1:2
    pool-1-thread-3:0
    pool-1-thread-3:1
    pool-1-thread-3:2
    pool-1-thread-2:0
    pool-1-thread-2:1
    pool-1-thread-2:2

Executors可以很轻松的创建一个线程池并执行任务。在Executors中有三类线程池由三种方法实现：

*  固定大小的线程池：Executors.newFixedThreadPool(int nThreads)，自定义大小固定的线程池
 *  大小不限的线程池：Executors.newCachedThreadPool()；有多少任务同时执行就开多少线程去执行任务（线程数最大不超过0x7fffffff）
 *  单例线程池：Executors.newSingleThreadExecutor()只有一个线程的线程池

在创建完线程后使用submit，execute的方法其执行任务。

在讲完Executors中的三种线程池后我们来看看其具体实现：

*  Executors.newFixedThreadPool(int nThreads)

*  Executors.newCachedThreadPool()

*  Executors.newSingleThreadExecutor()

可以看到线程池最终都是由ThreadPoolExecutor类实现的，下面分析一下ThreadPoolExecutor类

### 2.对ThreadPoolExecutor的分析 ###

Executors类生成线程池时都是通过ThreadPoolExecutor的构造方法传入不同的值来实现的。下面我们来讲一讲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;
        }

构造方法中有7个参数，其各自含以为：

*  int corePoolSize：**线程池的核心线程数，线程池会一直维护这些线程，如果线程池处于空闲状态这些线程也不会被回收**。
 *  int maximumPoolSize：**最大线程数，如果同时执行任务过多，超过了corePoolSize的数量，此时线程池就会申请新的线程来处理任务，可以创建的线程数不能超过maximumPoolSize**
 *  long keepAliveTime：**超过corePoolSize的线程在空闲keepAliveTime时间后如果还是空闲状态就会被线程池回收**
 *  TimeUnit unit：**keepAliveTime的时间单位（时，分、秒，毫秒等）**
 *  BlockingQueue workQueue ：**阻塞队列，在同一时间进来的任务过多超过了maximumPoolSize，此时传入的任务不能立即执行，会存放在这个队列中**
 *  ThreadFactory threadFactory：**线程工厂，负责创建新的线程给线程池管理，默认线程工厂所创建的线程都是用户线程且优先级为正常优先级**
 *  RejectedExecutionHandler handler：**拒绝策略，当线程池中的线程忙于处理任务，且阻塞队列也放满了的情况下，后续任务通过阻塞队列处理**

对上述的几个重要参数进行详解：

#### 2.1 BlockingQueue workQueue阻塞队列 ####

就是一个队列，额外的支持一些操作，如获取元素的时候等待队列变为非空；在存储以上的时候将等待空间变为可用。其具体实现使用了大量的CAS，AQS等之前写过的Lock的基础知识。这里我只是对BlockingQueue做简单的介绍，其详细的介绍后续内容中给出。

其具体实现我简单写几个：

##### LinkedBlockingQueue：基于链表的阻塞队列 #####

这是一个有界队基于链表线程的阻塞队列，按照FIFO的方式对元素进行排序，故队头元素就是存放在队列中时间最长的元素；队尾元素是存放在队列时间最短的元素，因为新的元素会被插入到队列1尾端，队列的获取或者检索操作会从队头开始。基于链表的阻塞队列，比基于数组的队列有更高的吞吐量。

其内部维护了一个Node为节点的链表

private transient Node<E> last;
    
    static class Node<E> { 
        E item;
    
        /** * One of: * - the real successor Node * - this Node, meaning the successor is head.next * - null, meaning there is no successor (this is the last node) */
        Node<E> next;
    
        Node(E x) {  item = x; }
    }

其内部还维护了两个ReentrantLock和两个Condition

/** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();
    
    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
    
    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();
    
    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

##### ArrayBlockingQueue：基于数组的阻塞队列 #####

是一个通过数组维护的阻塞队列，其他介绍同LinkedBlockingQueue。只是ArrayBlockingQueue吞吐量比LinkedBlockingQueue低。

其内部维护了一个item数组来存放队列的元素

/** The queued items */
    final Object[] items;

##### SynchronousQueue：同步阻塞队列 #####

这是一个阻塞队列，每一个插入操作必须等待其他线程的移出操作完成才能执行，反之亦然。从这可用直到SynchronousQueue队列只维护了一个元素

#### 2.2 ThreadFactory threadFactory：线程工厂 ####

ThreadFactory：根据需要创建新的线程，使用线程工厂就减少了直接使用*new Thread*，使创建线程更加简便，可用应用自定义其子类实现具体线程的创建。

其最简单的实现就是:

class SimpleThreadFactory implements ThreadFactory { 
       public Thread newThread(Runnable r) { 
         return new Thread(r);
       }
     }

线程工厂使用的最多的子类为Executors.DefaultThreadFactory

##### DefaultThreadFactory:默认线程工厂，使用最多的线程工厂 #####

其实现为：

static class DefaultThreadFactory implements ThreadFactory { 
    	//poolNumber定义为static是要确保不同的线程池使用同一个工厂时，按照同一个基准来对
    	//线程池号自增
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        //threadNumber线程计数为非静态，确保不同线程池使用同一个工厂生成的线程名称都是从1
        //开始
        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)
            	//确保创建的线程优先级都是正常优先级（5）
                t.setPriority(Thread.NORM_PRIORITY);
            return t;
        }
    }

#### 2.3 RejectedExecutionHandler handler：拒绝策略 ####

当任务数量超出阻塞队列时会执行，拒绝策略。还有一种情况，就是线程池执行了shutdown方法时，因为线程池已经关闭了，所以对后续任务也要执行拒绝策略。

RejectedExecutionHandler有四个实现了，代表了四种不同的拒绝策略。

##### AbortPolicy：放弃策略 #####

这个处理器对被拒绝的任务会抛出一个异常

其实现为：

/** * A handler for rejected tasks that throws a * {@code RejectedExecutionException}. */
    public static class AbortPolicy implements RejectedExecutionHandler { 
        /** * Creates an {@code AbortPolicy}. */
        public AbortPolicy() {  }
    
        /** * Always throws RejectedExecutionException. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task * @throws RejectedExecutionException always */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
        	//对被拒绝的任务抛出异常，并且把异常传播给execute方法
        	//RejectedExecutionException：是一个运行期异常
            throw new RejectedExecutionException("Task " + r.toString() +
                                                 " rejected from " +
                                                 e.toString());
        }
    }

对AbortPolicy的使用：

public class PoolDemo2 { 
    
        public static void main(String[] args) { 
            AtomicInteger integer = new AtomicInteger(0);
            ExecutorService pool = new ThreadPoolExecutor(3,5,0l, TimeUnit.MICROSECONDS,
                    new LinkedBlockingDeque<>(3),new ThreadPoolExecutor.AbortPolicy());
    
            for (int i = 0; i < 9; i++) { 
                pool.execute(() -> { 
                    System.out.println(Thread.currentThread().getName() + ": " + integer.incrementAndGet());
    
                });
            }
        }
    }

输出结果：

pool-1-thread-1: 1
    pool-1-thread-1: 3
    pool-1-thread-1: 4
    pool-1-thread-1: 6
    pool-1-thread-2: 2
    pool-1-thread-3: 7
    pool-1-thread-4: 5
    pool-1-thread-5: 8
    Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task wf.threadpool.PoolDemo2$$Lambda$1/1831932724@3b9a45b3 rejected from java.util.concurrent.ThreadPoolExecutor@7699a589[Running, pool size = 5, active threads = 4, queued tasks = 0, completed tasks = 4]
    	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
    	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
    	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
    	at wf.threadpool.PoolDemo2.main(PoolDemo2.java:22)

因为申请的线程池最大线程数为5，阻塞队列能存储的任务为3个，线程池同时能处理8个任务。所以我们给9个任务时最后一个被拒绝了，且抛出了异常。

##### DiscardPolicy #####

会丢弃被拒绝的任务，不做任何处理

实现类：

/** * A handler for rejected tasks that silently discards the * rejected task. */
    public static class DiscardPolicy implements RejectedExecutionHandler { 
        /** * Creates a {@code DiscardPolicy}. */
        public DiscardPolicy() {  }
    
        /** * Does nothing, which has the effect of discarding task r. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
        }
    }

使用代码：

public class PoolDemo2 { 
    
        public static void main(String[] args) { 
            AtomicInteger integer = new AtomicInteger(0);
            ExecutorService pool = new ThreadPoolExecutor(3,5,0l, TimeUnit.MICROSECONDS,
                    new LinkedBlockingDeque<>(3),new ThreadPoolExecutor.DiscardPolicy());
    
            for (int i = 0; i < 9; i++) { 
                pool.execute(() -> { 
                    System.out.println(Thread.currentThread().getName() + ": " + integer.incrementAndGet());
    
                });
            }
        }
    }

输出结果：

pool-1-thread-1: 1
    pool-1-thread-2: 2
    pool-1-thread-3: 4
    pool-1-thread-1: 3
    pool-1-thread-3: 6
    pool-1-thread-2: 5
    pool-1-thread-4: 7
    pool-1-thread-5: 8

最终只执行了8个任务，最后一个任务被丢弃了。

##### DiscardOldestPolicy #####

这个处理器会丢弃最老的未被处理的请求，即阻塞队列的队头元素，并且把被拒绝的任务，执行execute，尝试放入阻塞队列。除非线程池被关闭

/** * A handler for rejected tasks that discards the oldest unhandled * request and then retries {@code execute}, unless the executor * is shut down, in which case the task is discarded. */
    public static class DiscardOldestPolicy implements RejectedExecutionHandler { 
        /** * Creates a {@code DiscardOldestPolicy} for the given executor. */
        public DiscardOldestPolicy() {  }
    
        /** * Obtains and ignores the next task that the executor * would otherwise execute, if one is immediately available, * and then retries execution of task r, unless the executor * is shut down, in which case task r is instead discarded. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
        	//确定线程池不为null
            if (!e.isShutdown()) { 
            	//移出阻塞队列的队头元素
                e.getQueue().poll();
                //对任务执行execute方法
                e.execute(r);
            }
        }
    }
    
    测试代码；
    
    ```cpp
    public class PoolDemo2 { 
    
        public static void main(String[] args) { 
            AtomicInteger integer = new AtomicInteger(0);
            ExecutorService pool = new ThreadPoolExecutor(3,5,0l, TimeUnit.MICROSECONDS,
                    new LinkedBlockingDeque<>(3),new ThreadPoolExecutor.DiscardOldestPolicy());
    
            for (int i = 0; i < 9; i++) { 
                pool.execute(() -> { 
                    System.out.println(Thread.currentThread().getName() + ": " + integer.incrementAndGet());
    
                });
            }
        }
    }

结果：

pool-1-thread-1: 1
    pool-1-thread-2: 3
    pool-1-thread-1: 4
    pool-1-thread-1: 6
    pool-1-thread-5: 7
    pool-1-thread-3: 2
    pool-1-thread-2: 5
    pool-1-thread-4: 8

最后老任务被丢弃，最终执行任务有8个，不过因为没有提示所以看不出来。

##### CallerRunsPolicy #####

会在调用execute方法的线程中直接运行任务的run方法。除非线程池关闭，任务会被直接丢弃。

/** * A handler for rejected tasks that runs the rejected task * directly in the calling thread of the {@code execute} method, * unless the executor has been shut down, in which case the task * is discarded. */
    public static class CallerRunsPolicy implements RejectedExecutionHandler { 
        /** * Creates a {@code CallerRunsPolicy}. */
        public CallerRunsPolicy() {  }
    
        /** * Executes task r in the caller's thread, unless the executor * has been shut down, in which case the task is discarded. * * @param r the runnable task requested to be executed * @param e the executor attempting to execute this task */
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) { 
        	//判断线程池是否执行shutdown方法
            if (!e.isShutdown()) { 
            	//直接调用run方法
                r.run();
            }
        }
    }

这种方法如果任务执行时间过长会阻塞调用者的线程，导致性能问题。

测代码：

public class PoolDemo2 { 
    
        public static void main(String[] args) { 
            AtomicInteger integer = new AtomicInteger(0);
            ExecutorService pool = new ThreadPoolExecutor(3,5,0l, TimeUnit.MICROSECONDS,
                    new LinkedBlockingDeque<>(3),new ThreadPoolExecutor.CallerRunsPolicy());
    
            for (int i = 0; i < 9; i++) { 
                pool.execute(() -> { 
                    System.out.println(Thread.currentThread().getName() + ": " + integer.incrementAndGet());
    
                });
            }
        }
    }

结果：

pool-1-thread-1: 1
    pool-1-thread-4: 5
    main: 4
    pool-1-thread-3: 2
    pool-1-thread-2: 3
    pool-1-thread-3: 9
    pool-1-thread-4: 8
    pool-1-thread-5: 7
    pool-1-thread-1: 6

可以看到任务都被执行了，而被拒绝的任务有主线程执行了。

### 3 对Executors的再次分析 ###

通过上面对线程池的创建的理解我们重新审视一下Executors中的三个线程池

#### 3.1 newFixedThreadPool创建一个大小固定的线程池 ####

public static ExecutorService newFixedThreadPool(int nThreads) { 
        return new ThreadPoolExecutor(nThreads, nThreads,
                                      0L, TimeUnit.MILLISECONDS,
                                      new LinkedBlockingQueue<Runnable>());
    }
    
    //线程池创建
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) { 
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }
    
    //拒绝策略
    private static final RejectedExecutionHandler defaultHandler =
        new AbortPolicy();
    
    //阻塞队列
    public LinkedBlockingQueue() { 
    	//参数固定创建的是一个无解阻塞队列
        this(Integer.MAX_VALUE);
    }
    
    /** * Creates a {@code LinkedBlockingQueue} with the given (fixed) capacity. * * @param capacity the capacity of this queue * @throws IllegalArgumentException if {@code capacity} is not greater * than zero */
    public LinkedBlockingQueue(int capacity) { 
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

因为其corePoolSize和maximumPoolSize都是nThreads，故线程池只能创建nThreads个线程。故keepAliveTime和unit两个属性没有直接给置为0。其采用了一个LinkedBlockingQueue无参构造创建一个无界阻塞队列。其拒绝策略为AbortPolicy，给拒绝的任务抛出异常。

#### 3.2 newSingleThreadExecutor：单线程线程池 ####

public static ExecutorService newSingleThreadExecutor() { 
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }
    
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) { 
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

newSingleThreadExecutor线程池处理线程数只有1外，其他与newFixedThreadPool相同

#### 3.3 newCachedThreadPool：无界线程池 ####

public static ExecutorService newCachedThreadPool() { 
        return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                      60L, TimeUnit.SECONDS,
                                      new SynchronousQueue<Runnable>());
    }
    
    public ThreadPoolExecutor(int corePoolSize,
                              int maximumPoolSize,
                              long keepAliveTime,
                              TimeUnit unit,
                              BlockingQueue<Runnable> workQueue) { 
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
             Executors.defaultThreadFactory(), defaultHandler);
    }

其corePoolSize设置为0，maximumPoolSize设置为Integer.MAX\_VALUE声明，线程池中不维护常驻线程，只有当需要时才会创建线程。而其阻塞队列使用了SynchronousQueue，这个阻塞队列只能维护一个元素，让存入到线程不会阻塞在队列上。

#### Executors的局限 ####

Executors类创建的线程池只能依靠默认的方法来对任务进行处理，导致其局限较大，在特定的环境中，不能较好的完成任务，故在开发对时候如果要使用线程池，最好根据项目需要通过ThreadPoolExecutor自己配置需要的线程池。

而且其使用了无界队列，很有可能在大的任务量情况下导致OOM异常

**线程池使用时最好把偏向锁关闭，使用偏向锁会影响性能。**