java并发编程的艺术第十章——Executor框架

2017-08-26 本文已影响223人会跳舞的机器人

1、Executor框架

1.1、Executor框架的结构

Executor主要由3大部分组成。

任务。包含被执行任务需要实现的接口：Runnable接口或Callable接口。
任务的执行。包括任务执行机制的核心接口Executor，以及继承自Executor接口的ExecutorService接口。Executor有两个关键类实现了ExecutorService接口（ThreadPoolExecutor和ScheduledThreadPoolExecutor）
异步计算的结果。包括接口Future和实现Future接口的FutureTask类。

下面是这些类和接口的简介。

Executor是一个接口，它是Executor框架的基础，它将任务的提交和任务的执行分离开来。
ThreadPoolExecutor是线程池的核心实现类，用来执行被提交的任务。
ScheduledThreadPoolExecutor是一个实现类，可以在给定的延迟后运行命令，或者定期执行命令。SchduledThreadPoolExecutor比Timer更灵活，功能更强大。
Future接口和实现Future接口的FutureTask类，代表异步计算的结果
Runnable接口和Callable接口的实现类，都可以被ThreadPoolExecutor或SchduledThreadPoolExecutor执行

1.2、Executor框架的成员

Executor框架的主要成员：ThreadPoolExecutor、ScheduledThreadPoolExecutor、Future接口、Runnable接口、Callable接口、Executors。

1.2.1、ThreadPoolExecutor

ThreadPoolExecutor通常使用工厂类Executors来创建，Executors可以创建3种类型的ThreadPoolExecutor：SingleThreadExecutor、FixThreadPool、CachedThreadPool。

FixThreadPool：创建固定线程数的线程池，构造函数中可以指定线程数量，适用于为了满足资源管理的需求，而需要限制当前线程数量的应用场景，它适用于负载比较重的服务器。
FixThreadPool内部使用无界队列LinkedBlockingQueue作为任务队列，队列的容量为Integer.MAX_VALUE，由于是无界队列，所以不会拒绝任务，可能会造成任务无限堆积，从而导致系统资源耗尽的情况。
SingleThreadExecutor：创建单个线程的线程池，可以保证顺序执行任务。与FixThreadPool类似，只是SingleThreadExecutor的线程数固定为1
CachedThreadPool：可以根据需要创建新的线程，CachedThreadPool是大小无界的线程池，适用于执行很多短期异步任务的小程序，或者是负载比较轻的服务器。CachedThreadPool的corePool为空，maximumPoolSize为Integer.MAX_VALUE，keepAliveTime为60L，这意味着线程空闲超过60秒则会进行回收。CachedThreadPool内部使用不存储元素的SynchronousQueue作为任务队列（一个put操作等待着一个take操作），这意味着如果任务的提交速度高于线程的处理速度，那么CachedThreadPool则会不断的创建新的线程，在极端的情况下，会耗尽CPU和内存资源。

1.2.2、SchduledThreadPoolExecutor

SchduledThreadPoolExecutor通常使用工厂类Executors来创建，Executors可以创建2中类型的SchduledThreadPoolExecutor，如下：

SchduledThreadPoolExecutor。包含若干个线程的SchduledThreadPoolExecutor。
SingleThreadSchduledExecutor。只包含一个线程的SchduledThreadPoolExecutor。

1.2.3、Future接口

Future接口和实现Future接口的FutureTask类用来表示异步计算的结果，当我们把Runnable接口或者Callable接口的实现类提交（submit）给ThreadPoolExecutor或者SchduledThreadPoolExecutor时，ThreadPoolExecutor或者SchduledThreadPoolExecutor会向我们返回一个FutureTask对象。下面是对应的API

1.2.4、Runnable和Callable接口

Runnable和Callable接口的实现类都可以被hreadPoolExecutor或者SchduledThreadPoolExecutor执行。它们之间的区别是Runnable不会返回结果，而Callable可以返回结果。

