java - 在 Java 中从 C++ 复制延迟/异步启动策略

Question

在 C++ 中，您可以使用延迟或异步启动策略启动线程。有没有办法在 Java 中复制此功能？

auto T1 = std::async(std::launch::deferred, doSomething());
auto T2 = std::async(std::launch::async, doSomething()); 

每个的描述——
异步:

If the async flag is set, then async executes the callable object f on a new thread of execution (with all thread-locals initialized) except that if the function f returns a value or throws an exception, it is stored in the shared state accessible through the std::future that async returns to the caller.

延期:

If the deferred flag is set, then async converts f and args... the same way as by std::thread constructor, but does not spawn a new thread of execution. Instead, lazy evaluation is performed: the first call to a non-timed wait function on the std::future that async returned to the caller will cause the copy of f to be invoked (as an rvalue) with the copies of args... (also passed as rvalues) in the current thread (which does not have to be the thread that originally called std::async). The result or exception is placed in the shared state associated with the future and only then it is made ready. All further accesses to the same std::future will return the result immediately.

见 documentation详情。

Answer 1

future
首先，我们要观察 std::async 是执行给定任务并返回 std::future 的工具一旦计算结果可用，就会保存计算结果的对象。
例如我们可以拨打 result.get()阻塞并等待结果到达。此外，当计算遇到异常时，只要我们调用 result.get() 就会存储并重新抛出给我们。 .
Java提供了类似的类，接口(interface)是 Future 最相关的实现是 CompletableFuture .std::future#get大致翻译为 Future#get .甚至异常行为也非常相似。而 C++ 在调用 get 时重新抛出异常, Java 会抛出 ExecutionException其原始异常设置为 cause .

如何获得Future？
在 C++ 中，您可以使用 std::async 创建 future 的对象.在 Java 中，您可以使用 CompletableFuture 中的众多静态辅助方法之一。 .在您的情况下，最相关的是

CompletableFuture#runAsync , 如果任务没有返回任何结果和

CompletableFuture#supplyAsync , 如果任务在完成时返回结果

所以为了创造一个只打印 Hello World!的 future ，例如你可以做

CompletableFuture<Void> task = CompletableFuture.runAsync(() -> System.out.println("Hello World!"));
/*...*/
task.get();

Java 不仅有 lambda，还有方法引用。假设您有一个计算繁重数学任务的方法:

class MyMath {
    static int compute() {
        // Very heavy, duh
        return (int) Math.pow(2, 5);
    }
}

然后你可以创建一个 future ，一旦它可用就返回结果

CompletableFuture<Integer> task = CompletableFuture.runAsync(MyMath::compute);
/*...*/
Integer result = task.get();

异步与延迟
在 C++ 中，您可以选择指定一个启动策略，该策略规定任务的线程行为。让我们把 C++ 对内存的 promise 放在一边，因为在 Java 中你没有那么多的内存控制。
不同之处在于 async将立即安排线程的创建并在该线程中执行任务。结果将在某个时候可用，并在您可以继续执行主要任务时进行计算。它是新线程还是缓存线程的确切细节取决于编译器，未指定。 deferred行为完全不同。当您拨打 std::async 时，基本上没有任何 react ，不会创建额外的线程，也不会计算任务。在此期间根本不会提供结果。但是，只要您拨打 get ，该任务将在您当前的线程中计算并返回结果。基本上就像您自己直接调用该方法一样，没有任何 async公用事业。
std::launch::async在 Java 中
也就是说，让我们关注如何将此行为转换为 Java。让我们从 async 开始.
这是一个简单的方法，因为它基本上是 CompletableFuture 中提供的默认和预期行为。 .所以你只要做 runAsync或 supplyAsync ，取决于您的方法是否返回结果。让我再次展示前面的例子:

// without result
CompletableFuture<Void> task = CompletableFuture.runAsync(() -> System.out.println("Hello World!"));
/*...*/ // the task is computed in the meantime in a different thread
task.get();

// with result
CompletableFuture<Integer> task = CompletableFuture.supplyAsync(MyMath::compute);
/*...*/
Integer result = task.get();

请注意，除了 Executor 之外，还有方法的重载。如果您有自己的可以使用 线程池并想要 CompletableFuture使用它而不是它自己的(更多细节见 here)。
std::launch::deferred在 Java 中
我尝试了很多来模拟这种行为 CompletableFuture但似乎不可能不创建您自己的实现(如果我错了，请纠正我)。无论如何，它要么在创建时直接执行，要么根本不执行。
所以我只是建议使用您提供给 CompletableFuture 的底层任务接口(interface)。 ，例如 Runnable 或 Supplier ， 直接地。在我们的例子中，我们也可以使用 IntSupplier 以避免自动装箱。
这是两个代码示例，但这次具有延迟行为:

// without result
Runnable task = () -> System.out.println("Hello World!");
/*...*/ // the task is not computed in the meantime, no threads involved
task.run(); // the task is computed now

// with result
IntSupplier task = MyMath::compute;
/*...*/
int result = task.getAsInt();

Java 中的现代多线程
最后，我想让您更好地了解当今 Java 中通常如何使用多线程。提供的工具比 C++ 默认提供的要丰富得多。
理想情况下，您的系统设计方式应该不必关心如此小的线程细节。您可以使用 Executors 创建一个自动管理的动态线程池然后针对该任务启动您的初始任务(或使用 CompletableFuture 提供的默认执行程序服务)。之后，您只需在 future 对象上设置一个操作管道，类似于 Stream API，然后等待最终的 future 对象。
例如，假设您有一个文件名列表 List<String> fileNames你想

读取文件

验证其内容，如果无效则跳过

压缩文件

将文件上传到某个 Web 服务器

检查响应状态码

并计算在何处 无效 , 不成功 和 成功 .假设你有一些方法，比如

class FileUploader {
    static byte[] readFile(String name) { /*...*/ }
    static byte[] requireValid(byte[] content) throws IllegalStateException { /*...*/ }
    static byte[] compressContent(byte[] content) { /*...*/ }
    static int uploadContent(byte[] content) { /*...*/ }
}

那么我们可以很容易地做到这一点

AtomicInteger successfull = new AtomicInteger();
AtomicInteger notSuccessfull = new AtomicInteger();
AtomicInteger invalid = new AtomicInteger();

// Setup the pipeline
List<CompletableFuture<Void>> tasks = fileNames.stream()
    .map(name -> CompletableFuture
        .completedFuture(name)
        .thenApplyAsync(FileUploader::readFile)
        .thenApplyAsync(FileUploader::requireValid)
        .thenApplyAsync(FileUploader::compressContent)
        .thenApplyAsync(FileUploader::uploadContent)
        .handleAsync((statusCode, exception) -> {
            AtomicInteger counter;
            if (exception == null) {
                counter = statusCode == 200 ? successfull : notSuccessfull;
            } else {
                counter = invalid;
            }
            counter.incrementAndGet();
        })
    ).collect(Collectors.toList());

// Wait until all tasks are done
tasks.forEach(CompletableFuture::join);

// Print the results
System.out.printf("Successfull %d, not successfull %d, invalid %d%n", successfull.get(), notSuccessfull.get(), invalid.get());

这样做的巨大好处是它将达到最大吞吐量并使用系统提供的所有硬件容量。所有任务都是完全动态和独立执行的，由自动线程池管理。而你只是等到一切都完成。