并发编程5：如何执行任务？

喜欢ヅ旅行 2024-03-23 18:15 86阅读 0赞

**目录**

1、线程中执行任务的方式

2、Executor 框架

2.1 - 线程的执行策略

2.2 - 线程池

2.3 - Executor 的生命周期

2.4 - 延任务与周期任务

3、找出可利用的并行性-代码示例

3.1 - 单线程的 I/O 操作

3.2 - 携带任务结果的 Callable 与 Future（重要）

3.3 - 使用 Future 实现页面渲染器

3.5 - CompletionService：Executor 与 BlockingQueue

--------------------

大多数并发应用程序都是围绕***“任务执行 (Task Execution)”***来构造的：任务通常是一些抽象的且离散的工作单元。通过把应用程序的工作分解到多个任务中，可以简化程序的组织结构，提供一种事务边界来优化错误恢复过程，以及提供一种并行工作结构来提升并发性。//目的：如何把一个工作拆解成多个任务，并发执行->清晰的任务边界(独立任务有利于并发)

#### 1、线程中执行任务的方式 ####

**串行执行：**在服务器应用程序中，串行处理机制通常都无法提供高吞吐率或快速响应性。//一次只能执行一个请求，主线程阻塞

**并行执行：**通过多个线程来提供服务，从而实现更高的响应性。//多线程执行，不阻塞主线程（将任务从主线程中分离出来）-> 更快的响应性和更高的吞吐率

需要注意的是，线程生命周期的开销非常高（Java中创建线程需要内核态的支持）。活跃的线程会消耗系统资源，尤其是内存（TCB）。

所以，在一定的范围内，增加线程可以提高系统的吞吐率，但如果超出了这个范围，再创建更多的线程只会降低程序的执行速度，并且如果过多地创建一个线程，那么整个应用程序将崩溃。要想避免这种危险，就应该对应用程序可以创建的线程数量进行限制，并且全面地测试应用程序，从而确保在线程数量达到限制时，程序也不会耗尽盗源。//选择合适的线程数量，并需要对线程资源进行管理（避免无限创建线程）

#### 2、Executor 框架 ####

***Executor 基于生产者- 消费者模式***，提交任务的操作相当于生产者 ，执行任务的线程则相当于消费者。如果要在程序中实现一个生产者-消费者的设计，那么最简单的方式通常就是使用 Executor。

在 TaskExecutionWebServer 中，通过使用 Executor，将请求处理任务的提交与任务的实际执行解耦开来，代码如下所示：//Executor是一个接口，可使用Java提供的实现，也可以自己去实现

public class TaskExecutionWebServer {
        private static final int NTHREADS = 100;
        private static final Executor exec = Executors.newFixedThreadPool(NTHREADS);
    
        public static void main(String[] args) throws IOException {
            ServerSocket socket = new ServerSocket(80);
            while (true) {
                final Socket connection = socket.accept();
                //任务
                Runnable task = () -> handleRequest(connection);
                //使用Executor执行任务
                exec.execute(task);
            }
        }
    
        private static void handleRequest(Socket connection) {
            // request-handling logic here
        }
    }

##### 2.1 - 线程的执行策略 #####

通过将任务的提交与执行解耦开来，从而无须太大的困难就可以为某种类型的任务指定和修改执行策略。在执行策略中定义了任务执行的 “What、Where、When、How” 等方面，包括：

*  在什么线程中执行任务?
 *  任务按照什么顺序执行 (FIFO、LIFO、优先级)?
 *  有多少个任务能并发执行?
 *  在队列中有多少个任务在等待执行?
 *  如果系统由于过载而需要拒绝一个任务，那么应该选择哪一个任务?另外，如何通知应用程序有任务被拒绝?
 *  在执行一个任务之前或之后，应该进行哪些动作?

各种执行策略都是一种资源管理工具，最佳策略取决于可用的计算资源以及对服务质量的需求。通过限制并发任务的数量，可以确保应用程序不会由于资源耗尽而失败，或者由于在稀缺资源上发生竞争而严重影响性能。通过将任务的提交与任务的执行策略分离开来，有助于在部署阶段选择与可用硬件资源最匹配的执行策略。//对于如何执行任务的描述

