Java 线程池

桃扇骨 2023-10-09 14:13 3阅读 0赞

#### 文章目录 ####

*  Java线程池
 *   *  一、为什么需要线程池
     *  二、什么是线程池
     *  三、线程池的架构
     *  四、Executors创建常见的线程池
     *   *  1、newFixedThreadPool:固定数量的线程池
         *  2、newCachedThreadPool：可缓存工作线程线程池
         *  3、newSingleThreadExecutor：单个线程的线程池
         *  4、newScheduledThreadPool：周期性执行任务线程池
     *  五、ScheduledThreadPoolExecutor
     *  六、ThreadPoolExecutor
     *   *  ThreadPoolExecutor线程池执行任务流程

## Java线程池 ##

### 一、为什么需要线程池 ###

多线程技术主要解决处理器单元内多个线程执行的问题，可以减少处理器单元的闲置时间，增加处理器单元吞吐能力，多线程的情况下确实可以最大限度发挥多核处理器的计算能力，但是如果随意使用线程池，对系统性能反而不利。

*  **创建和销毁线程是需要时间：假如一个服务器完成意向任务所需要的时间为：T1表示线程创建时间，T2表示线程执行任务时间，T3表示销毁时间，如果T1+T3远大于T2，那么会得不偿失**
 *  **线程也需要占用内存空闲，大量的线程会抢占宝贵的内存资源，可能会导致OOM的异常**
 *  **大量的线程回收也会给GC带来很大的压力，延长GC停顿的时间**
 *  **大量的线程也会抢占CPU资源，CPU不停的在各个线程上下文切换，反而没有时间去处理线程运行的时候该处理的任务**

### 二、什么是线程池 ###

> `线程池`**就是实现创建若干的可执行的线程放入一个池中，需要的时候从池中获取线程，不用自行创建，使用完成不需要销毁线程而放入线程池中，从而减少创建和销毁对象的开销。**  
> 因此通过池资源来避免频繁的创建和销毁线程，让创建的线程进行复用，就有了线程池的概念，线程池里会维护一部分活跃的线程，如果有需要，就去线程池中取出线程使用，用完归还到线程池中，免去创建和销毁线程的开销，且线程池也会对线程的数量有一定的限制，线程池的本质是对线程资源的复用。

**线程池的优势：**

**1、****降低资源的消耗**，通过重复利用已创建的线程降低线程创建和销毁造成的消耗  
**2、****提高响应速度**，当任务到达时，任务可以不需要等待线程创建就可以立即执行  
**3、****提高了线程的可管理性**，不仅可以降低系统资源消耗，提高了系统稳定性  
**4、****线程池可以实现统一的分配，调优和监控**

### 三、线程池的架构 ###

![在这里插入图片描述][6dd365597087415bb0ec1b193c5b7d75.png]

> `Executor`接口是最基础的接口。  
> `ExecutorService`接口继承了**Executor**，在其上添加了一些扩展方法，可以说是真正的线程池接口。  
> `AbstractExecutorService`抽象类实现了**ExecutorService**中的大部分接口功能。  
> `ThreadPoolExecutor`继承了**AbstractExecutorService**，是线程池的具体实现。  
> `ScheduledExecutorService`接口继承自**ExecutorService**接口，提供了"周期性执行"的功能。  
> `ScheduledThreadPoolExecutor`类即继承自**ThreadPoolExecutor**由实现了**ScheduledExecutorService**接口，是”带有周期性执行“功能的线程池Executors是线程池的静态工厂，提供了快速创建线程池的静态方法。

**Executor接口**  
**提交任务接口，只提供了一个execute方法，执行Runable类型任务**

public interface Executor {
        
        void execute(Runnable command);
    }

**ExecutorService接口**

public interface ExecutorService extends Executor {
        
        //关闭线程池
        void shutdown();
        //立即关闭
        List<Runnable> shutdownNow();
        //是否关闭
        boolean isShutdown();
        //是否终止
        boolean isTerminated();
        boolean awaitTermination(long timeout, TimeUnit unit)
            throws InterruptedException;
        //提交任务 ，提交任务类型多样，还具有异步功能
        <T> Future<T> submit(Callable<T> task);
        <T> Future<T> submit(Runnable task, T result);
        Future<?> submit(Runnable task);
        
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
            throws InterruptedException;
        <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
                                      long timeout, TimeUnit unit)
            throws InterruptedException;
        <T> T invokeAny(Collection<? extends Callable<T>> tasks)
            throws InterruptedException, ExecutionException;
    
        <T> T invokeAny(Collection<? extends Callable<T>> tasks,
                        long timeout, TimeUnit unit)
            throws InterruptedException, ExecutionException, TimeoutException;
    }

