ThreadPoolExecutor线程池类使用解析

ゞ浴缸里的玫瑰 2023-07-25 14:24 12阅读 0赞

在Java并发编程中，使用线程池技术对于系统性能的提升是非常大的，通过对线程的统一管理，我们可以重复利用已经创建的线程来执行任务，有效避免频繁的线程创建和销毁带来的资源消耗。

在使用ThreadPoolExecutor之前，先看一下ThreadPoolExecutor的核心方法execute源码，这个方法是线程池执行流程的核心方法，下面这段源码，我在其中附了一些中文注释。

public void execute(Runnable command) {
            if (command == null)
                throw new NullPointerException();
            /*
             * Proceed in 3 steps:
             *
             * 1. If fewer than corePoolSize threads are running, try to
             * start a new thread with the given command as its first
             * task.  The call to addWorker atomically checks runState and
             * workerCount, and so prevents false alarms that would add
             * threads when it shouldn't, by returning false.
             *
             * 2. If a task can be successfully queued, then we still need
             * to double-check whether we should have added a thread
             * (because existing ones died since last checking) or that
             * the pool shut down since entry into this method. So we
             * recheck state and if necessary roll back the enqueuing if
             * stopped, or start a new thread if there are none.
             *
             * 3. If we cannot queue task, then we try to add a new
             * thread.  If it fails, we know we are shut down or saturated
             * and so reject the task.
             */
            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);//创建失败，执行拒绝策略
        }

JDK源码中，包含有作者的大量英文注释，通过源码中的注释可以看出，整个执行过程分为三个步骤

### **Step1:** ###

**当任务提交执行时，如果运行的线程少于核心线程数，则创建核心线程来执行任务。**

### **Step2:** ###

**当核心线程数满时(运行的线程等于核心线程数时)，则将任务加入队列排队。**

### **Step3:** ###

**如果我们无法将任务进行排队(队列已满或线程池关闭)，则创建非核心线程来执行任务，创建失败时执行拒绝策略。**

下面通过一个Demo来了解ThreadPoolExecutor如何使用

定义一个类实现Runable接口用于模拟任务执行

public class MyRunnable implements Runnable {
    
        private String cmd;
    
        MyRunnable(String cmd){
            this.cmd = cmd;
        }
    
        @Override
        public void run() {
            process();
            System.out.println(new Date() + Thread.currentThread().getName() + "执行" + cmd + "任务完成");
        }
    
