Java线程池的使用总结

忘是亡心i 2022-06-16 07:45 226阅读 0赞

Java中的线程池是运用场景最多的并发框架， 几乎所有需要异步或并发执行任务的程序都可以使用线程池。 在开发过程中， 合理地使用线程池能够带来3个好处。

第一： 降低资源消耗。 通过重复利用已创建的线程降低线程创建和销毁造成的消耗。  
　　第二： 提高响应速度。 当任务到达时， 任务可以不需要等到线程创建就能立即执行。  
　　第三： 提高线程的可管理性。 线程是稀缺资源， 如果无限制地创建， 不仅会消耗系统资源，还会降低系统的稳定性， 使用线程池可以进行统一分配、 调优和监控。 但是， 要做到合理利用线程池， 必须对其实现原理了如指掌。

## **1、线程池工作流程** ##

当向线程池提交一个任务之后， 线程池是如何处理这个任务的呢？下图展示了线程池的主要处理流程

![/][SouthEast]

从图中可以看出， 当提交一个新任务到线程池时， 线程池的处理流程如下：  
　　1） 线程池判断核心线程池里的线程是否都在执行任务。 如果不是， 则创建一个新的工作线程来执行任务。 如果核心线程池里的线程都在执行任务， 则进入下个流程。  
　　2） 线程池判断工作队列是否已经满。 如果工作队列没有满， 则将新提交的任务存储在这个工作队列里。 如果工作队列满了， 则进入下个流程。  
　　3） 线程池判断线程池的线程是否都处于工作状态。 如果没有，则创建一个新的工作线程来执行任务。 如果已经满了，则交给饱和策略来处理这个任务。

## **2、线程池的使用** ##

Java提供了自己的线程池。每次只执行指定数量的线程，java.util.concurrent.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.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }

创建一个线程池时需要输入几个参数，对参数的解释如下：  
　　1） corePoolSize（线程池的基本大小）：当提交一个任务到线程池时， 线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程， 等到需要执行的任务数大于线程池基本大小时就不再创建。不超过maximumPoolSize值时，线程池中最多有corePoolSize 个线程工作。 如果调用了线程池的prestartAllCoreThreads()方法，线程池会提前创建并启动所有基本线程。

2）runnableTaskQueue（任务队列）： 用于保存等待执行的任务的阻塞队列。可以选择以下几个阻塞队列：  
　　ArrayBlockingQueue：是一个基于数组结构的有界阻塞队列， 此队列按FIFO（先进先出） 原则对元素进行排序；  
　　LinkedBlockingQueue： 一个基于链表结构的阻塞队列， 此队列按FIFO排序元素， 吞吐量通常要高于ArrayBlockingQueue。 静态工厂方法Executors.newFixedThreadPool()使用了这个队列；  
　　SynchronousQueue： 一个不存储元素的阻塞队列。 每个插入操作必须等到另一个线程调用移除操作， 否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列；  
　　PriorityBlockingQueue： 一个具有优先级的无界阻塞队列。

*  直接交接（SynchronousQueue）：任务不多时，只需要用队列进行简单的任务中转，这种队列无法存储任务，在使用这种队列时，需要将maxPoolSize设置的大一点。
 *  无界队列（LinkedBlockingQueue）：如果使用无界队列当作runnableTaskQueue，将maxQueue设置的多大都没有用，使用无界队列的优点是可以防止流量突增，缺点是如果处理任务的速度跟不上提交任务的速度，这样就会导致无界队列中的任务越来越多，从而导致OOM异常。
 *  有界队列（ArrayBlockingQueue）：使用有界队列可以设置队列大小，让线程池的maxPoolSize有意义。

3）maximumPoolSize（线程池最大数量）：线程池允许创建的最大线程数。 如果队列满了， 并且已创建的线程数小于最大线程数， 则线程池会再创建新的线程执行任务。 值得注意的是， 如果使用了无界的任务队列这个参数就没什么效果。