##### 2.2 - 线程池 #####

线程池，指管理一组同构工作线程的资源池。线程池是与***工作队列 (Work Oueue)*** 密切相关的，其中在工作队列中保存了所有等待执行的任务。***工作线程 (Worker Thread)*** 的任务很简单：从工作队列中获取一个任务，执行任务，然后返回线程池并等待下一个任务。//线程池->储存线程，工作队列->储存任务

Java 类库提供了一个灵活的线程池以及一些有用的默认配置。可以通过调用 Executors 中的静态工厂方法之一来创建一个线程池：

***newFixedThreadPool***。newFixedThreadPool 将创建一个固定长度的线程池，每当提交一个任务时就创建一个线程，直到达到线程池的最大数量，这时线程池的规模将不再变化（如果某个线程由于发生了未预期的 Exception 而结束，那么线程池会补充一个新的线程）。//限制线程数量

***newCachedThreadPool***。newCachedThreadPool 将创建一个可缓存的线程池，如果线程池的当前规模超过了处理需求时，那么将回收空闲的线程，而当需求增加时，则可以添加新的线程，线程池的规模不存在任何限制。//不限制线程数量

***newSingleThreadExecutor***。newSingleThreadExecutor 是一个单线程的 Executor，它创建单个工作者线程来执行任务，如果这个线程异常结束，会创建另一个线程来替代。newSingleThreadExecutor 能确保依照任务在队列中的顺序来串行执行（例如 FIFO、LIFO优先级）。//按顺序执行任务

***newScheduledThreadPool***。newScheduledThreadPool 创建了一个固定长度的线程池，而且以延迟或定时的方式来执行任务，类似于 Timer。//执行定时任务

##### 2.3 - Executor 的生命周期 #####

为了解决执行服务的生命周期问题，Executor 扩展了 ***ExecutorService 接口***，添加了一些用于生命周期管理的方法，同时还有一些用于任务提交的便利方法。

