Java中的并发编程Executor

我就是我 2022-10-11 06:13 202阅读 0赞

### 并发编程中的Executor ###

*  基本概念
 *  Executor
 *  ExecutorService
 *  Executors
 *  Executor和ExecutorService和Executors的比较
 *   *  Executor
     *  ExecutorService
     *  Executors
 *  自定义线程池

# 基本概念 #

*  **Thread**存在的问题:
    
     *  性能很差,每次调用**new Thread()** 时都会创建新的对象
     *  线程缺乏统一的管理. 可能会无限的新建线程,相互之间竞争,可能会导致占用过多的系统资源而发生死机或者内存溢出问题
     *  缺乏对线程的功能控制. 比如定时执行,定期执行,线程中断等
 *  **Executor**框架:
    
     *  **Java 1.5**之后引入的,位于**java.util.concurrent**包
     *  内部使用了线程池的机制,可以控制线程的启动,执行和关闭,简化并发线程的操作
 *  **Executor特点:**
    
     *  **Executor**来启动线程比**Thread**中的**start()** 方法启动线程更加易于管理,效率更高,节约开销
     *  有助于避免**this**逃逸问题: 如果在构造器中启动一个线程,因为另一个任务可能会在构造器结束之前开始执行,此时可能访问到初始化一半的对象,导致引用不完整问题
     *  性能高. 可以重用已经存在的线程,减少对象的创建开销
     *  可以有效控制最大并发线程数,提高系统资源利用率的同时,也能避免过多的资源竞争,造成阻塞
     *  提供线程的定时执行,定期执行,单线程,并发数控制等功能
 *  **Executor**作为异步执行框架,支持多种不同类型的任务执行策略,提供了一种标准方法将任务的提交过程和执行过程进行解耦:
    
     *  基于生产者-消费者模式,使用**Runnable**来表示任务,提交任务的线程相当于生产者,执行任务的线程相当于消费者
 *  **Executor**中实现了对生命周期的支持,统计信息收集,应用程序管理机制和性能监视机制
 *  **Executor**的类图:  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center]
 *  **Executor**框架包括:
    
     *  **线程池**
     *  **Executor**
     *  **Executors**
     *  **ExecutorService**
     *  **CompletionService**
     *  **Future**
     *  **Callable**

# Executor #

*  **Executor接口:** 定义了一个接收**Runnable**对象的方法**execute()**
    
        void execute(Runnable command);
    
     *  该方法接收一个**Runnable**实例,用来执行一个任务
     *  任务就是一个实现了**Runnable**接口的类
 *  一般情况下,在新线程中使用**Runnable**时创建一个**Runnable**任务如下:

new Thread(new RunnableTask()).start();

但是在**Executor**中,不需要显式地创建线程:

executor.execute(new RunnableTask());

# ExecutorService #

*  **ExecutorService接口:**
    
     *  **Executor**的子类接口
     *  提供生命周期管理的方法,返回**Future**对象
     *  提供可以跟踪一个或者多个异步任务执行情况的方法,返回**Future**对象
 *  **shutdown():**
    
     *  可以通过调用**ExecutorService**的**shutdown(**) 方法平滑的关闭**ExecutorService.** 调用**shutdown()** 方法之后 **,ExecutorService**将停止接收新的任务并且等待已经提交的任务执行完成.已经提交的任务包括提交了已经在执行的任务和提交了还没有执行的任务.当所有已经提交的任务执行完成就关闭**ExecutorService**
     *  可以使用**ExecutorService**接口来实现和管理多线程
 *  **submit():**
    
     *  通过**ExecutorService.submit()** 方法返回的**Future**对象,可以通过调用**Future**对象的**isDone()** 方法查询**Future**是否完成
     *  当任务完成后,会得到一个处理结果,可以通过调用**Future**对象的**get()** 方法获取处理结果
     *  也可以不使用**Future**对象的**isDone()** 方法进行检查,直接调用**Future**对象的**get()** 方法获取结果,此时**get()** 方法将会发生阻塞,直到获取到处理结果
     *  在任务执行过程中,可以使用**Future**对象的**cancel()** 方法取消正在执行中的任务

# Executors #

*  **Executors类:**
    
     *  主要用于提供线程池的相关操作
     *  提供了一系列的工厂方法用于创建线程池,返回的线程池都实现了**ExecutorService**接口
 *  **newFixedThreadPool:**
    
        public static ExecutorService newFixedThreadPool(int nThreads);
    
     *  创建一个固定数目线程的线程池
     *  任意时间点,最多只能存在固定数目的活动线程. 此时如果创建了新的线程,只能在等待队列中等待线程池中的某个线程终止被移除线程池后才能进行线程池
     *  固定数目线程的线程池可以重用线程,但是不能随时新建新的线程
     *  固定数目线程的线程池没有超时的问题,通常用于稳定固定的正规的长时间的并发线程,常常用于服务器中的线程
 *  **newCachedThreadPool:**
    
        public static ExecutorService newCachedThreadPool();
    
     *  创建一个可以缓存的线程池
     *  调用**execute()** 可以重用之前构造过的可用线程
     *  如果没有可用的线程,就会创建一个新的线程添加到线程池中
     *  缓存型线程通常用于执行一些生存周期很短的异步任务. 所以在面向连接的**Daemon**型的**Server**中不常见,但是对于生存期短的异步任务,缓存型线程是**Executors**的首选
     *  如果有线程超过**60**秒没有被使用会被终止并且从缓存中移除
     *  能够重用的线程,必须在有效期**IDLE**之内,默认是**60**秒
     *  缓存型线程池中的线程不用担心终止关闭的问题,因为线程超过**60**秒未被使用会自动终止并从缓存型线程池中移除
 *  **newSingleThreadExecutor:**
    
        public static ExecutorService newSingleThreadExecutor();
    
     *  创建一个单线程的线程**Executor**
     *  任意时间点,单线程的线程池中只能有一个线程
 *  **newScheduledThreadPool:**
    
        public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize);
    
     *  创建一个支持定时以及周期性任务执行的线程池
     *  通常情况下,可以用来替代**Timer**类
     *  周期性任务执行的线程池可以按照**schedule**依次延迟执行或者周期性执行