4）ThreadFactory： 用于设置创建线程的工厂， 可以通过线程工厂给每个创建出来的线程设置更有意义的名字。 使用开源框架guava提供的ThreadFactoryBuilder可以快速给线程池里的线程设置有意义的名字。

5）RejectedExecutionHandler（饱和策略）：当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。 这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。在JDK 1.5中Java线程池框架提供了以下4种策略：  
　　AbortPolicy：直接抛出异常；  
　　CallerRunsPolicy：只用调用者所在线程来运行任务；  
　　DiscardOldestPolicy：丢弃队列里最近的一个任务， 并执行当前任务；  
　　DiscardPolicy：不处理，丢弃掉。

6）keepAliveTime（线程活动保持时间）： 线程池的工作线程空闲后， 保持存活的时间。 所以，如果任务很多， 并且每个任务执行的时间比较短，可以调大时间，提高线程的利用率。

7）TimeUnit（线程活动保持时间的单位）：可选的单位有天（DAYS）、小时（HOURS）、分钟（MINUTES）、毫秒（MILLISECONDS）、 微秒（MICROSECONDS， 千分之一毫秒） 和纳秒（NANOSECONDS， 千分之一微秒）。

线程池执行execute() 方法的示意图如下图所示：

![SouthEast 1][]

当调用线程池的 execute() 方法添加一个任务时，线程池会做如下判断：  
　　a. 如果正在运行的线程数量小于 corePoolSize， 则创建新线程来执行任务（注意， 执行这一步骤需要获取全局锁）；  
　　b. 如果正在运行如果运行的线程等于或多于corePoolSize， 则将任务加入BlockingQueue；  
　　c. 如果无法将任务加入BlockingQueue（队列已满），而且正在运行的线程数量小于 maximumPoolSize，则创建新的线程来处理任务（注意， 执行这一步骤需要获取全局锁）；  
　　d. 如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，任务将被拒绝，线程池调用RejectedExecutionHandler.rejectedExecution()方法，执行饱和策略 。

这个过程说明，并不是先加入的任务就一定会先执行。**假设队列大小为 4，corePoolSize为2，maximumPoolSize为6，那么当加入15个任务时，执行的顺序类似这样：首先执行任务 1、2，然后任务3~6被放入队列。这时候队列满了，任务7、8、9、10 会被马上执行，而任务 11~15 则会抛出异常。最终顺序是：1、2、7、8、9、10、3、4、5、6。**当然这个过程是针对指定大小的ArrayBlockingQueue<Runnable>任务队列来说，如果是LinkedBlockingQueue<Runnable>任务队列，初始化时又不指定容量，因为该队列无大小限制(默认限制是Integer.MAX\_VALUE)，所以不存在上述问题。

线程池的任务队列选用LinkedBlockingQueue<Runnable>时，测试代码如下：

