Java多线程——Executor框架详解

小咪咪 2023-06-07 10:34 183阅读 0赞

## Java多线程——Executor框架详解 ##

>   **什么是Executor的两级调度模型**

###  一、Executor框架两级调度模型 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70]

####   两级调度模型： ####

#####     应用层+操作系统层 #####

#####     应用层：将应用分解为若干个任务，使用用户级的调度器（Executor框架）将这些任务映射为固定数量的线程。 #####

#####     操作系统层：操作系统内核将这些线程映射到硬件处理器上。 #####

###  二、Executor框架详解 ###

####   1、Executor框架的结构（三大块） ####

#####    ①任务：被执行的任务需要实现Runnable或Callable接口 #####

#####    ②任务的执行：任务执行机制的核心接口Executor，以及其子接口ExecutorService，以及ExecutorService较为常用的两个子接口（ThreadPoolExecutor和ScheduledThreadPoolExecutor） #####

#####    ③异步计算的结果：Future接口及其实现类FutureTask类 #####

####   2、Executor框架的组成成员 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 1]

#####    ①Executor接口：Executor框架的基础，将任务的提交与任务的执行分离。 #####

#####    ②ThreadPoolExecutor：线程池核心实现类，用来执行被提交的任务。 #####

#####    ③ScheduledThreadPoolExecutor：线程池实现类，在给定的延迟后运行命令，或定期执行命令（比Timer更好） #####

#####    ④Future接口及其子类FutureTask类：异步计算的结果 #####

#####    ⑤Runnable接口和Callable接口的实现类：可以被线程池实现类(ThreadPoolExecutor…)所执行。 #####

####   3、Executor框架的执行流程(类似线程池执行流程) ####

#####    ①主线程创建Runnable或Callable接口的任务对象。（Executors工具类可以将Runnable封装成Callable对象） #####

#####    ②任务提交（两种方式） #####

######    将`Runnable`对象通过`ExecutorService.execute(Runnable command)`交给ExecutorService执行（`没有返回值`） ######

######    将`Callable`对象通过`ExecutorService.submit(Callable task)`交给ExecutorService执行（`有返回值`） ######

######    可以把Runnable对象通过Executors包装成Callable对象并使用submit()提交，也可获取返回值。 ######

#####    ③若执行的是submit()，ExecutorService会返回一个FutureTask对象（`实现了Future接口的对象均可，不一定非得是FutureTask`），关于FutureTask详解请参考：FutureTask #####

###  三、`普通调度池（ThreadPoolExecutor）` ###

####   Executors工具类可以创建三种类型的ThreadPoolExecutor ####

1、可重用固定线程数的线程池：
    	public static ExecutorService newFixedThreadPool(int nThreads)
    	 2、单线程池
    	 public static ExecutorService newSingleThreadExecutor()
    	 3、无大小限制的线程池
    	 public static ExecutorService newCachedThreadPool()

####   实际上Executors工具类也是调用的ThreadPoolExecutor创建线程池。和自定义创建线程池原理一模一样。 ####

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

####   1、`可重用固定线程数的线程池：FixedThreadPool` ####

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

#####    FixedThreadPool使用`无界队列LinkedBlockingQueue`作为线程池的⼯作队列（队列的容量为Integer.MAX\_VALUE）。因此，线程池中的线程数不会超过corePoolSize(其他任务都在无界队列中等着呢) #####

#####    由于使用的是无界队列，因此maximumPoolSize毫无意义。（队列不会满的） #####

#####    同样，keepAliveTime也毫无意义（线程池中没有非核心线程池外的线程会执行） #####

#####    同样是无界队列，运⾏中的FixedThreadPool（未调用停止方法）不会拒绝任务（不会调⽤RejectedExecutionHandler.rejectedExecution⽅法）。 #####

#####    FixedThreadPool的运行流程为： #####

######     a）若当前运行的线程数（任务数）少于corePoolSize，则会创建新线程来执行任务。 ######

######     b）若当前运行的线程数（任务数）等于corePoolSize，将任务加入LinkedBlockingQueue队列中。 ######