**`ExecutorService`接口是真正线程池的接口，在Executor基础上做了一些扩展，主要是提交任务，终止任务。**

**ScheduledExecutorService接口**  
提供了具有周期性执行任务的方法

public interface ScheduledExecutorService extends ExecutorService {
        
        public ScheduledFuture<?> schedule(Runnable command,
                                           long delay, TimeUnit unit);
        public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                               long delay, TimeUnit unit);
    
        public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                      long initialDelay,
                                                      long period,
                                                      TimeUnit unit);
    
        public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                         long initialDelay,
                                                         long delay,
                                                         TimeUnit unit);
    
    }

### 四、Executors创建常见的线程池 ###

**Executor工厂类提供线程初始化接口，主要有如下几种：**

#### 1、newFixedThreadPool:固定数量的线程池 ####

**代码示例：**

public class ExecutorsTest01 {
        
    
        public static void main(String[] args) {
        
            AtomicInteger num = new AtomicInteger();
            ExecutorService executorService = Executors.newFixedThreadPool(3);
            for (int i = 0; i < 10; i++) {
        
                executorService.submit(new Runnable() {
        
                    @Override
                    public void run() {
        
                        Random random = new Random();
                        System.out.println("线程名：" + Thread.currentThread().getName() + " 执行任务：" + num.getAndIncrement());
                        try {
        
                            Thread.sleep(random.nextInt(100));
                        } catch (InterruptedException e) {
        
                            throw new RuntimeException(e);
                        }
                    }
                });
            }
        }
    }

**执行结果：**  
![在这里插入图片描述][5f9c4ae1b01c455a88f248423793b8fa.png]  
**固定线程数量线程池实现：**

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

**创建一个指定工作线程数的线程池，其中corepoolsize和maxnumPoolSize相等，阻塞队列是基于LinkedBlockingQueue实现的**  
`FixedThreadPool`：传入的核心线程数是固定的，所有称为有界线程池，最大线程数和核心线程数相等。  
假设核心线程是3，一次性提交10个任务，先启动了3个线程执行3个任务，剩下7个任务进入阻塞队列，因为核心线程和最大线程数相等，所以`keepAlivetime`参数没有意义，等待任一线程执行结束就会继续从等待队列中获取一个任务进行执行。

**固定线程数量的线程池可以提供程序效率和节省创建线程所消耗的时间开销。**

#### 2、newCachedThreadPool：可缓存工作线程线程池 ####

**代码示例：**

public class ExecutorsTest02 {
        
        public static void main(String[] args) {
        
            ExecutorService executorService = Executors.newCachedThreadPool();
            AtomicInteger num = new AtomicInteger();
            for (int i = 0; i < 10; i++) {
        
                Random random = new Random();
                executorService.submit(new Runnable() {
        
                    @Override
                    public void run() {
        
                        System.out.println("线程名：" + Thread.currentThread().getName() + " 执行任务：" + num.getAndIncrement());
                        try {
        
                            Thread.sleep(random.nextInt(100));
                        } catch (InterruptedException e) {
        
                            throw new RuntimeException(e);
                        }
                    }
                });
            }
        }
    }

**执行结果：**  
![在这里插入图片描述][419674a560df43f7b91a02566f02e923.png]

**可缓存工作线程线程池实现：**

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

创建一个可缓存的工作线程池，核心线程是0，线程池中线程数可达**Integer.MAX\_VALUE**，即**2^32**，线程默认存活时间**60秒**，内部使用的队列SynchronousQueue同步阻塞队列，在没有任务执行时，当线程空闲时间超过60秒，则工作线程将会终止，当提交新任务时，如果没有空闲线程，则创建新线程执行任务。

可缓存线程池： 如果有新的任务且没有可用的空闲线程，则新建线程，如果线程长时间空闲（60s）则对线程进行回收，此线程池不会对线程池大小做线程，**线程池大小完全依赖于操作系统能够创建的最大线程大小**。  
可缓存线程池： 可缓存线程池应用场景**适合耗时短**，不需要考虑同步的场合。

#### 3、newSingleThreadExecutor：单个线程的线程池 ####

**代码示例：**

public class ExecutorsTest03 {
        
        public static void main(String[] args) {
        
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            AtomicInteger num = new AtomicInteger();
    
            for (int i = 0; i < 10; i++) {
        
                executorService.submit(new Runnable() {
        
                    @Override
                    public void run() {
        
                        Random random = new Random();
                        System.out.println("线程名：" + Thread.currentThread().getName() + " 执行任务：" + num.getAndIncrement());
                        try {
        
                            Thread.sleep(random.nextInt(100));
                        } catch (InterruptedException e) {
        
                            throw new RuntimeException(e);
                        }
                    }
                });
            }
        }
    }

