聊聊 Java 21 中的结构化并发（预览版）

原创

Lorin 洛林

发布于 2023-11-08 18:14:29

5140

文章被收录于专栏：Java 技术小屋Java 技术小屋

hello，大家好，我是 Lorin，今天和大家一起聊聊 Java 21 中另一个有意思的预览特性 - 结构化并发。

结构化编程

在开始聊结构化并发之前，我们先简单聊聊一下结构化编程：

Goto Statement Considered Harmful

在计算机发展的早期，程序员使用汇编语言进行编程，在之后的一段时期，诞生了比汇编略微高级的编程语言，如 FORTRAN、FLOW-MATIC 等。这些语言虽然在一定程度上提高了可读性，但是仍然存在很大的局限性。如下所示就是一段 FLOW-MATIC 代码：

由于当时块语法还没有发明，因此 FLOW-MATIC 不支持 if 块、循环块、函数调用、块修饰符等现代语言必备的基础特性。整段代码就是一系列按顺序排列并打平的命令。关于控制流，程序支持两种方式，分别是：顺序执行、跳转执行，即 GOTO 语句。
顺序执行的逻辑非常简单，它总是能够找到执行入口与出口。与之相反，跳转执行则充满了不确定性。如果程序中存在 GOTO 语句，那么它可以在任何时候跳转至任何指令位置。一旦程序大量使用了 GOTO 语句，那么最终将变成面条式代码（Spaghetti code）。如下图所示：

结构化编程

在发表《Goto Statement Considered Harmful》之后，Dijkstra 又发表了《Notes on Structured Programming》表达了其理想的编程范式，提出了结构化编程的概念。
结构化编程在现在看来是理所当然的，但是在当时并不是。结构化编程的核心是基于块语句，实现代码逻辑的抽象与封装，从而保证控制流具有单一入口和单一出口。现代编程语言中的条件语句、循环语句、函数定义与调用都是结构化编程的体现。

相比 GOTO 语句，基于块的控制流有一个显著的特征：控制流从程序入口进入，中途可能会经历条件、循环、函数调用等控制流转换，但是最终控制流都会从程序出口退出。这种编程范式使得代码结构变得更加结构化，思维模型变得更加简单，也为编译器在低层级提供了优化的可能。
因此，完全禁用 GOTO 语句已经成为了大部分现代编程语言的选择。虽然，少部分编程语言仍然支持 GOTO，但是它们大都支持高德纳（Donald Ervin Knuth）所提出的前进分支和后退分支不得交叉的理论。类似 break、continue 等控制流命令，依然遵循结构化的基本原则：控制流拥有单一的入口与出口。

非结构化并发

在开始了解结构化并发前，我们先回顾一下 Java 中非结构化并发的写法。

        ExecutorService executorService= Executors.newFixedThreadPool(3);
        Future<String> user = executorService.submit(() -> getUser());
        Future<Integer> order = executorService.submit(() -> getOrder());
        String theUser = user.get();   // 加入 getUser
        int theOrder = order.get();    // 加入 getOrder

非结构化并发存在的一些问题

线程泄漏

当 getUser 或者 getOrder 抛出异常时，另外一个任务并不会停止执行，一方面会导致线程资源的浪费，另一方面可能干扰其它任务。
又或者其中一个线程已经执行失败，继续执行的线程执行时间很长，这时候需要阻塞等待线程的完成，同样造成资源的浪费。

代码本身不会体现任务间的关系

上面的各种情况其实都是在开发人员的脑海中，程序逻辑本身并不会体现出来，这样不仅会产生更多的错误空间，而且会使错误排查更加困难。

排查错误困难

多线程编程中一个比较大的难点就是对错误的追踪，任务运行在不同的线程上，当然我们现在有跨线程追踪的方案，但是远远没有我们使用非并发编程时的简单和方便。

结构化并发

在单线程编程模型中，编程语言通过代码块避免控制流随意跳转，从而实现程序的结构化。但在多线程编程（并发编程）模型中，线程之间控制和归属关系仍然存在很多问题，其面临的问题与 GOTO 的问题非常相似，这也是结构化并发所要解决的问题。
什么是结构化并发呢？结构化并发的核心是在并发模型下，也要保证控制流的单一入口和单一出口。程序可以产生多个控制流来实现并发，但是所有并发控制流在出口时都应该处于完成或取消状态，控制流最终在出口处完成合并。
下面是非结构化并发（图一）和结构化并发（图二）的运行示例图：

Java 结构化并发示例

public class Test {
    public static void main(String[] args) {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Future<String> userFuture = scope.fork(() -> getUser());
            Future<Integer> orderFuture = scope.fork(() -> getOrder());
            scope.join()            // Join both subtasks
                    .throwIfFailed();  // ... and propagate errors
            System.out.println("User: " + userFuture.get());
            System.out.println("Order: " + orderFuture.get());
        } catch (ExecutionException | InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    private static int getOrder() throws Exception {
        // throw new Exception("test");
        return 1;
    }

    private static String getUser() {
        return "user";
    }

}

结构化并发带来的好处

下面我们看看结构化并发如何解决非结构化并发中可能存在的一些问题：

短路处理

如果一个getOrder()或getUser()一个子任务失败，则另一个尚未完成的任务将被取消。（这是由实施的关闭策略管理的ShutdownOnFailure；其他策略也是可能的，同时支持自定义策略）。避免了线程资源浪费以及可能的无意义阻塞。

取消传播

如果线程在调用期间被中断join()，则当线程退出作用域时，两个子任务都会自动取消。避免了线程资源浪费。

清晰性

上面的代码有一个清晰的结构：设置子任务，等待它们完成或被取消，然后决定是成功（并处理已经完成的子任务的结果）还是失败（没有什么需要清理的）。

可观察性

线程转储 - 线程堆栈信息可以清楚的显示任务层次结构：

Exception in thread "main" java.lang.RuntimeException: java.util.concurrent.ExecutionException: java.lang.Exception: test
	at Test.main(Test.java:21)
Caused by: java.util.concurrent.ExecutionException: java.lang.Exception: test
	at jdk.incubator.concurrent/jdk.incubator.concurrent.StructuredTaskScope$ShutdownOnFailure.throwIfFailed(StructuredTaskScope.java:1188)
	at Test.main(Test.java:17)
Caused by: java.lang.Exception: test
	at Test.getOrder(Test.java:26)
	at Test.lambda$main$1(Test.java:15)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317)
	at java.base/java.lang.VirtualThread.run(VirtualThread.java:305)
	at java.base/java.lang.VirtualThread$VThreadContinuation.lambda$new$0(VirtualThread.java:177)
	at java.base/jdk.internal.vm.Continuation.enter0(Continuation.java:327)
	at java.base/jdk.internal.vm.Continuation.enter(Continuation.java:320)