Java结构化并发和线程池，谁更香?

发表于 更新于 分类于 阅读次数： 本文字数： 4.8k 阅读时长 ≈ 4 分钟

你好，我是猿java。

StructuredTaskScope是 Java 19 引入的一个实验性特性，Java 21正式纳入java.util.concurrent包，它通过将多个子任务视为一个单一的工作单元来简化并发编程，从而提高可靠性和可观察性。那么，StructuredTaskScope和传统的线程池（如ExecutorService）相比，有哪些独特的优点和缺点，这篇文章，我们来聊一聊。

工作原理

StructuredTaskScope的核心理念是“结构化并发”，即任务的启动和结束是有明确边界的，这样可以更容易管理任务的生命周期。StructuredTaskScope提供了一种更高的抽象层次，使得开发者可以更容易地管理一组并发任务，并确保这些任务在某个作用域内被正确地启动、监控和终止。StructuredTaskScope 的工作原理可以归纳为以下几个关键点：

作用域管理

StructuredTaskScope提供了一个明确的作用域（Scope），在这个作用域内，可以启动多个并发任务。 当作用域结束时，所有在这个作用域内启动的任务都会被自动管理，包括等待任务完成或取消未完成的任务。

任务启动

使用fork方法在作用域内启动任务。这个方法会返回一个Future对象，代表异步计算的结果。这些任务可以是普通的阻塞任务，也可以是使用虚拟线程（Virtual Threads）执行的任务。

任务等待和异常处理

使用join方法等待所有任务完成或任一任务失败。使用throwIfFailed方法检查是否有任务失败，并在有任务失败时抛出异常。这些方法确保了任务的错误处理和结果获取更加集中和明确。

资源管理

当StructuredTaskScope的作用域结束时，它会自动等待所有任务完成或取消所有未完成的任务。这简化了资源管理，因为开发者不需要手动管理线程池或其他资源。

StructuredTaskScope 是 Java 19 引入的一个实验性特性，用于简化并发任务的管理。由于它是一个预览特性，核心源码可能在未来版本中发生变化。尽管如此，我们可以通过 Java 19 的源码来了解其核心实现。

StructuredTaskScope 的核心实现涉及多个类和接口，包括任务的启动、管理、监控和终止等功能。下面是一个简化版的核心源码示例，帮助你理解其工作原理。

核心源码分析

下面是StructuredTaskScope核心源码的简化版本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
package java.util.concurrent;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;

public abstract class StructuredTaskScope<T> implements AutoCloseable {
    private final List<Future<?>> futures = new ArrayList<>();
    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final AtomicReference<Throwable> failure = new AtomicReference<>(null);

    protected StructuredTaskScope() {}

    public Future<T> fork(Callable<T> task) {
        if (closed.get()) {
            throw new IllegalStateException("Scope is closed");
        }
        FutureTask<T> future = new FutureTask<>(task);
        futures.add(future);
        new Thread(future).start(); // 可以替换为虚拟线程
        return future;
    }

    public void join() throws InterruptedException {
        for (Future<?> future : futures) {
            try {
                future.get(); // 等待所有任务完成
            } catch (ExecutionException e) {
                failure.compareAndSet(null, e.getCause());
            }
        }
    }

    public void throwIfFailed() throws ExecutionException {
        Throwable ex = failure.get();
        if (ex != null) {
            throw new ExecutionException(ex);
        }
    }

    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            for (Future<?> future : futures) {
                future.cancel(true); // 取消所有未完成的任务
            }
        }
    }

    public static class ShutdownOnFailure<T> extends StructuredTaskScope<T> {
        public ShutdownOnFailure() {
            super();
        }

        @Override
        public void join() throws InterruptedException {
            super.join();
            if (failure.get() != null) {
                close(); // 关闭作用域
            }
        }
    }
}

关键点解释

1. 任务启动 (fork)

  • fork 方法用于启动一个新的任务，并将其封装在 FutureTask 中。
  • 任务被添加到 futures 列表中，并通过新线程启动（可以替换为虚拟线程）。

2. 任务等待 (join)

  • join 方法等待所有任务完成。
  • 如果某个任务抛出异常，异常会被记录在 failure 中。