**执行结果：**  
![在这里插入图片描述][7a9c3c2b3c0c4460a74cae55389c19a1.png]  
**线程池中只有一个线程来执行任务**

**单线程线程池底层实现：**

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

单线程线程池中只有一个线程，如果该线程异常结束，会重新创建一个新的线程继续执行任务，唯一的一个线程可以保证提交任务的顺序执行，内部使用**LinkedBlockingQueue**作为阻塞队列

> `newFixedThreadPool(1)`固定数量的线程的线程池在给定参数为1的情况下就可以看做是`newSingleThreadExecutor`。

#### 4、newScheduledThreadPool：周期性执行任务线程池 ####

**代码示例：**

**DIYRunnable类**

public class DIYRunnable implements Runnable {
        
        private Integer num;
    
        public DIYRunnable(Integer num) {
        
            this.num = num;
        }
    
        @Override
        public void run() {
        
            System.out.println("线程名：" + Thread.currentThread().getName() + " 执行任务编号：" + num + " 当前时间：" + System.currentTimeMillis());
        }
    }

**测试类**

public class ExecutorsTest04 {
        
        public static void main(String[] args) {
        
            ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(3);
            for (int i = 0; i < 10; i++) {
        
                scheduledExecutorService.scheduleWithFixedDelay(new DIYRunnable(i),2,2, TimeUnit.SECONDS);
            }
        }
    }

**执行结果：**  
![在这里插入图片描述][a93660797db745f9ac2a2069449c266d.png]  
**线程池可以在执行时间间隔内周期性执行所提交的任务，在实现的业务场景中可以使用该线程时定期的同步数**

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

**Executors的慎用**

**【强制】线程池不允许使用 Executors 去创建，而是通过ThreadPoolExecutor 的方式，这样的处理方式让写的同学更加明确线程池的运行规则，规避资源耗尽的风险。**

**Executors 返回的线程池对象的弊端如下:**

*  `FixedThreadPool`和 `SingleThreadPool` : 允许的请求队列长度为 **Integer.MAX\_VALUE**，可能会堆积大量的请求，从而导致 **OOM**。
 *  `CachedThreadPool`和 `ScheduledThreadPool`: 允许的创建线程数量为 **Integer.MAX\_VALUE**，可能会创建大量的线程，从而导致 **OOM**。

### 五、ScheduledThreadPoolExecutor ###

**ScheduledThreadPoolExecutor:周期性线程池**  
提供了**两个**主要方法：

/**
    * 创建并执行一个在给定初始延迟后首次启用的定期操作，后续操作具有给定的周期
    * 也就是将在 initialDelay 后开始执行，然后在 initialDelay+period 后执行，接着在 initialDelay + 2 * period 后执行，依此类推
    * 如果执行任务发生异常，随后的任务将被禁止，否则任务只会在被取消或者Executor被终止后停止
    * 如果任何执行的任务超过了周期，随后的执行会延时，不会并发执行
    */
    public ScheduledFuture<?> scheduleAtFixedRate(
        Runnable command,
        long initialDelay,
        long period,
        TimeUnit unit);
    
    /**
    * 创建并执行一个在给定初始延迟后首次启用的定期操作，随后，在每一次执行终止和下一次执行开始之间都存在给定的延迟
    * 如果执行任务发生异常，随后的任务将被禁止，否则任务只会在被取消或者Executor被终止后停止
    */
    public ScheduledFuture<?> scheduleWithFixedDelay(
        Runnable command,
        long initialDelay,
        long delay,
        TimeUnit unit);

其中schedule方法用于单次调度执行任务。这里主要理解下后面两个方法:

*  `scheduleAtFixedRate`：该方法在**initialDelay**时长后第一次执行任务，以后每隔**period**时长，再次执行任务。注意，period是从任务开始执行算起的。开始执行任务后，定时器每隔period时长检查该任务是否完成，如果完成则再次启动任务，否则等该任务结束后才再次启动任务，看下图示例  
    ![在这里插入图片描述][fec7ac3767bd46fbbfc72be2ef2cea6e.png]

● `scheduleWithFixDelay`：该方法在initialDelay时长后第一次执行任务，以后每当任务执行完成后，等待delay时长，再次执行任务，看下图示例。  
![在这里插入图片描述][efce0eff743848f4b3da5856a39125f4.png]

### 六、ThreadPoolExecutor ###

**ThreadPoolExecutor构造参数**

