你不能用线程毁掉 Java：第四部分 - Callable、Future 和朋友

介绍

我们已经在第一部分中了解了如何创建线程。让我们再回忆一下。 线程是Thread在其中运行的东西run，所以让我们使用tutorialspoint java在线编译器并执行以下代码：

public class HelloWorld {

    public static void main(String []args){
        Runnable task = () -> {
            System.out.println("Hello World");
        };
        new Thread(task).start();
    }
}

这是在线程中运行任务的唯一选择吗？

java.util.concurrent.Callable

原来java.lang.Runnable有一个兄弟，他的名字叫java.util.concurrent.Callable，他出生在Java 1.5中。有什么区别？如果我们仔细查看该接口的 JavaDoc，我们会发现，与 不同的是Runnable，新接口声明了一个call返回结果的方法。另外，默认情况下它会抛出异常。也就是说，它使我们无需为try-catch受检查的异常编写块。已经不错了，对吧？现在我们有Runnable一个新任务：

Callable task = () -> {
	return "Hello, World!";
};

但该怎么办呢？为什么我们甚至需要一个在返回结果的线程上运行的任务？显然，将来我们期望收到将来执行的操作的结果。英语中的未来 - 未来。还有一个同名的接口：java.util.concurrent.Future

java.util.concurrent.Future

java.util.concurrent.Future接口描述了一个用于处理我们计划在将来获取结果的任务的 API：用于获取结果的方法、用于检查状态的方法。我们对其实现java.util.concurrent.FutureTaskFuture感兴趣。也就是说，这就是将在 中执行的内容。此实现的另一个有趣之处在于它实现了 和。您可以将其视为在线程中处理任务的旧模型和新模型（在 java 1.5 中出现的意义上的新模型）的适配器。这是一个例子： TaskFutureRunnable

import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class HelloWorld {

    public static void main(String []args) throws Exception {
        Callable task = () -> {
            return "Hello, World!";
        };
        FutureTask<String> future = new FutureTask<>(task);
        new Thread(future).start();
        System.out.println(future.get());
    }
}

从例子中可以看出，使用该方法我们得到了get问题的结果task。 （！）重要的，在使用该方法获得结果的那一刻，get执行变得同步。您认为这里会使用什么机制？没错，没有同步块 - 因此我们在 JVisualVM 中看到的WAITING不是作为monitor或wait，而是作为完全相同的一个park（因为使用了该机制LockSupport）。

功能接口

接下来我们将讨论 Java 1.8 中的类，因此做一个简单的介绍是很有用的。我们看下面的代码：

Supplier<String> supplier = new Supplier<String>() {
	@Override
	public String get() {
		return "String";
	}
};
Consumer<String> consumer = new Consumer<String>() {
	@Override
	public void accept(String s) {
		System.out.println(s);
	}
};
Function<String, Integer> converter = new Function<String, Integer>() {
	@Override
	public Integer apply(String s) {
		return Integer.valueOf(s);
	}
};

有很多不必要的代码，不是吗？每个声明的类都执行一个功能，但为了描述它，我们使用了一堆不必要的辅助代码。Java 开发人员也这么认为。因此，他们引入了一组“函数式接口”（@FunctionalInterface），并决定现在 Java 本身将为我们“想出”一切，除了重要的：

Supplier<String> supplier = () -> "String";
Consumer<String> consumer = s -> System.out.println(s);
Function<String, Integer> converter = s -> Integer.valueOf(s);

Supplier- 提供者。它没有参数，但它返回一些东西，也就是说，它提供它。 Consumer- 消费者。它接受某些东西作为输入（参数）并用它做一些事情，即消耗一些东西。还有另一个功能。它接受某些内容作为输入（参数s），执行某些操作并返回某些内容。正如我们所看到的，泛型被积极使用。如果你不确定，你可以记住它们并阅读“ Java中泛型的理论或如何在实践中放置括号”。

完整的未来

随着时间的推移，Java 1.8 引入了一个名为 的新类CompletableFuture。它实现了该接口Future，这意味着我们的接口task将在将来被执行，我们可以执行get并得到结果。但他也实现了一些CompletionStage。从翻译来看，它的目的已经很清楚了：它是某种计算的某个阶段。该主题的简要介绍可以在概述“ CompletionStage 和 CompletableFuture 简介”中找到。我们直接进入正题吧。让我们看一下可用的静态方法列表来帮助我们开始： 以下是使用它们的选项：

import java.util.concurrent.CompletableFuture;
public class App {
    public static void main(String []args) throws Exception {
        // CompletableFuture уже содержащий результат
        CompletableFuture<String> completed;
        completed = CompletableFuture.completedFuture("Просто meaning");
        // CompletableFuture, запускающий (run) новый поток с Runnable, поэтому он Void
        CompletableFuture<Void> voidCompletableFuture;
        voidCompletableFuture = CompletableFuture.runAsync(() -> {
            System.out.println("run " + Thread.currentThread().getName());
        });
        // CompletableFuture, запускающий новый поток, результат которого возьмём у Supplier
        CompletableFuture<String> supplier;
        supplier = CompletableFuture.supplyAsync(() -> {
            System.out.println("supply " + Thread.currentThread().getName());
            return "Значение";
        });
    }
}