/*
     * 该Executor提供管理线程池终止的方法，以及提供用于跟踪一个或多个异步任务进度的Future的方法。
     */
    public interface ExecutorService extends Executor {
        //1-关闭执行器(线程池)：不再接收新任务，然后等待正在执行的线程执行完毕
        void shutdown();
        //关闭执行器(线程池)：停止所有正在执行的任务，并返回等待执行的任务列表
        List<Runnable> shutdownNow();
        
        //2-判断执行器(线程池)是否关闭
        boolean isShutdown();
    
        //3-判断执行器(线程池)关闭后所有任务是否已完成
        //注意，除非先调用shutdown/shutdownNow，否则isTerminated永远不会为true。
        boolean isTerminated();
    
        //4-阻塞当前线程，直到所有任务完成或中断或当前等待超时
        boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;
    
        //5-执行给定的任务，并返回表示该任务的Future。
        <T> Future<T> submit(Callable<T> task);
        <T> Future<T> submit(Runnable task, T result);
        Future<?> submit(Runnable task);
    
        //6-执行给定的任务，并在所有任务完成后返回保存其状态和结果的future列表。
        <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;
        
        //7-执行给定的任务，如果有成功完成的任务，则返回成功完成的任务的结果
        //未完成的任务将被取消
        <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 的生命周期有 3 种状态：**运行**、**关闭**和**已终止**。ExecutorService 在初始创建时处于运行状态。

*  *shutdown 方法*将执行平缓的关闭过程：不再接受新的任务，同时等待已经提交的任务执行完成一一包括那些还未开始执行的任务。
 *  *shutdownNow 方法*将执行粗暴的关闭过程：它将尝试取消所有运行中的任务，并且不再启动队列中尚未开始执行的任务。

在 ExecutorService 关闭后提交的任务将由 ***"拒绝执行处理器(Reiected Execution Handler)"*** 来处理，它会抛弃任务，或者使得 execute 方法抛出一个未检查的 ReiectedExecutionException。等所有任务都完成后，ExecutorService 将转入终止状态。

可以调用 *awaitTermination* 来等待 ExecutorService 到达终止状态，或者通过调用*isTerminated* 来轮询 ExecutorService 是否已经终止。通常在调用 awaitTermination 之后会立即调用 shutdown，从而产生同步地关闭 ExecutorService 的效果。//两个方法结合使用，安全的关闭 ExecutorService：awaitTermination + shutdown

##### 2.4 - 延任务与周期任务 #####

*Timer* 类负责管理延迟任务以及周期任务。然而，Timer 存在一些缺陷，因此应该考虑使用 *ScheduledThreadPoolExecutor* 来代替它。可以通过 ScheduledThreadPoolExecutor 的构造函数或 newScheduledThreadPool 工方法来创建该类的对象。

**Timer 类的问题：**

（1）Timer 在执行所有定时任务时只会创建一个线程。如果某个任务的执行时间过长，那么将破坏其他 TimerTask 的定时精确性。//任务执行时间重叠引发的问题，使用多线程可避免

（2）Timer 线程并不捕获异常，因此当 TimerTask 抛出未检查的异常时将终止定时线程。这种情况下，Timer 也不会恢复线程的执行，而是会错误地认为整个 Timer 都被取消了。因此，已经被调度但尚未执行的 TimerTask 将不会再执行，新的任务也不能被调度，这个问题称之为*"线程泄漏"*。//不能处理异常，不能从异常中恢复

#### 3、找出可利用的并行性-代码示例 ####

//从示例中总结方法

##### 3.1 - 单线程的 I/O 操作 #####

例如，下边的一个页面渲染器程序，程序中图像下载过程的大部分时间都是在等待 I/O 操作执行完成，在这期间 CPU 几乎不做任何工作。因此，这种串行执行方法没有充分地利用 CPU，使得用户在看到最终页面之前要等待过长的时间。

import java.util.*;
    
    /**
     * 使用单线的程渲染器
     */
    public abstract class SingleThreadRenderer {
    
        /**
         * 渲染页面
         */
        void renderPage(CharSequence source) {
            //1-加载文本数据
            renderText(source);
            List<ImageData> imageData = new ArrayList<>();
            //下载多个图片资源
            for (ImageInfo imageInfo : scanForImageInfo(source)) {
                //TODO:图像下载大部分时间都是I/O操作
                imageData.add(imageInfo.downloadImage());
            }
            for (ImageData data : imageData) {
                //2-加载图片数据
                renderImage(data);
            }
        }
    
        interface ImageData {
    
        }
    
        interface ImageInfo {
    
            ImageData downloadImage();
        }
    
        abstract void renderText(CharSequence s);
    
        abstract List<ImageInfo> scanForImageInfo(CharSequence s);
    
        abstract void renderImage(ImageData i);
    }

通过将问题分解为多个独立的任务并发执行，能够获得更高的 CPU 利用率和响应灵敏度。

##### 3.2 - 携带任务结果的 Callable 与 Future（重要） #####

Executor 框架使用 Runnable 作为其基本的任务表示形式。不过 Runnable 具有很大的局限性，虽然 run 方法能通过写入日志文件或者将结果放入某个共享的数据结构来保存执行结果，但是它不能返回一个值或抛出一个受检查的异常。//Runnable不具备返回值

许多任务实际上都存在延迟的计算，比如执行数据库查询，从网络上获取资源，或者计算某个复杂的功能等。对于这些任务，Callable 是一种更好的抽象：它认为主入口点（即 call）将返回一个值，并可能抛出一个异常。//通过 Callable 返回的 Future 对象能取消未执行的任务

***Runnable*** 和 ***Callable*** 描述的都是抽象的计算任务。这些任务通常是有范围的，即都有一个明确的起始点，并且最终会结束。Executor 执行的任务有 4 个生命周期阶段：**创建、提交、开始**和**完成**。由于有些任务可能要执行很长的时间，因此通常希望能够取消这些任务。在 Executor 框架中，已提交但尚未开始的任务可以取消，但对于那些已经开始执行的任务，只有当它们能响应中断时，才能取消。取消一个已经完成的任务不会有任何影响。//任务可以提交也可以取消，执行任务是具有生命周期的

*Future 表示一个任务的生命周期*，并提供了相应的方法来判断任务是否已经完成或取消，以及获取任务的结果和取消任务等。在 *Future* 规范中包含的隐含意义是，任务的生命周期只能前进，不能后退，就像 ExecutorService 的生命周期一样。当某个任务完成后，它就永远停留在 "完成" 状态上。//任务执行生命周期的顺序不能打乱，必须按照规定的顺序进行

***get** 方法的行为取决于任务的状态****(尚未开始、正在运行、已完成)***。

如果任务已经完成，那么 **get** 会立即返回或者抛出一个 **Exception**，如果任务没有完成，那么 **get** 将阻塞并直到任务完成。//get方法可能获得结果也可能抛出异常

如果任务抛出了异常，那么 ***get*** 将该异常封装为 ***ExecutionException*** 并重新抛出。如果任务被取消，那么 ***get*** 将抛出 ***CancellationException***。如果 ***get*** 抛出了 ***ExecutionException***，那么可以通过 ***getCause*** 来获得被封装的初始异常。

//Future接口
    public interface Future<V> {
    
        //尝试取消执行此任务。
        //如果任务已经完成或取消，或者由于其他原因无法取消，则此方法不起作用(返回false)。
        //    否则，如果在调用cancel时该任务尚未启动，则该任务不应运行。
        //如果任务已经启动，那么mayInterruptIfRunning参数决定是否中断正在执行的任务，
        //    true 进行中断，false允许程序执行完成。
        boolean cancel(boolean mayInterruptIfRunning);
    
        //如果此任务在正常完成之前被取消，则返回true。
        boolean isCancelled();
    
        //如果此任务完成，则返回true。
        //完成可能是由于正常终止、异常或取消——在所有这些情况下，此方法都将返回true。
        boolean isDone();
    
        //等待计算完成，然后检索其执行结果。
        V get() throws InterruptedException, ExecutionException;
        V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;
    }

可以通过许多种方法创建一个 *Future* 来描述任务。ExecutorService 中的所有 submit 方法都将返回一个 Future，从而将一个 Runnable 或 Callable 提交给 Executor，并得到一个 Future 用来获得任务的执行结果或者取消任务。

还可以显式地为某个指定的 Runnable 或 Callable 实例化一个 *FutureTask*。由于 *FutureTask*实现了 Runnable，因此可以将它提交给 Executor 来执行或者直接调用它的 run 方法。

##### 3.3 - 使用 Future 实现页面渲染器 #####

为了使页面渲染器实现更高的并发性，首先将渲染过程分解为两个任务，一个是渲染所有的文本，另一个是下载所有的图像。因为其中一个任务是 CPU 密集型，而另一个任务是 I/O 密集型，因此这种方法即使在单 CPU 系统上也能提升性能。//思路一：将I/O密集型任何和CPU密集型任务分开

import java.util.List;
    import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    
    /**
     * 使用Future的渲染器
     */
    public abstract class FutureRenderer {
    
        private final ExecutorService executor = Executors.newCachedThreadPool();
    
        void renderPage(CharSequence source) {
            //1-获取图片路径信息
            final List<ImageInfo> imageInfos = scanForImageInfo(source);
            //2-下载图片任务-> I/O密集型
            Callable<List<ImageData>> task = () -> {
                List<ImageData> result = new ArrayList<>();
                for (ImageInfo imageInfo : imageInfos) {
                    //下载图片资源
                    result.add(imageInfo.downloadImage());
                }
                return result;
            };
            //3-使用线程池执行下载图片任务
            Future<List<ImageData>> future = executor.submit(task);
            //4-加载文本数据-> CPU密集型
            renderText(source);
            //5-加载图片数据
            try {
                //TODO:同步获取所有结果，线程阻塞
                List<ImageData> imageData = future.get();
                for (ImageData data : imageData) {
                    renderImage(data);
                }
            } catch (InterruptedException e) {
                // 重新设置线程的中断标记
                Thread.currentThread().interrupt();
                // 我们不需要这个结果，所以也取消这个任务
                future.cancel(true);
            } catch (ExecutionException e) {
                throw launderThrowable(e.getCause());
            }
        }
    
        interface ImageData {
    
        }
    
        interface ImageInfo {
    
            ImageData downloadImage();
        }
    
        abstract void renderText(CharSequence s);
    
        abstract List<ImageInfo> scanForImageInfo(CharSequence s);
    
        abstract void renderImage(ImageData i);
    }

FutureRenderer 使得染文本任务与下载图像数据的任务并发地执行。当所有图像下载完后，会显示到页面上。这将提升用户体验，不仅使用户更快地看到结果，还有效利用了并行性，但我们还可以做得更好。用户不必等到所有的图像都下载完成，而希望看到每当下载完一幅图像时就立即显示出来。//要求不等到所有结果出来才渲染，而是出来就一个渲染一个

##### 3.5 - CompletionService：Executor 与 BlockingQueue #####

如果向 Executor 提交了一组计算任务，并且希望在计算完成后获得结果，那么可以保留与每个任务关联的 Future，然后反复使用 get 方法，同时将参数 timeout 指定为 0，从而通过轮询来判断任务是否完成。这种方法虽然可行，但却有些繁琐。幸运的是，还有一种更好的方法：完成服务(*CompletionService*)。

CompletionService 将 Executor 和 BlockingQueue 的功能融合在一起。你可以将 Callable 任务提交给它来执行，然后使用类似于队列操作的 take 和 poll 等方法来获得已完成的结果，而这些结果会在完成时将被封装为 Future。*ExecutorCompletionService* 实现了 CmpletionService 并将计算部分委托给一个 Executor。

ExecutorCompletionService 的实现非常简单。在构造函数中创建一个 BlockingQueue 来保存计算完成的结果。当计算完成时，调用 FutureTask 中的 done 方法。当提交某个任务时，该任务将首先包装为一个 QueueingFuture，这是 FutureTask 的一个子类，然后再改写子类的 done 方法，并将结果放入 BlockingQueue 中。

使用 CompletionService 实现页面渲染器：

import java.util.List;
    import java.util.concurrent.CompletionService;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.ExecutorCompletionService;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Future;
    
    /**
     * 使用 CompletionService 实现页面渲染器
     */
    public abstract class Renderer {
    
        private final ExecutorService executor;
    
        Renderer(ExecutorService executor) {
            this.executor = executor;
        }
    
        void renderPage(CharSequence source) {
            final List<ImageInfo> info = scanForImageInfo(source);
            //1-使用CompletionService执行任务
            CompletionService<ImageData> completionService = new ExecutorCompletionService<>(executor);
            for (final ImageInfo imageInfo : info) {
                completionService.submit(() -> imageInfo.downloadImage());
            }
    
            renderText(source);
    
            try {
                //2-从CompletionService中获取执行任务的结果，遍历次数为提交任务的数量
                for (int t = 0, n = info.size(); t < n; t++) {
                    Future<ImageData> f = completionService.take();
                    ImageData imageData = f.get();
                    renderImage(imageData);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } catch (ExecutionException e) {
                throw launderThrowable(e.getCause());
            }
        }
    
        interface ImageData {
    
        }
    
        interface ImageInfo {
    
            ImageData downloadImage();
        }
    
        abstract void renderText(CharSequence s);
    
        abstract List<ImageInfo> scanForImageInfo(CharSequence s);
    
        abstract void renderImage(ImageData i);
    }

通过 CompletionService 从两个方面来提高页面染器的性能：缩短总运行时间以及提高响应性。为每一幅图像的下载都创建一个独立任务，并在线程池中执行它们，从而将串行的下载过程转换为并行的过程，这将减少下载所有图像的总时间。此外，通过从 CompletionService 中获取结果以及使每张图片在下载完成后立刻显示出来，能使用户获得一个更加动态和更高响应性的用户界面。

至此，全文结束。