# Executor和ExecutorService和Executors的比较 #

## Executor ##

*  **Executor**中定义了一个**execute()** 方法用来接收一个**Runnable**接口的对象
 *  **Executor**中的**execute()** 方法不返回任何结果

## ExecutorService ##

*  **ExecutorService**接口继承自**Executor**接口 **,ExecutorService**是**Executor**的子接口
 *  **ExecutorService**中定义的**submit()** 方法可以接收**Runnable**接口的对象和**Callable**接口的对象
 *  **ExecutorService**接口中的**submit()** 方法可以通过**Future**对象返回运算的结果
 *  **ExecutorService**中提供任务提交的方法外,还提供线程声明周期的控制方法等

## Executors ##

*  **Executors**类中提供了工厂方法来创建不同类型的线程池:
    
     *  **newSingleThreadExecutor()** 用来创建一个单线程的线程池
     *  **newFixedThreadPool()** 用来创建一个固定线程数的线程池
     *  **newCachedThreadPool()** 用来根据需要创建新的线程,如果已有线程是空闲的可以重用已有的线程

# 自定义线程池 #

*  **ThreadPoolExecutor:** 自定义线程池. 提供多个构造方法来创建线程池
    
        public ThreadPoolExecutor(int corePoolSize,
                                    int maximumPoolSize,
                                    long keepAliveTime,
                                    TimeUnit unit,
                                    BlockingQueue<Runnable> workQueue,
                                    ThreadFactory threadFactory,
                                    RejectedExecutionHandler handler);
    
     *  **corePoolSize:** 线程池中保存的核心线程数,包括空闲线程
     *  **maximumPoolSize:** 线程池中允许的最大线程数
     *  **keepAliveTime:** 线程池中的空闲线程持续的最长时间
     *  **unit:** 线程池中的空闲线程持续时间的单位
     *  **workQueue:** 任务执行前保存任务的队列,只保存由**execute()** 方法提交的**Runnable**任务
     *  **threadFactory:** 使用**Executor**创建新线程时用到的工厂方法
     *  **handler:** 线程执行因为线程数和队列满时发生阻塞的处理方法
 *  **ThreadPoolExecutor中线程Runnable任务的处理顺序:** 先看线程池中的线程数量是否大于corePoolSize,再看缓冲队列workQueue中是否满了,最后看线程池中的线程
    
     *  如果线程池中的线程数量小于**corePoolSize,** 即使线程池中存在空闲线程,也会创建一个新的线程来执行添加的任务
     *  如果线程池中的线程数量大于等于**corePoolSize,** 但是缓冲队列**workQueue**未满,就将添加的任务放入到**workQueue**中,然后按照**FIFO**的原则等待线程池中有线程空闲出来后将缓冲队列中的任务交由空闲线程执行
     *  如果线程池中的线程数量大于等于**corePoolSize,** 并且缓冲队列**workQueue**已满,但是线程池中的线程数量小于**maximumPoolSize,** 就会创建一个新的线程来执行添加的任务
     *  如果线程池中的线程数量等于了**maximumPoolSize,** 就根据**RejectedExecutionHandler**中4种方式里选择的**handler**执行线程任务处理
 *  线程池中的线程数量大于**corePoolSize**时,如果线程池中的空闲线程持续时间超过**keepAliveTime,** 就会从线程池中移除,这样就可以动态调整线程池中线程的数量了
 *  **缓冲队列中的几种排队策略:**
    
     *  **直接提交:**
        
         *  采用**SynchronousQueue,** 不保持任务而是将任务直接提交给线程处理
         *  如果线程池中的线程都在工作,不存在用于立即执行任务的线程,则试图将任务加入到缓冲队列就会失败,此时会构造一个新的线程来处理添加的任务,并将线程加入到线程池中
         *  直接提交要求是无界的,即最大线程数量**maximumPoolSize**的值为**Integer.MAX\_VALUE.** 这样才能避免拒绝新提交的任务
         *  **newCachedThreadPool**采用的就是直接提交策略
     *  **无界队列:**
        
         *  采用**LinkedBlockingQueue,** 理论上可以对无限多的任务进行排序
         *  使用无界队列可以在线程池中所有**corePoolSize**个线程都工作的情况下将任务添加到缓冲队列中
         *  这样不用创建新的线程来处理添加的任务,线程池中线程的数量不会超过**corePoolSize.** 此时最大线程数量**maximumPoolSize**的值为**Integer.MAX\_VALUE,** 是无界的
         *  当每个任务完全独立于其余的任务,各个任务之间的执行互不影响时,适合使用无界队列
         *  **newFixedThreadPool**采用的就是无界队列策略
     *  **有界队列:**
        
         *  采用**ArrayBlockingQueue,** 并制定队列的最大长度
         *  有界队列使用有限的**maxmumPoolSize,** 有助于防止资源耗尽
         *  有界队列较难调整和控制,需要设定合理的参数,对队列的大小和线程池最大数量的大小相互进行折衷

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0pld2F2ZU94Zm9yZA_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/e030f089bfc4469fbeda7c16da6f697e.png