多线程

发布日期: 2023-02-02

更新日期: 2023-08-27

文章字数: 2k

阅读时长: 7 分

阅读次数:

Executor 介绍

Executor 是 java.util.concurrent 包中重要的接口，是用来执行提交线程任务的对象。

线程池是线程的集合，线程池中的单个线程是工作单元也是执行机制。

JDK 1.5 之后，为了将工作单元与执行机制分离，Executor 框架就出现了，Executor 框架实现了线程池的功能，是一个用于统一创建与运行的接口。

Executor 源码

Executor 是一个顶层接口，其源码中仅定义了 execute() 方法

public interface Executor &#123;
    void execute(Runnable command);
&#125;

其实现类中，最终要的就是 ExecutorService 接口了

Executor 框架结构

Executor 框架由三大部分组成，包括任务对象、任务执行、以及任务执行结果。

任务，即线程池的工作单元，包括被执行任务需要实现的接口，Runnable 或 Callable 接口
任务执行，即将任务分给执行机制，执行机制如 Executor 接口或 ExecutorService 接口（继承了 Executor ）
任务执行结果，线程的执行往往是异步的，对于异步计算结果，有 Future 或 FutureTask（实现了 Future ）

Executor 框架继承实现关系

FutureTask 类，实现了 RunnableFuture 接口，而 RunnableFuture 接口继承了 Future 接口和 Runnable 接口。
Callable 接口
Executor 接口作为顶层接口，
- ExecutorService 接口继承了 Executor 接口
  - AbastractExecutorService 抽象类实现了 ExecutorService 接口
    - ThreadPoolExecutor 类继承了 AbastractExecutorService 抽象类
  - ScheduledExecutorService 接口继承了 ExecutorService 接口
    - ScheduledThreadPoolExecutor 类实现了 ScheduledExecutorService 接口，同时又继承了 ThreadPoolExecutor 类

Executor 框架成员

Executor，顶层接口

Runnable/Callable 接口，任务接口

ThreadPoolExecutor，实现类

ScheduledThreadPoolExecutor，实现类

Executors，工厂类，其中提供了常见的配置线程池的方法，也可以使用 ThreadPoolExecutor 自行初始化线程池。但是因为 ThreadPoolExecutor 的参数众多且意义重大，为避免出错，才有了 Executors 工厂类。

Future / FutureTask，结果接口和实现类

Executor 框架使用流程

使用 Runnable 或 Callable 接口创建线程任务对象

class callableTest implements Callable<String >&#123;
    @Override
    public String call() &#123;
        try&#123;
            String a = "return String";
            return a;
        &#125;
        catch(Exception e)&#123;
            e.printStackTrace();
            return "exception";
        &#125;
    &#125;
&#125;

使用 Executor 的实现类 ThreadPoolExecutor 或 ScheduledThreadPoolExecutor 创建线程执行对象，也可以使用 Executors 工厂类获取。
- 调用创建对象的 submit() 或 execute() 方法来将线程任务添加到线程中
- execute() 方法接收参数并从线程池中选择线程进行执行，一般 Runnable 任务无返回值
- submit() 接收参数从线程池中选择线程执行后，提供一个返回值 Future 对象，Callable 类型对象会有返回值。
- 也可以通过 callable() 方法将 Runnable 传入实现线程结果返回
```
ThreadPoolExecutor tpe = new ThreadPoolExecutor(5, 10,
                100, MILLISECONDS, new ArrayBlockingQueue<Runnable>(5));
Future<String> future = tpe.submit(new callableTest());
```
对于有返回结果的，调用 Future 对象的 get() 方法获取返回结果，注意 get() 方法获取结果是阻塞的，即线程没有执行完则调用结果的线程会阻塞直到获取执行结果。

ExecutorService

并发 API 引入了 ExecutorService 作为一个在程序中直接使用 Thread 的高层次的替换方案。

ExecutorService 作为接口，直接继承了 Executor 接口，并在其基础上，提供了终止线程池线程的 shutdown() 方法，并新增了可以生成 Future 结果对象的 submit() 方法。

Executors 必须显式的停止，否则它们将持续监听新的任务，即 Java 线程不停止，保持活跃等待。