######     c）执行完a）中任务的工作线程（Worker）会在循环中反复从LinkedBlockingQueueed获取任务来执行。 ######

#####    `适用场景`：为了满足资源管理的需求而需要限制当前线程数量的应用场合，适用于`负载比较重`的服务器。 #####

>      **举个栗子**

package com.xiaoaxiao.test.thread_test.thread_pool_test;
    		
    		import java.time.LocalDateTime;
    		import java.util.concurrent.ExecutorService;
    		import java.util.concurrent.Executors;
    		
    		/**
    		 * Created by xiaoaxiao on 2019/7/19
    		 * Description: 测试FixedThreadPool，无界队列
    		 */
    		public class ExecutorsTest {
    		
    		    public static void main(String[] args) {
    		        ExecutorService executorService = Executors.newFixedThreadPool(5);
    		
    		        for (int i=0;i<2;i++){
    		            executorService.execute(new Runnable() {
    		                @Override
    		                public void run() {
    		                    System.out.println(Thread.currentThread().getName()
    		                            +" "+ LocalDateTime.now());
    		                }
    		            });
    		        }
    		    }
    		}

####   2、单线程池：SingleThreadExecutor ####

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

#####    将corePoolSize和maximumPoolSize设置为1，其他参数与①相同。 #####

#####    同样使用`无界队列LinkedBlockingQueue`作为工作队列。只不过核心线程池中只有一个工作线程。 #####

#####    SingleThreadExecutor的运行流程为： #####

######     a）若当前运行的线程数少于corePoolSize（线程池中无运行的线程），则会创建新线程来执行任务。 ######

######     b）若当前运行的线程数等于corePoolSize（当前线程池中只有一个运行的线程），将任务加入LinkedBlockingQueue。 ######

######     c）执行完a）中任务的工作线程（Worker）会在循环中反复从LinkedBlockingQueueed获取任务来执行。 ######

#####    `适用场景`：需要保证`顺序的执行`各个任务，并且在任意时间点，不会有多个线程活动的场景 #####

####   3、无大小限制的线程池：CachedThreadPool ####

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

#####    corePoolSize（核心线程数）为0，maximumPoolSize（线程池数）为Integer.MAX\_VALUE，keepAliveTime为60L（线程池中的空闲线程超过60s后被终止） #####

#####    CachedThreadPool采用SynchronousQueue作为工作队列，maximumPool是无界的，`SynchronousQueue`的机制是：`将任务与空闲线程进行配对`。 #####

#####    若主线程提交任务的速度高于maximumPool中线程处理任务的速度，该线程池就会`不断创建新的线程`。极端情况下，CachedThreadPool的风险会因为创建过多线程而`耗尽CPU和内存资源，服务器崩溃` #####

#####    `CachedThreadPool的运行流程为`： #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 2]

#####    概括的说一下CachedThreadPool的执行流程（及SynchronousQueue的执行流程），与普通的队列还是有所不同。 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 3]

#####    由于SynchronousQueue（以下简称S）是一个没有容量的队列，因此不能保存任务，主线程执行execute()后，通过调用SynchronousQueue.offer()将任务给S队列，而S队列此时需要判断，是否有空闲线程通过SynchronousQueue.poll()执行并通知S队列 #####

#####    如果有，则任务与空闲线程匹配成功，如果没有的话有两种情况：1、没任务了2、没空闲线程了，如果是没任务了，则execute()直接结束，如果没空闲线程了，CachedThreadPool再开一个新的线程执行任务。 #####

#####    当空闲线程执行完当前任务是，就会执行SynchronousQueue.poll()通知S队列当前已存在一个空闲线程了，可以分配任务给它。 #####

#####    `适用场景`：很多短期异步任务的小程序，负载较轻的服务器（线程不会等太久，内存要求不太严，任务一来就能直接处理） #####

####   4、使用无界队列作为⼯作队列会对线程池带来如下影响 ####

#####    a）当线程池中的线程数达到corePoolSize后，新任务将在⽆界队列中等待，因此线程池中 的线程数不会超过corePoolSize。 #####

