java.util.concurrent.Future 使用指南

2023-11-10

1.介绍

本篇文章将了解 Future。一个自 Java 1.5 以来就存在的接口，它在处理异步调用和并发处理时非常有用。

2.创建Futures

简单地说，Future 类表示异步计算的未来结果。这个结果最终会在处理完成后出现在 Future 中。

长时间运行的方法很适合异步处理和Future接口，因为可以在等待Future封装的任务完成的同时执行其他进程。

一些可以利用Future异步特性的操作示例如下:

计算密集型过程(数学和科学计算)
操作大型数据结构(大数据)
远程方法调用(下载文件、HTML替换、web服务)

2.1. 使用 FutureTask 实现 Futures

创建一个非常简单的类，用于计算Integer的平方。这显然不适合长时间运行的方法类别，但给它添加一个Thread.sleep()调用，以便它在完成之前持续1秒:

public class SquareCalculator {
      private ExecutorService executor
            = Executors.newSingleThreadExecutor();

    public Future<Integer> calculate(Integer input) {
        return executor.submit(() -> {
             System.out.println("计算："+input+"的平方");
            Thread.sleep(1000);
            return input * input;
        });
    }

    public void shutdown() {
        executor.shutdown();
    }
}

实际执行计算的那段代码包含在 call() 方法中，并作为 lambda 表达式提供。除了前面提到的 sleep() 调用之外，这段没有什么特别之处。

将注意力集中在 Callable 和 ExecutorService 的使用上。

Callable 是一个接口，表示返回结果的任务，并具有单个 call() 方法。在这里，使用 lambda 表达式创建了它的一个实例。

创建 Callable 的实例不去任何地方；仍然需要将这个实例传递给一个执行器，该执行器将负责在新线程中启动任务，并将有价值的 Future 对象返回给我们。这就是 ExecutorService 的用武之地。

有几种方法可以访问 ExecutorService 实例，其中大部分由实用程序类 Executors 的静态工厂方法提供。在这个例子中，使用了基本的 newSingleThreadExecutor()，它提供了一个能够一次处理单个线程的ExecutorService。

一旦有了一个 ExecutorService 对象，只需要调用 submit()，将 Callable 作为参数传递。然后 submit() 将启动任务并返回一个 FutureTask 对象，该对象是 Future 接口的实现。

3.消费Futures

到目前为止，已经学习了如何创建 Future 的实例。

接下来学习和了解 Future API 的所有方法。

3.1. 使用 isDone() 和 get() 获取结果

现在需要调用calculate()，并使用返回的Future 来获取结果Integer。 Future API 中的两个方法将帮助完成这项任务。

Future.isDone() 告诉执行器是否已完成处理任务。如果任务完成，则返回true；否则，它返回 false。

返回实际计算结果的方法是 Future.get()。可以看到这个方法会阻塞执行，直到任务完成。但是，这在我们的示例中不会成为问题，因为我们将通过调用 isDone() 来检查任务是否完成。

   @Test
    public void test1() throws Exception {
        Future<Integer> future = new SquareCalculator().calculate(10);

        while(!future.isDone()) {
            System.out.println("Calculating...");
            Thread.sleep(300);
        }

        Integer result = future.get();
        System.out.println("result:"+result);
    }

get() 方法将阻止执行，直到任务完成。同样，这不会成为问题，因为在我们的示例中，只有在确保任务完成后才会调用 get()。所以在这种情况下，future.get() 将总是立即返回。

值得一提的是 get() 有一个重载版本，它以超时和 TimeUnit 作为参数：

Integer result = future.get(500, TimeUnit.MILLISECONDS);

get(long, TimeUnit) 和 get() 的区别在于，如果任务在指定的超时期限之前没有返回，则前者将抛出 TimeoutException。

3.2. 使用 cancel() 取消 Future

假设触发了一个任务，但由于某种原因，不再关心结果。可以使用 Future.cancel(boolean) 告诉 executor 停止操作并中断其底层线程：

Future<Integer> future = new SquareCalculator().calculate(4);

boolean canceled = future.cancel(true);

Future 实例，来自上面的代码，永远不会完成它的操作。事实上，如果尝试从该实例调用 get()，在调用 cancel() 之后，结果将是 CancellationException。 Future.isCancelled() 会告诉Future 是否已经被取消。这对于避免获得 CancellationException 非常有用。

对 cancel() 的调用也有可能失败。在这种情况下，返回的值将为 false。需要注意的是，cancel() 将一个布尔值作为参数。这控制着执行任务的线程是否应该被中断。