ExecutorService 接口则包含大量用于控制任务进度和管理服务终止的方法。我们可以使用此接口来提交要执行的任务，还可以使用此接口返回的 Future 实例控制任务的执行。

ExecutorService 源码结构

ExecutorService 在 Executor 的基础上增加了执行停止、有返回结果的执行方法、以及有返回结果的批量执行方法。

执行停止相关方法

void shutdown();，停止任务的执行，如果当前有任务正在执行，则会等待任务执行完成后停止，先前已提交但是没有执行的任务不会开始执行，所有线程执行完当前任务后停止。
List<Runnable> shutdownNow(); ，该方法在 shutdown() 方法的基础上，会立即停止所有线程，即使有正在执行的任务也会停止，方法返回正在等待执行的任务列表。
boolean isShutdown();，判断执行器是否停止
boolean isTerminated();，判断是否在所有的任务都正常执行完成后关闭线程执行，如果没有执行 shutdown() 等方法，则线程永远不会停止。
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;，该方法停止线程池时，会阻塞等待所有线程任务执行完成，或线程抛出异常、或线程被迫中断。

带返回结果的执行方法

<T> Future<T> submit(Callable<T> task);，方法传入 Callable 任务对象，提供线程执行后，返回任务执行结果封装的 Future 对象，返回类型为 Callable 的泛型类型
<T> Future<T> submit(Runnable task, T result);，增加了指定返回结果类型的 result 参数
Future<?> submit(Runnable task);，返回结果无指定类型

批量执行任务并返回结果

传入 Callable 任务的集合，返回结果为所有任务的结果的列表

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;

参数传入 Callable 任务的集合、超时时间和时间单位，在最大超时时间内执行任务，并返回所有任务结果列表
```
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException;
```

参数传入 Callable 任务的集合，在其中任意一个任务执行得到结果后返回

<T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;

参数传入 Callable 任务的集合、超时时间和单位，在最大超时时间内，返回任意一个任务执行得到结果后返回，如果都没有执行完则返回超时异常
```
<T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
```

ExecutorService 实现和继承关系

ExecutorService 接口下，比较常见有的 AbstractExecutorService 抽象类和 ScheduledExecutorService 接口。

ExecutorService 创建和使用

ExecutorService 作为一个在程序中直接使用 Thread 的高层次的替换方案，可以通过 Executors 工厂类来创建 ExecutorService 对象使用。

// 创建
ExecutorService executor = Executors.newSingleThreadExecutor();
// 执行
Future<String> future = executor.submit(() -> &#123;
String threadName = Thread.currentThread().getName();
System.out.println("Hello " + threadName);
&#125;);
// 阻塞获取结果
future.get();

Executor 和 ExecutorService 的使用区别

Executor 中仅定义了一个 execute() 方法用来执行 Rnnable 类型的任务，如果可以满足要求则可以使用 Executor 类型接收对象；如果操作较多，则使用 ExecutorService 类型接收对象。

ScheduledExecutorService

ScheduledExecutorService 接口也是继承 ExecutorService 接口的，与 ExecutorService 接口不同的是，ScheduledExecutorService 可以设置任务在指定的延迟之后、或者以指定周期执行。

源码结构

ScheduledExecutorService 接口中仅定义了三种扩展方法，分别是 schedule、scheduleAtFixedRate、和 scheduleWithFixedDelay。

schedule 方法用来在指定延迟之后执行任务，方法的任务参数可以是 Runnable 或 Callable

public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);

public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                           long delay, TimeUnit unit);

scheduleAtFixedRate 方法是在指定的延迟之后，使用固定周期进行执行任务，该周期是指两次任务的开始时间间隔。

public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                                  long initialDelay,
                                                  long period,
                                                  TimeUnit unit);

scheduleWithFixedDelay 方法也是是在指定的延迟之后，使用固定周期进行执行任务，与 scheduleAtFixedRate 不同的是，scheduleWithFixedDelay 方法的周期是指的上次任务执行完成后到下次任务开始时的时间间隔，即纯间隔时间。

public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                     long initialDelay,
                                                     long delay,
                                                     TimeUnit unit);