public ThreadPoolExecutor(int corePoolSize,
                                  int maximumPoolSize,
                                  long keepAliveTime,
                                  TimeUnit unit,
                                  BlockingQueue<Runnable> workQueue,
                                  ThreadFactory threadFactory,
                                  RejectedExecutionHandler handler) {

*  **corePoolSize：核心线程数量**  
    **线程池中的核心线程数，当提交一个任务时，线程池创建一个新线程执行任务，直到当前线程数等于corePoolSize，如果当前线程数为corePoolSize，继续提交的任务会被存放到阻塞队列中，等待被执行**
 *  **maximumPoolSize：最大线程数量**  
    **线程池中允许的最大的线程数，如果当前阻塞队列满了，且继续提交任务，则创建新的线程执行任务，前提是当前的线程小于maximumPoolSize**
 *  **keepAliveTime：空闲线程存活时间**  
    **线程空闲时存活时间，即当线程没有任务执行时，线程会继续存活keepAliveTime，结合TimeUnit给定时间单位，默认情况下，该参数只有线程数大于corePoolSize时才有用**
 *  **keepAliveTime：空闲线程存活时间**  
    **workQueue必须是BlockingQueue的阻塞队列，当线程池中的线程数超过corePoolSize的时候，线程会进入阻塞队列进行等待，通过workQueue线程池实现了阻塞功能**  
    **几种阻塞队列：**  
    **同步阻塞队列：SyschronousQueue，提交的任务直接提交给线程而不保存它  
    无界阻塞队列：LinkedBlockingQueue，基于链表实现的有界队列，可以存放无限多的提交的任务  
    有界阻塞队列：ArrayBlockingQueue，基于数组实现的有界队列，指定对列的最大长度，防止资源耗尽。**
 *  **threadFactory：线程工厂**  
    **创建线程的工厂，通过自定义的线程工厂可以给每个新建的线程设置一个具有识别度的线程名：默认线程名：poll-X-Therad-XX  
    实现ThreadFactory接口，接口声明如下：**

public interface ThreadFactory {
        
        Thread newThread(Runnable r);
    }

*  RejectedExecutionHandler：饱和策略  
    **线程池的饱和策略，当阻塞队列满了，且没有空闲的工作线程，如果继续提交任务，必须采取一种策略处理该任务，线程池提供了4种策略  
    AbortPolicy：直接抛出异常，默认策略  
    CallerRunsPolicy：用调用者所在的线程来执行任务  
    DiscardOldestPolicy：丢弃阻塞队列中最靠前的任务，并将当前任务添加到队列中  
    DiscardPolicy：直接丢弃任务**

**也可以根据应用场景自定义饱和策略执行：RejectedExecutionHandler**

#### ThreadPoolExecutor线程池执行任务流程 ####

![在这里插入图片描述][6c4355663c034486a063879494209ea4.png]  
**（1）**如果线程池中的线程数量少于corePoolSize，就创建新的线程来执行新添加的任务；  
**（2）**如果线程池中的线程数量大于等于corePoolSize，但队列workQueue未满，则将新添加的任务放到workQueue中，按照FIFO的原则依次等待执行（线程池中有线程空闲出来后依次将队列中的任务交付给空闲的线程执行）；  
**（3）**如果线程池中的线程数量大于等于corePoolSize，且队列workQueue已满，但线程池中的线程数量小于maximumPoolSize，则会创建新的线程来处理被添加的任务；  
**（4）**如果线程池中的线程数量等于了maximumPoolSize，就用RejectedExecutionHandler来做拒绝处理

**总结，当有新的任务要处理时，先看线程池中的线程数量是否大于corePoolSize，再看缓冲队列workQueue是否满，最后看线程池中的线程数量是否大于maximumPoolSize。另外，当线程池中的线程数量大于corePoolSize时，如果里面有线程的空闲时间超过了keepAliveTime，就将其移除线程池。**

[6dd365597087415bb0ec1b193c5b7d75.png]: https://img-blog.csdnimg.cn/6dd365597087415bb0ec1b193c5b7d75.png
[5f9c4ae1b01c455a88f248423793b8fa.png]: https://img-blog.csdnimg.cn/5f9c4ae1b01c455a88f248423793b8fa.png
[419674a560df43f7b91a02566f02e923.png]: https://img-blog.csdnimg.cn/419674a560df43f7b91a02566f02e923.png
[7a9c3c2b3c0c4460a74cae55389c19a1.png]: https://img-blog.csdnimg.cn/7a9c3c2b3c0c4460a74cae55389c19a1.png
[a93660797db745f9ac2a2069449c266d.png]: https://img-blog.csdnimg.cn/a93660797db745f9ac2a2069449c266d.png
[fec7ac3767bd46fbbfc72be2ef2cea6e.png]: https://img-blog.csdnimg.cn/fec7ac3767bd46fbbfc72be2ef2cea6e.png
[efce0eff743848f4b3da5856a39125f4.png]: https://img-blog.csdnimg.cn/efce0eff743848f4b3da5856a39125f4.png
[6c4355663c034486a063879494209ea4.png]: https://img-blog.csdnimg.cn/6c4355663c034486a063879494209ea4.png