import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class ThreadPoolTest implements Runnable { 
    
        @Override
        public void run() { 
            synchronized (this) { 
                System.out.println(Thread.currentThread().getName());
                try { 
                    Thread.sleep(3000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) { 
            BlockingQueue<Runnable> queue = new LinkedBlockingDeque<Runnable>();
            ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 6, 1,
                    TimeUnit.DAYS, queue);
            for (int i = 0; i < 10; i++) { 
                executor.execute(new Thread(new ThreadPoolTest(), "TestThread"
                        .concat("" + i)));
                int threadSize = queue.size();
                System.out.println("线程队列大小为-->" + threadSize);
            }
            executor.shutdown();
        }
    }

运行效果如下图：

![这里写图片描述][SouthEast 2]

可见，共执行了10个任务，有3个任务立即被创建线程，获得执行。7个任务存储在任务队列中（代码中使用了无界的任务队列，参数maximumPoolSize失效，即代码中设置的最大线程数量6无效）。因为是从线程池里运行的线程，所以虽然将线程的名称设为”TestThread”.concat(“”+i)，但输出后还是变成了pool-1-thread-x。

线程池的任务队列选用ArrayBlockingQueue<Runnable>时，测试代码如下：

import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingDeque;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    public class ThreadPoolTest implements Runnable { 
    
        @Override
        public void run() { 
            synchronized (this) { 
                System.out.println(Thread.currentThread().getName());
                try { 
                    Thread.sleep(3000);
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
        }
    
        public static void main(String[] args) { 
            ThreadPoolExecutor executor = new ThreadPoolExecutor(3, 6, 1,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(5), Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
    
            for (int i = 0; i < 12; i++) { 
                executor.execute(new Thread(new ThreadPoolTest(), "TestThread"
                        .concat("" + i)));
                int threadSize = queue.size();
                System.out.println("线程队列大小为-->" + threadSize);
            }
            executor.shutdown();
        }
    }

运行效果如下图：

![这里写图片描述][SouthEast 3]

因为线程池的最大线程数量为6，基本大小为3，有界阻塞队列长度为5，所以线程池能够容纳的任务最多是11个。编号为1，2，3的任务被立刻创建线程执行，然后编号为4，5，6，7，8的任务进阻塞队列。编号9，10，11的任务被立刻创建线程执行（此时阻塞队列已满，而线程数量没有达到线程池允许的最大线程数量）。此时，线程池处于饱和状态。编号为12的任务提交后就抛出异常。

可以通过调用线程池的shutdown()或shutdownNow()方法来关闭线程池。 它们的原理是遍历线程池中的工作线程， 然后逐个调用线程的interrupt()方法来中断线程，所以无法响应中断的任务可能永远无法终止。 但是它们存在一定的区别， shutdownNow()首先将线程池的状态设置成STOP， 然后尝试停止所有的正在执行或暂停任务的线程， 并返回等待执行任务的列表， 而shutdown()只是将线程池的状态设置成SHUTDOWN状态， 然后中断所有没有正在执行任务的线程。

只要调用了这两个关闭方法中的任意一个， isShutdown()方法就会返回true。 当所有的任务都已关闭后， 才表示线程池关闭成功， 这时调用isTerminaed()方法会返回true。 至于应该调用哪一种方法来关闭线程池， 应该由提交到线程池的任务特性决定， 通常调用shutdown()方法来关闭线程池， 如果任务不一定要执行完， 则可以调用shutdownNow()方法。isTerminated()可以判断线程池是否被完全终止了，当调用了shutdown()或shutdownNow()方法中的任意一个后，isShutdown()方法都会返回true但isTerminated()方法只有在线程池完全终止了才会返回true。

ArrayBlockingQueue是一个由数组支持的有界阻塞队列。在读写操作上都需要锁住整个容器，因此吞吐量与一般的实现是相似的，适合于实现“生产者消费者”模式。

LinkedBlockingQueue是基于链表的阻塞队列，默认最大是Integer.MAX\_VALUE。同ArrayListBlockingQueue类似，其内部也维持着一个数据缓冲队列（该队列由一个链表构成），当生产者往队列中放入一个数据时，队列会从生产者手中获取数据，并缓存在队列内部，而生产者立即返回；只有当队列缓冲区达到最大值缓存容量时（LinkedBlockingQueue可以通过构造函数指定该值），才会阻塞生产者队列，直到消费者从队列中消费掉一份数据，生产者线程会被唤醒，反之对于消费者这端的处理也基于同样的原理。其中主要用到put和take方法，put方法在队列满的时候会阻塞直到有队列成员被消费，take方法在队列空的时候会阻塞，直到有队列成员被放进来。而LinkedBlockingQueue之所以能够高效的处理并发数据，还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步，这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据，以此来提高整个队列的并发性能。

## **3、直接调用JDK封装好的线程池会带来的问题** ##

### 3.1 newFixedThreadPool ###

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

newFixedThreadPool线程池通过传入相同的corePoolSize和maxPoolSize可以保证线程数量固定，0的keepAliveTime表示线程一旦空闲立刻被销毁，workQueue使用的是无界队列。

这样潜在的问题就是当处理任务的速度赶不上任务提交的速度的时候，就可能会让大量任务堆积在workQueue中，从而引发OOM异常。

/** * 演示newFixedThreadPool线程池OOM问题 */
    public class FixedThreadPoolOOM { 
    
        private static ExecutorService executorService = Executors.newFixedThreadPool(1);
    
        public static void main(String[] args) { 
            for (int i = 0; i < Integer.MAX_VALUE; i++) { 
                executorService.execute(new SubThread());
            }
        }
    }
    
    class SubThread implements Runnable { 
    
        @Override
        public void run() { 
            try { 
                //延长任务时间
                Thread.sleep(1000000000);
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
        }
    }

更改JVM参数：-Xmx8m -Xms8m  
运行示例，即可发生OutOfMemeryError

### 3.2 newSingleThreadExecutor ###

public static ExecutorService newSingleThreadExecutor() { 
        return new FinalizableDelegatedExecutorService
            (new ThreadPoolExecutor(1, 1,
                                    0L, TimeUnit.MILLISECONDS,
                                    new LinkedBlockingQueue<Runnable>()));
    }

从源码可以看出newSingleThreadExecutor和newFixedThreadPool基本类似，不同的只是corePoolSize和maxPoolSize的值，所以newSingleThreadExecutor也存在内存溢出问题。

### 3.3 newCachedThreadPool ###

newCachedThreadPool也被称为可缓存线程池，它是一个无界线程池，具有自动回收多余线程的功能。

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

newCachedThreadPool的maxPoolSize设置的值为Integer.MAX\_VALUE，所以可能会导致线程被无限创建，最终导致OOM异常。

## **4、合理使用线程池** ##

要想合理地配置线程池， 就必须首先分析任务特性， 可以从以下几个角度来分析：  
　　任务的性质： CPU密集型任务、 IO密集型任务和混合型任务。  
　　任务的优先级： 高、 中和低。  
　　任务的执行时间： 长、 中和短。  
　　任务的依赖性： 是否依赖其他系统资源， 如数据库连接。  
　　性质不同的任务可以用不同规模的线程池分开处理。假设N为设备的CPU个数， CPU密集型任务应配置尽可能小的线程， 如配置N+1个线程的线程池。 由于IO密集型任务线程并不是一直在执行任务， 则应配置尽可能多的线程， 如2\*N。 混合型的任务， 如果可以拆分， 将其拆分成一个CPU密集型任务和一个IO密集型任务， 只要这两个任务执行的时间相差不是太大， 那么分解后执行的吞吐量将高于串行执行的吞吐量。 如果这两个任务执行时间相差太大， 则没必要进行分解。 可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数。

建议使用有界队列。 有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点儿， 比如几千。 有一次， 我们系统里后台任务线程池的队列和线程池全满了，不断抛出抛弃任务的异常， 通过排查发现是数据库出现了问题， 导致执行SQL变得非常缓慢， 因为后台任务线程池里的任务全是需要向数据库查询和插入数据的， 所以导致线程池里的工作线程全部阻塞， 任务积压在线程池里。 如果当时我们设置成无界队列，那么线程池的队列就会越来越多，有可能会撑满内存， 导致整个系统不可用，而不只是后台任务出现问题。 当然， 我们的系统所有的任务是用单独的服务器部署的， 我们使用不同规模的线程池完成不同类型的任务， 但是出现这样问题时也会影响到其他任务。

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

本文内容大部分出自《Java并发编程的艺术》第9章。

[SouthEast]: /images/20220616/5f1d1dfaa12b477199e2701f655d9f44.png
[SouthEast 1]: /images/20220616/3e58b8d1188a4f97915e3c14c2d67f00.png
[SouthEast 2]: /images/20220616/457c6c67b2dc4a8381f14487b947cdc0.png
[SouthEast 3]: /images/20220616/863cf0f8dfcf4327be6fa7b66ff5553d.png