3. 异常处理 (throwIfFailed)

  • throwIfFailed 方法检查是否有任务失败，并在有任务失败时抛出异常。

4. 资源管理 (close)

  • close 方法用于取消所有未完成的任务，并确保作用域只关闭一次。

5. 作用域管理 (ShutdownOnFailure)

  • ShutdownOnFailure 是一个具体的 StructuredTaskScope 实现，如果有任务失败，会自动关闭作用域。

示例用法

以下是使用 StructuredTaskScope 的基本示例：

1
2
3
4
5
6
7
8
9
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<Weather>()) {
var taskA = scope.fork(() -> Weather.readWeatherFromServerA());
var taskB = scope.fork(() -> Weather.readWeatherFromServerB());
var taskC = scope.fork(() -> Weather.readWeatherFromServerC());

    scope.join(); // 等待所有子任务完成
    Weather weather = scope.result(); // 获取结果
    return weather;
}

在这个示例中，三个子任务被并行执行，join() 方法会阻塞直到所有子任务完成，而 result() 方法则返回第一个成功任务的结果。

优点

代码可读性

使用 StructuredTaskScope 可以使代码结构更加清晰，所有相关的子任务都在一个作用域内定义，易于理解和维护。

自动管理任务生命周期

所有子任务的生命周期与其父任务紧密相连。当父任务完成或失败时，所有子任务会自动终止，避免了悬挂线程的问题。

智能错误处理

StructuredTaskScope 可以根据任务之间的关系智能地处理错误。例如，使用 ShutdownOnFailure 策略时，如果任何子任务失败，其他子任务会自动取消，确保资源的有效利用。

支持虚拟线程

默认情况下，StructuredTaskScope 使用虚拟线程，这使得创建和管理大量线程变得高效，避免了传统线程模型中的阻塞问题。

灵活的任务管理

允许开发者通过 fork 方法创建任意数量的子任务，并可以根据不同的策略（如成功或失败）来管理这些任务的执行和结果。

缺点

对开发者的责任要求高

开发者需要确保子任务能够正确处理取消请求和中断标志。如果子任务未能妥善处理这些标志，可能导致响应性问题。

学习曲线
对于习惯于传统多线程编程的开发者，StructuredTaskScope 的新模型可能需要一定的学习时间来适应和理解其工作原理和使用方式。

性能开销
尽管使用虚拟线程可以提高并发性能，但在某些情况下，创建和管理大量子任务可能会引入额外的开销，尤其是在任务数量非常庞大的时候

使用场景

StructuredTaskScope的使用场景通常是需要管理一组相关任务的地方，例如：

  • 并行处理多个独立的子任务并等待所有任务完成。
  • 启动多个任务并在第一个任务完成时取消其他任务。
  • 在一个作用域内管理任务的启动和关闭，确保资源的正确释放。

对比线程池 ExecutorService

相似点

  • 都用于管理并发任务。
  • 都提供了任务提交和执行的机制。
  • 都可以处理任务的异常。

不同点

1. 抽象层次

  • ExecutorService：提供了一个低层次的线程池管理机制，适用于需要高度自定义并发行为的场景。
  • StructuredTaskScope：提供了一个更高层次的结构化并发管理机制，适用于需要清晰任务边界和生命周期管理的场景。

2. 任务生命周期管理

  • ExecutorService：需要手动管理任务的启动、监控和终止。
  • StructuredTaskScope：自动管理任务的生命周期，确保任务在作用域结束时被正确处理。

3. 资源管理

  • ExecutorService：需要手动管理线程池的关闭和资源释放。
  • StructuredTaskScope：自动管理资源，在作用域结束时自动处理未完成的任务。

总结

StructuredTaskScope提供了一种更结构化的方式来管理并发任务，适用于需要清晰任务边界和生命周期管理的场景。相比传统的ExecutorService，它在任务管理和资源管理方面提供了更高的抽象层次，但也有一些限制和不稳定性。在选择使用哪种工具时，需要根据具体的应用场景和需求进行权衡。

学习交流

如果你觉得文章有帮助，请帮忙转发给更多的好友，或关注公众号：猿java，持续输出硬核文章。

drawing