4.更多使用线程池的多线程

当前的 ExecutorService 是单线程的，因为它是通过 Executors.newSingleThreadExecutor 获得的。为了突出这个单线程，同时触发两个计算：

 @Test
    public  void test2() throws Exception {
        SquareCalculator squareCalculator = new SquareCalculator();

        Future<Integer> future1 = squareCalculator.calculate(10);
        Future<Integer> future2 = squareCalculator.calculate(100);

        while (!(future1.isDone() && future2.isDone())) {
            System.out.println(
                    String.format(
                            "future1 任务 %s 和 future2 任务 %s",
                            future1.isDone() ? "【已完成】" : "【未完成】",
                            future2.isDone() ? "】已完成】" : "【未完成】"
                    )
            );
            Thread.sleep(300);
        }

        Integer result1 = future1.get();
        Integer result2 = future2.get();

        System.out.println(result1 + " and " + result2);

        squareCalculator.shutdown();
    }

future1 任务【未完成】和 future2 任务【未完成】
计算：10的平方
future1 任务【未完成】和 future2 任务【未完成】
future1 任务【未完成】和 future2 任务【未完成】
future1 任务【未完成】和 future2 任务【未完成】
计算：100的平方
future1 任务【已完成】和 future2 任务【未完成】
future1 任务【已完成】和 future2 任务【未完成】
future1 任务【已完成】和 future2 任务【未完成】
100 and 10000

很明显，这个过程不是并行的。可以看到第二个任务只有在第一个任务完成后才开始，整个过程大约需要 2 秒才能完成。

为了使程序真正是多线程的，应该使用不同风格的 ExecutorService。如果使用工厂方法提供的线程池，调整示例如下

public class SquareCalculator {
 
    private ExecutorService executor = Executors.newFixedThreadPool(2);
    
    //...
}

通过对 SquareCalculator 类的一个简单更改，现在有了一个能够使用 2 个并发线程的执行器。

future1 任务【未完成】和 future2 任务【未完成】
计算：10的平方
计算：100的平方
future1 任务【未完成】和 future2 任务【未完成】
future1 任务【未完成】和 future2 任务【未完成】
future1 任务【未完成】和 future2 任务【未完成】
100 and 10000

这现在看起来。可以看到 2 个任务同时开始和完成运行，整个过程大约需要 1 秒才能完成。

还有其他工厂方法可用于创建线程池，例如 Executors.newCachedThreadPool()，它在可用时重用以前使用的线程，以及 Executors.newScheduledThreadPool()，它调度命令在给定延迟后运行。

5.ForkJoinTask 概述

ForkJoinTask是一个抽象类，实现了Future，能够运行ForkJoinPool中由少量实际线程承载的大量任务。

ForkJoinTask的主要特征是，它通常会生成新的子任务，作为完成主要任务所需工作的一部分。它通过调用fork()生成新任务，并使用join()收集所有结果，因此是类的名称。

有两个抽象类实现了ForkJoinTask: RecursiveTask，它在完成时返回一个值，以及RecursiveAction，它不返回任何东西。顾名思义，这些类用于递归任务，比如文件系统导航或复杂的数学计算。

新的示列如下：

给定一个整数，它将计算其所有阶乘元素的平方和。因此，例如，如果我们将数字 4 传递给我们的计算器，我们应该从 4² + 3² + 2² + 1² 的总和中得到结果，即 30。

首先，需要创建 RecursiveTask 的具体实现并实现它的 compute() 方法。：

public class FactorialSquareCalculator extends RecursiveTask<Integer> {

    private Integer n;

    public FactorialSquareCalculator(Integer n) {
        this.n = n;
    }

    @Override
    protected Integer compute() {
        if (n <= 1) {
            return n;
        }

        FactorialSquareCalculator calculator
                = new FactorialSquareCalculator(n - 1);

        calculator.fork();

        return n * n + calculator.join();
    }
}

注意如何通过在 compute() 中创建 FactorialSquareCalculator 的新实例来实现递归。通过调用非阻塞方法 fork()，要求 ForkJoinPool 启动此子任务的执行。

join() 方法将返回该计算的结果，将添加当前访问的数字的平方。

    @Test
    public void test4(){
        ForkJoinPool forkJoinPool = new ForkJoinPool();

        FactorialSquareCalculator calculator = new FactorialSquareCalculator(10);

        forkJoinPool.execute(calculator);
        System.out.println("calculator:"+ calculator.join());
    }

在这里插入图片描述

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Java并发编程

Java