除了可以自已创建实现Callable接口的对象外，还可以使用工厂类Executors来把一个Runnable包装成一个Callable。

当我们把一个Callable对象提交给ThreadPoolExecutor或者SchduledThreadPoolExecutor执行时，summit()会向我们返回一个FutureTask对象。我们可以执行FutureTask.get()来等待任务执行完成。当任务完成后FutureTask.get()将会返回任务的结果。

附录：简单的结构脑图

5495889.png

2、ThreadPoolExecutor

FixThreadPool：FixThreadPool内部使用无界队列LinkedBlockingQueue作为任务队列，队列的容量为Integer.MAX_VALUE，由于是无界队列，所以不会拒绝任务，可能会造成任务无限堆积，从而导致系统资源耗尽的情况，keepAliveTime为0L，意味着空闲的线程会被立马终止。
SingleThreadExecutor：与FixThreadPool类似，只是SingleThreadExecutor的线程数固定为1
CachedThreadPool：CachedThreadPool的corePool为空，maximumPoolSize为Integer.MAX_VALUE，keepAliveTime为60L，这意味着线程空闲超过60秒则会进行回收。CachedThreadPool内部使用不存储元素的SynchronousQueue作为任务队列（一个put操作等待着一个take操作），这意味着如果任务的提交速度高于线程的处理速度，那么CachedThreadPool则会不断的创建新的线程，在极端的情况下，会耗尽CPU和内存资源。

3、ScheduledThreadPoolExecutor

ScheduledThreadPoolExecutor主要用来执行需要延迟或者定时执行的任务，功能与Timer类似，不同的是Timer只能单线程允许，ScheduledThreadPoolExecutor可以指定多个线程。

ScheduledThreadPoolExecutor会把要执行的任务放在一个DelayQueue中，ScheduledThreadPoolExecutor会把Runable对象封装成ScheduledFutureTask，ScheduledFutureTask内部包含三个对象

time：任务要被执行的具体时间
sequenceNumber：任务排序编号，如果两个任务的time相同，那么则sequenceNumber较小的会先执行
period：任务执行的周期

ScheduledThreadPoolExecutor执行任务的流程大体是：

从DelayQueue中获取ScheduledFutureTask（time大于当前时间的任务）
执行ScheduledFutureTask
修改ScheduledFutureTask的time为下次要执行的时间
将ScheduledFutureTask再次放入DelayQueue中

简单代码示例：

package main.java.com.robot.demo;

import java.util.Date;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/8/24 10:14
 * @description: ScheduledThreadPoolExecutor
 */
public class ScheduledThreadPoolExecutorDemo {
    private static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(4);

    public static void main(String[] args) {
        // 初始化3秒后开始执行任务，此后每隔1秒执行一次
        scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                System.out.println("正在执行任务，当前时间：" + new Date());
            }
        }, 3L, 1L, TimeUnit.SECONDS);
    }
}

4、FutureTask

Future接口和FutureTask实现代表线程异步执行的结果，可以通过FutureTask.get()方法获取异步返回的结果。

示例：

package main.java.com.robot.demo;

import java.util.Date;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

/**
 * @author: 会跳舞的机器人
 * @date: 2017/8/24 11:14
 * @description: FutureTask示例
 */
public class FutureTaskDemo {
    private static ExecutorService executorService = Executors.newSingleThreadExecutor();

    public static void main(String[] args) {
        FutureTask<String> futureTask = new FutureTask<>(new Callable<String>() {
            @Override
            public String call() throws Exception {
                System.out.println("当前时间：" + new Date() + "执行任务......");
                Thread.sleep(3000);
                return "success";
            }
        });
        // 提交任务
        executorService.submit(futureTask);
        // 获取返回结果
        String result = "";
        try {
            result = futureTask.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("当前时间：" + new Date() + "，线程执行结果返回：" + result);
        // 关闭线程池
        executorService.shutdown();
    }
}