        private void process(){
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

定义一个线程池类进行测试

public class ThreadPool {
    
        private static int corePoolSize = 4;
        private static int maximumPoolSize = 6;
        private static int keepAliveTime = 1000;
        private static int capacity = 2;
        public static void main(String[] args) throws Exception {
            ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
                    corePoolSize,
                    maximumPoolSize,
                    keepAliveTime,
                    TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(capacity),
                    new ThreadPoolExecutor.AbortPolicy()
            );
    
            for (int i=0; i<10; i++){
                MyRunnable runnable = new MyRunnable("任务" + i);
                poolExecutor.execute(runnable);
            }
            System.out.println("剩余容量：" + poolExecutor.getQueue().remainingCapacity());
        }
    }

运行上面的main方法，会发现只有8个任务执行完成(实际上我们加入了10个任务)，并且抛出了异常。

先来解释一下在这个Demo中使用的构造线程池的一些参数

### **构造参数** ###

**corePoolSize(核心线程数)**

线程池中的核心线程数，也是线程池中保持活跃的线程数

**maximumPoolSize(线程池中的最大线程数)**

线程池的最大线程数，其中包含核心线程数

**keepAliveTime(保持活跃时间)**

当超过核心线程时，其余线程的空闲等待时间，如果非核心线程超出这个时间还没有新的任务，则销毁这个非核心线程

**TimeUnit.MILLISECONDS(保持活跃时间的时间单位)**

在Demo中我用的是TimeUnit.MILLISECONDS毫秒为单位，在ThreadPoolExecutor中，实际还有NANOSECONDS(纳秒)、MICROSECONDS(微秒)、SECONDS(秒)、MINUTES(分钟)等参数

**new ArrayBlockingQueue<>(capacity)**

这个是线程池中的工作队列workQueue，工作队列有很多种，文章下面会解释其他工作队列，这里用到的是BlockingQueue的一个实现类，数组阻塞队列，创建一个容量为2的队列。

**new ThreadPoolExecutor.AbortPolicy()**

拒绝策略参数，Demo使用了ThreadPoolExecutor提供的AbortPolicy(中止策略)拒绝策略，当任务数超过最大线程数时，将会执行此策略，抛出异常，并丢弃该任务。除了这种拒绝策略，ThreadPoolExecutor还提供了其他拒绝策略，后面介绍。

了解了上面的参数，再来分析一下上面Demo的运行结果

**在Demo中，我们创建了10个任务，因为核心线程数为4，所以第1-4个任务线程池将会创建核心线程来执行任务，而第5-6两个任务在执行时，发现4个核心线程已满并且都在执行任务，此时这两个任务将加入到队列中等待(因为我们设置的队列容量为2)。如果此时不再创建任务，队列中的任务会在核心线程任务执行完后，再将队列中的任务交给核心线程执行，但是如果此时还有其他任务提交呢？这时核心线程已满，队列也已经占满，那么第7-8两个任务将会直接创建非核心线程来执行，因为我们最大线程数maximumPoolSize设置为6，其中包含了4个核心线程，只有2个为非核心线程，也就是当前最多加入8个任务(maximumPoolSize+队列容量)。当第9个任务提交执行时，已经没有其他线程可以帮它执行了，于是将执行拒绝策略，因为我们使用了AbortPolicy中止策略，所以任务会被丢弃，抛出异常。**

到这里，对于ThreadPoolExecutor的基本执行过程应该已经比较了解了，下面介绍线程池用到的其他工作队列和拒绝策略。

### **其他工作队列** ###

**ArrayBlockingQueue(数组阻塞队列)**

指定队列大小，在新的任务加入线程池时，如果corePoolSize未满则创建线程执行，如果已满则加入阻塞队列等待，如果阻塞队列也满了，则利用最大线程数maximumPoolSize中的非核心线程来创建，若最大线程数也占满了，则执行拒绝策略。

**LinkedBlockingQueue(链表阻塞队列)**

无界队列，在新的任务加入线程池时，如果corePoolSize未满则创建线程执行，如果已满则加入队列，因为这是个无界队列，所以就算核心线程满了，也不会利用最大线程数中的非核心线程去创建线程执行，这意味着maximumPoolSize的值也就无效了。

**SynchronousQueue(同步队列)**

SynchronousQueue队列没有容量，在新的任务加入线程池时，它会立即交给corePoolSize核心线程执行，如果corePoolSize满了，则尝试创建新线程，若任务数超过了最大线程数则执行拒绝策略，也就是使用这种队列，它不会将任务存于队列中，因为它是没有容量的。

**PriorityBlockingQueue(优先级阻塞队列)**

一种特殊的无界队列，ArrayBlockingQueue和LinkedBlockingQueue都是先进先出的队列，而它会根据任务的优先级来决定任务的执行顺序。

### **线程池提供的四种拒绝策略** ###

**AbortPolicy(中止策略)**

丢弃任务，并抛出异常。

**DiscardPolicy(抛弃策略)**

丢弃任务，不抛出异常。

**DiscardOldestPolicy(古老的抛弃策略？从名字大概可以看出它会丢弃最先/最老的任务，再加入当前任务)**

丢弃队列中最早加入的任务，将当前任务加入队列。

**CallerRunsPolicy(调用者运行策略)**

由调用者运行该任务，如Demo中使用这种策略，将由main主线程执行任务。

**自定义拒绝策略**

除了ThreadPoolExecutor提供的这几种拒绝策略外，我们还可以自定义拒绝策略

通过RejectedExecutionHandler接口实现其中的rejectedExecution方法即可

例如：

new RejectedExecutionHandler() {
            @Override
            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.out.println("拒绝了一个任务");
            }
    }