如果我们运行此代码，我们将看到创建CompletableFuture涉及启动整个链。因此，虽然与 Java8 中的 SteamAPI 有一些相似之处，但这就是这些方法之间的区别。例如：

List<String> array = Arrays.asList("one", "two");
Stream<String> stringStream = array.stream().map(value -> {
	System.out.println("Executed");
	return value.toUpperCase();
});

这是 Java 8 Stream Api 的示例（您可以在此处阅读更多相关信息“图片和示例中的 Java 8 Stream API 指南”）。如果运行此代码，它将Executed不会显示。也就是说，当在 Java 中创建流时，流不会立即启动，而是等待，直到需要从中获取值。但CompletableFuture它立即启动链执行，而不等待它被请求计算值。我认为理解这一点很重要。所以我们有CompletableFuture。我们如何创建一条链，我们有什么手段？让我们记住我们之前写过的函数式接口。

• 我们有一个函数 ( Function)，它接受 A 并返回 B。它只有一个方法 - apply(apply)。
• 我们有一个消费者 ( Consumer)，它接受 A 且不返回任何内容 ( Void )。它只有一种方法—— accept（接受）。
• Runnable我们的代码在不接受或返回的线程上运行。它有一个方法 - run（运行）。

要记住的第二件事是，CompletalbeFuture它在工作中使用Runnable消费者和函数。鉴于此，您可以始终记住您CompletableFuture可以这样做：

public static void main(String []args) throws Exception {
        AtomicLong longValue = new AtomicLong(0);
        Runnable task = () -> longValue.set(new Date().getTime());
        Function<Long, Date> dateConverter = (longvalue) -> new Date(longvalue);
        Consumer<Date> printer = date -> {
            System.out.println(date);
            System.out.flush();
        };
        // CompletableFuture computation
        CompletableFuture.runAsync(task)
                         .thenApply((v) -> longValue.get())
                         .thenApply(dateConverter)
                         .thenAccept(printer);
}

方法thenRun有thenApply版本thenAccept。_ Async_ 这意味着这些阶段将在新线程中执行。它将从一个特殊的池中取出，因此事先不知道它将是什么样的流量，新的还是旧的。这完全取决于任务的难度。除了这些方法之外，还有三种更有趣的可能性。为了清楚起见，让我们想象一下我们有一个特定的服务从某个地方接收消息并且需要时间：

public static class NewsService {
	public static String getMessage() {
		try {
			Thread.currentThread().sleep(3000);
			return "Message";
		} catch (InterruptedException e) {
			throw new IllegalStateException(e);
		}
	}
}

现在，让我们看看 的其他功能CompletableFuture。我们可以将结果CompletableFuture与另一个结果结合起来CompletableFuture：

Supplier newsSupplier = () -> NewsService.getMessage();

CompletableFuture<String> reader = CompletableFuture.supplyAsync(newsSupplier);
CompletableFuture.completedFuture("!!")
				 .thenCombine(reader, (a, b) -> b + a)
				 .thenAccept(result -> System.out.println(result))
				 .get();

值得注意的是，默认情况下，线程将是守护线程，因此为了清楚起见get，我们使用等待结果。而且我们不仅可以组合（combine），还可以返回CompletableFuture：

CompletableFuture.completedFuture(2L)
				.thenCompose((val) -> CompletableFuture.completedFuture(val + 2))
                               .thenAccept(result -> System.out.println(result));

这里我要指出的是，为了简洁起见，使用了该方法CompletableFuture.completedFuture。此方法不会创建新线程，因此链的其余部分将在调用它的同一线程中执行completedFuture。还有一个方法thenAcceptBoth。它与 非常相似accept，但如果thenAccept它接受consumer，那么thenAcceptBoth它接受另一个CompletableStage+作为输入BiConsumer，即consumer，它接受 2 个源作为输入，而不是一个。这个词还有另一种有趣的可能性Either： 这些方法接受替代方法CompletableStage，并将在CompletableStage第一个执行的方法上执行。我想用另一个有趣的功能CompletableFuture——错误处理来结束这篇评论。

CompletableFuture.completedFuture(2L)
				 .thenApply((a) -> {
					throw new IllegalStateException("error");
				 }).thenApply((a) -> 3L)
				 //.exceptionally(ex -> 0L)
				 .thenAccept(val -> System.out.println(val));

这段代码不会做任何事情，因为...... 将抛出异常并且什么也不会发生。但如果我们取消注释exceptionally，那么我们就定义了行为。CompletableFuture我还建议观看以下 有关此主题的视频：

• 完整的未来。我想拿走它并使用它 (2015)
• Java 8 中的 CompletableFuture，异步处理正确完成

以我的拙见，这些视频是互联网上最直观的视频之一。他们应该清楚这一切是如何运作的、我们拥有什么武器库以及为什么需要这一切。

结论

希望现在已经清楚如何使用线程在计算后检索计算结果。附加材料：

• 使用 CompletableFuture 编写异步代码
• CompletionStage 和 CompletableFuture 简介
• Java 8 完整的未来。第2部分
• 完整未来指南

#维亚切斯拉夫