#####    b）由于a），使⽤⽆界队列时maximumPoolSize将是⼀个⽆效参数。（无界队列没有边界，所有没等待的任务都放在队列里面，不会到下一步（maximumPoolSize）的） #####

#####    c）由于a）和b），使⽤⽆界队列时keepAliveTime将是⼀个⽆效参数。（线程池数==核心线程池数，所以不存在空闲线程(`除核心线程池外`的空闲线程)的时间限定） #####

#####    d）`无界队列的隐患`：会导致内存飙升，服务器崩溃 #####

###  四、定时调度池（ScheduledThreadPoolExecutor） ###

####   Executors可以创建两种类型的ScheduledThreadPoolExecutor，用ScheduledExecutorService接收，调用ScheduledExecutorService.schedule()方法，应用于定时器。 ####

>   `定时调度器的提交方式(非execute()、非submit())`：  
>   schedule(Runnable command,long delay, TimeUnit unit)：延迟delay unit(单位)启动线程，只执行一次  
>   scheduleAtFixedRate(Runnable command,long initialDelay,long period,TimeUnit unit)：固定时间间隔<**常用**>（任务启动时开始计时，一个没结束，另一个开始了）：等待initialDelay unit(单位)启动线程，并每隔period unit(单位)再次启动线程  
>   scheduleWithFixedDelay(Runnable command,long initialDelay,long delay,TimeUnit unit);：固定延迟时间（任务结束时开始计时，每次就执行一个）：延迟initialDelay unit(单位)启动线程，并每隔period unit(单位)再次启动线程

####   1、包含若干个线程的定时调度池：ScheduledThreadPoolExecutor ####

#####    使用DelayQueue无界队列作为工作队列，该线程池只需要设置corePoolSize（线程数），只有核心线程池。 #####

#####    ScheduledThreadPoolExecutor的任务传递流程：将待调度任务（ScheduledFutureTask，内部包含time<要被执行的时间>，sequenceNumber<加入到队列中的序号>，period<任务执行间隔>）放到DelayQueue中。 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 4]

#####    ScheduledThreadPoolExecutor的任务执行流程：从DelayQueue中获取已到期（time>=当前时间）的任务(ScheduledFutureTask)（通过DelayQueue.take()），线程1执行该任务，线程1将该任务的time变量(执行时间)修改为下次要执行的时间(周期执行)，将修改过time的任务放回到DelayQueue中（通过DelayQueue.add()）。 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 5]

#####    `适用场景`：需要多个后台线程执行周期任务，同时为了满足资源管理的需求而需要限制后台线程的数量的应用场景 #####

####   2、只包含一个线程的定时调度池：SingleThreadScheduledExecutor（类似于Timer） ####

#####    corePoolSize == 1 #####

#####    `适用场景`：需要单个后台线程执行周期任务，同时需要保证顺序地执行各个任务的应用场景 #####

// 实现一个定时调度池，执行定时重复任务
    	public static void main(String[] args) {
            // 创建一个核心线程数为5的定时调度池
            ScheduledExecutorService scheduledThreadPoolExecutor
                    = Executors.newScheduledThreadPool(5);
    
            for (int i = 0; i < 5; i++) {
                scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
                    @Override
                    public void run() {
                        for (int j = 0; j < 10; j++) {
                            System.out.println
                                    (Thread.currentThread().getName()+":"+j);
                        }
                    }
                },2,3, TimeUnit.SECONDS);
            }
        }

###  五、FutureTask详解 ###

####   未完待续。 ####

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70]: /images/20230601/4cbbcba6d6ec4c26909213dcf51ac9f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 1]: /images/20230601/7f63c40526934bf9931226700ec760ad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 2]: /images/20230601/ed04fd4e938449afb6624dad35d3033b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 3]: /images/20230601/2d6bf279c0dc4c4fa8bf8a0620fc7146.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 4]: /images/20230601/9211f288e68a451681f92e2ffbce58f6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzQ5MDQ0MA_size_16_color_FFFFFF_t_70 5]: /images/20230601/faef445ba6d4455ea7bf2bf9bd74da59.png