books

《Java 并发编程·线程与内存模型》15.2 Joiner、失败传播与取消

awaitAll 只等结果不管失败，真正让「一个下游挂了、其余立刻停」的是 Joiner。本节用三个下游扇出实测五个内置 Joiner 的返回类型与取消时机：失败后 2ms 兄弟任务被中断，join 抛 FailedException 并保留原始 cause。

Leeting Yan 9 分钟阅读 4393 字

承接的问题

15.1 末尾留了一个数字：查不到书的那次请求花了 410ms 才报错，而真正失败发生在 0ms。原因很简单，awaitAll 的语义是「等所有子任务结束」，不是「一有结果就做决定」。它把失败存进 Subtask，然后继续等另一个 400ms 的查询。

业务上这是浪费：库存已经确定不存在，再等会员配额毫无意义。更糟的是扇出场景——三个下游发通知，push 挂了，mail 和 sms 还在往用户手机上发，等它们发完才报错，等于把一次失败放大成三次副作用。

这一节解决的就是「失败之后剩下的任务怎么办」。JDK 25 的答案是 Joiner：它不是回调，而是 join() 的决策器——决定什么时候可以返回、返回什么、以及要不要取消其余子任务。先说版本：StructuredTaskScope 在 JDK 25 仍是 preview，未来版本 API 可能变化，本节代码都要 --enable-preview。

Joiner 是 join 的决策器

open() 的行为等价于 open(Joiner.awaitAllSuccessfulOrThrow())（15.1 实测两者成功与失败路径一致）。五个内置 Joiner 覆盖了绝大多数扇出形状：

Joinerjoin() 返回何时返回失败/成功后的动作
awaitAll()Void全部结束不取消，失败留在 Subtask 里
awaitAllSuccessfulOrThrow()Void首个失败，或全部成功首个失败即取消其余
allSuccessfulOrThrow()Stream<Subtask<T>>同上同上，并返回全部子任务
anySuccessfulResultOrThrow()T首个成功，或全部失败首个成功即取消其余
allUntil(Predicate)Stream<Subtask<T>>谓词返回 true触发后取消未完成者

两个类型细节必须记住，否则编译不过：

  • open(Joiner) 需要写两个类型参数 <T, R>，R 是 join() 的返回类型。allSuccessfulOrThrow() 的 R 是 Stream<Subtask<T>>，不是 Void；写成 <String, Void> 会报「找不到合适的方法」。
  • open(joiner, cfg) 也是两个类型参数，即使你只关心 Void，也得写全 StructuredTaskScope.<Object, Void>open(...)。

Joiner 只有一个抽象方法 result()，看起来可以写成 lambda，但不要自己实现：它还有两个 default 方法 onFork/onComplete 参与取消决策，写错一个就是泄漏。JDK 25 的内置实现覆盖了「全成功」「任一成功」「满足条件就停」三类，组合不出想要的语义时，先想想是不是该换问题拆法。

实测：失败传播与取消时机

下面这段把四个 Joiner 放在同一个场景里跑：mail、sms 各睡 300~800ms，push 在第 120ms 抛 503。每个子任务在 catch (InterruptedException) 里打印被打断的时刻，这样取消发生在哪一刻是可测的。

import java.time.Duration;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Subtask;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

public class JoinerSemantics {

    static long T0 = System.nanoTime();
    static long ms() { return (System.nanoTime() - T0) / 1_000_000; }

    /** push 在 120ms 抛 503，其余睡满 budgetMs。 */
    static String send(String name, long budgetMs, boolean fail) throws Exception {
        try {
            Thread.sleep(Duration.ofMillis(fail ? 120 : budgetMs));
        } catch (InterruptedException e) {
            System.out.printf("  [%-4s] +%4dms INTERRUPTED%n", name, ms());
            throw e;
        }
        if (fail) {
            System.out.printf("  [%-4s] +%4dms 503%n", name, ms());
            throw new IllegalStateException(name + " 下游 503");
        }
        System.out.printf("  [%-4s] +%4dms ok%n", name, ms());
        return name + "-ok";
    }

    public static void main(String[] args) throws Exception {
        System.out.println("A. awaitAll");
        T0 = System.nanoTime();
        try (var scope = StructuredTaskScope.<String, Void>open(
                StructuredTaskScope.Joiner.awaitAll())) {
            Subtask<String> mail = scope.fork(() -> send("mail", 300, false));
            Subtask<String> sms  = scope.fork(() -> send("sms",  300, false));
            Subtask<String> push = scope.fork(() -> send("push", 0, true));
            System.out.println("  join=" + scope.join() + " states="
                    + mail.state() + "/" + sms.state() + "/" + push.state());
            System.out.println("  push.exception()=" + push.exception());
            try {
                push.get();
            } catch (IllegalStateException e) {
                System.out.println("  push.get() -> ISE " + e.getMessage());
            }
        }

        System.out.println("B. allSuccessfulOrThrow");
        T0 = System.nanoTime();
        try (var scope = StructuredTaskScope.<String, Stream<Subtask<String>>>open(
                StructuredTaskScope.Joiner.allSuccessfulOrThrow())) {
            scope.fork(() -> send("mail", 800, false));
            scope.fork(() -> send("sms",  800, false));
            scope.fork(() -> send("push", 0, true));
            System.out.println("  不应到这里 " + scope.join().toList());
        } catch (StructuredTaskScope.FailedException e) {
            System.out.printf("  join threw FailedException at +%4dms, cause=%s%n", ms(), e.getCause());
        }

        System.out.println("C. anySuccessfulResultOrThrow");
        T0 = System.nanoTime();
        try (var scope = StructuredTaskScope.<String, String>open(
                StructuredTaskScope.Joiner.anySuccessfulResultOrThrow())) {
            scope.fork(() -> send("mail", 600, false));
            scope.fork(() -> send("sms",  150, false));
            scope.fork(() -> send("push", 900, false));
            String winner = scope.join();
            System.out.printf("  winner=%s at +%4dms cancelled=%s%n", winner, ms(), scope.isCancelled());
        }

        System.out.println("D. allUntil：满两个成功就停");
        T0 = System.nanoTime();
        var ok = new AtomicInteger();
        try (var scope = StructuredTaskScope.<String, Stream<Subtask<String>>>open(
                StructuredTaskScope.Joiner.<String>allUntil(st -> {
                    if (st.state() == Subtask.State.SUCCESS && ok.incrementAndGet() == 2) {
                        System.out.printf("  [pred] +%4dms 第二个成功，触发停止%n", ms());
                        return true;
                    }
                    return false;
                }))) {
            scope.fork(() -> send("mail", 100, false));
            scope.fork(() -> send("sms",  200, false));
            scope.fork(() -> send("push", 800, false));
            var done = scope.join().toList();
            System.out.printf("  收到 %d 个 subtask，states=%s cancelled=%s%n",
                    done.size(), done.stream().map(Subtask::state).toList(), scope.isCancelled());
        }
    }
}

本机输出（Temurin 25.0.5+8，--enable-preview，load average 124）：

A. awaitAll
  [push] + 140ms 503
  [mail] + 314ms ok
  [sms ] + 314ms ok
  join=null states=SUCCESS/SUCCESS/FAILED
  push.exception()=java.lang.IllegalStateException: push 下游 503
  push.get() -> ISE Result is unavailable or subtask did not complete successfully
B. allSuccessfulOrThrow
  [push] + 131ms 503
  [mail] + 133ms INTERRUPTED
  [sms ] + 133ms INTERRUPTED
  join threw FailedException at + 134ms, cause=java.lang.IllegalStateException: push 下游 503
C. anySuccessfulResultOrThrow
  [sms ] + 153ms ok
  [mail] + 154ms INTERRUPTED
  [push] + 154ms INTERRUPTED
  winner=sms-ok at + 154ms cancelled=true
D. allUntil：满两个成功就停
  [mail] + 109ms ok
  [sms ] + 201ms ok
  [pred] + 204ms 第二个成功，触发停止
  [push] + 204ms INTERRUPTED
  收到 3 个 subtask，states=[SUCCESS, SUCCESS, UNAVAILABLE] cancelled=true

逐段读：

  • A：push 在 140ms 就失败了，但 join 等到 314ms 才返回，mail/sms 全程没被打断。join 返回 null，失败只能通过 push.state() == FAILED 发现。同一个 Subtask 上，exception() 给出原始 IllegalStateException，get() 却只给一句 ISE。这就是 15.1 那个 410ms 的来源。
  • B：push 在 131ms 失败，mail/sms 在 133ms 被中断——取消延迟 2ms，join 在 134ms 抛 FailedException，getCause() 就是原始异常，类型和信息都没丢。800ms 的两个任务实际只活了 133ms，总耗时从 800ms 降到 134ms。
  • C：sms 153ms 先成功，join 立刻返回它的结果，另外两个在 154ms 被打断，isCancelled() 为 true。这是「多副本查询取最快」的正确写法。
  • D：谓词在第二个成功时返回 true，剩下的 push 立刻被中断。注意返回的 Stream 里有三个 Subtask，状态是 [SUCCESS, SUCCESS, UNAVAILABLE]——它返回的是「这次 join 涉及的全部子任务」，不是「成功的那几个」，过滤要自己做。

取消语义的三个细节

一、取消靠中断，不靠杀线程。 上面每次 INTERRUPTED 都是子任务自己 catch (InterruptedException) 打出来的。如果子任务干的是不可中断的活（紧循环不检查标志、阻塞在 synchronized 里等锁、IO 不响应中断），它不会停，join 会一直等到它结束。作用域只能保证「关闭返回时没有活线程」，代价是 join/close 会一直等下去；能不能及时停下，取决于子任务是否配合中断——这条前提和 6.2 的两阶段终止一样。

二、owner 被中断时，join 不取消子任务。 本机实测：主线程在 join() 里被 interrupt()，join 抛 InterruptedException，此刻 scope.isCancelled() 仍是 false；真正的中断发生在随后的 close()。所以正确写法是 catch (InterruptedException e) { Thread.currentThread().interrupt(); } 交给 try-with-resources 关作用域，而不是自己 fork 一堆取消逻辑。

三、FailedException 只带一个 cause。 三个子任务同时失败时，join 抛出的 FailedException 只保留第一个失败作为 cause，本机实测 getSuppressed().length == 0，另一个异常没有被挂上去。依赖「所有失败都能看到」的监控会漏报，需要完整失败列表时只能在每个子任务里自己记录。

Subtask 的状态语义也因此分成两路：FAILED 表示「跑过并且抛了异常」，UNAVAILABLE 表示「没跑完或被取消」。前者能用 exception() 取原始异常，后者不能。判断「是不是被取消」不要看 state()，看 scope.isCancelled() 更直接。

在 libraryd-conc 里落地

把通知扇出单独抽成 StructuredNotifier：三个下游 mail、sms、push，全部成功才算通知成功，任一失败立刻取消其余——避免用户收到一半通知、系统却报错。

package com.plume.conc.core;

import java.util.List;
import java.util.Map;
import java.util.concurrent.StructuredTaskScope;
import java.util.stream.Stream;

/** 通知扇出：mail / sms / push 三个下游，任一失败即取消其余。 */
public final class StructuredNotifier {

    public interface Channel {
        void send(String memberId, String text) throws Exception;
    }

    private static final List<String> CHANNELS = List.of("mail", "sms", "push");

    private final Map<String, Channel> channels;

    public StructuredNotifier(Map<String, Channel> channels) {
        this.channels = channels;
    }

    /** 三个下游全部成功才返回；任何一个失败，join 立刻抛 FailedException 并取消其余。 */
    public void fanout(String memberId, String text) {
        try (var scope = StructuredTaskScope.<String, Stream<StructuredTaskScope.Subtask<String>>>open(
                StructuredTaskScope.Joiner.allSuccessfulOrThrow(),
                cfg -> cfg.withThreadFactory(Thread.ofVirtual().name("notify-", 0).factory()))) {

            for (String name : CHANNELS) {
                Channel channel = channels.get(name);
                scope.fork(() -> {
                    channel.send(memberId, text);
                    return name;
                });
            }

            scope.join().toList();        // 全部成功才会走到这里
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new LoanException(ErrorCode.INTERRUPTED, "通知扇出被中断");
        } catch (StructuredTaskScope.FailedException e) {
            Throwable cause = e.getCause();
            if (cause instanceof LoanException loan) {
                throw loan;
            }
            throw new LoanException(ErrorCode.CONFLICT, "通知扇出失败: " + cause);
        }
    }
}

fanout 返回 void，因为调用方只关心「三个都发出去没有」。如果业务要区分「哪个渠道失败」，就得给 Channel 加返回值并在 catch 里读 exception()——但那样语义就变成 15.1 的 awaitAll 了，取舍见下一节。

mvn -q compile -Dmaven.compiler.enablePreview=true
java --enable-preview --source 25 -cp target/classes com.plume.conc.core.DemoNotifier

本机输出（三个渠道正常，然后 push 在第 120ms 抛 503）：

  [sms ] +  19ms start virtual=true name=notify-1
  [mail] +  19ms start virtual=true name=notify-0
  [push] +  19ms start virtual=true name=notify-2
  [push] + 849ms ok
  [mail] + 849ms ok
  [sms ] + 849ms ok
all-ok   + 851ms
  [mail] +   0ms start virtual=true name=notify-0
  [sms ] +   0ms start virtual=true name=notify-1
  [push] +   0ms start virtual=true name=notify-2
  [push] + 131ms 503
  [mail] + 138ms INTERRUPTED
  [sms ] + 138ms INTERRUPTED
fanout   + 138ms code=CONFLICT msg=push 下游 503

注意最后一行：原始 LoanException 穿过 FailedException 回到了调用方，ErrorCode.CONFLICT 和消息都没变。前提是 catch 里显式解包 getCause()；不写这段，抛出去的就是 FailedException，上层按 LoanException.code() 分流的逻辑全部失效。这是 15.3 要展开的迁移陷阱之一。

什么时候不要用

  • 需要拿到部分成功结果：allSuccessfulOrThrow 一失败就全取消，已经发出去的 mail 无法回滚。要「尽力而为」就该用 awaitAll 收集每个渠道的状态，自己决定重试哪几个。
  • 下游之间有先后依赖：sms 要用 mail 返回的模板 ID，这不是扇出而是流水线，作用域只能并行独立任务，串行依赖用 thenCompose 或直接顺序写。
  • 失败必须全部上报：前面说过 FailedException 只带一个 cause。要完整失败列表，anySuccessfulResultOrThrow / allSuccessfulOrThrow 都不合适，用 awaitAll + 遍历 Subtask.exception()。
  • 取消代价高于等待：下游是幂等的短查询时，取消省下的 600ms 可能不值一次连接重建。取消本身有成本，别为了取消而取消。

常见错误与边界

一、用 get() 判断失败。 FAILED 的 get() 抛 IllegalStateException，把你精心设计的错误码全吞了。先看 state()，再决定走 get() 还是 exception()。

二、以为返回的 Stream 只有成功的子任务。 顺序是 fork 顺序（本机实测，与完成先后无关），但 allUntil 的流里还包含被取消的 UNAVAILABLE 子任务。直接 map(Subtask::get) 会在那一项上抛 IllegalStateException，必须先按状态过滤。

三、在 catch 里忘了重设中断标志。 join 抛 InterruptedException 时会清掉中断位，不 Thread.currentThread().interrupt() 恢复，上层就再也不知道自己被取消了。

四、以为取消是同步的。 join 返回/抛出时，被取消的子任务可能还在跑它的 finally。本机实测两个场景的取消延迟分别是 2ms 和 7ms，下游负载高时会更长。想确认「全部停下」只能等 close() 返回。

五、谓词里抛异常。 本机实测谓词不在 owner 线程上执行，而是在每个子任务完成时的回调里、由一个匿名虚拟线程执行；抛出的异常成为那个线程的未捕获异常，join 照常返回，问题被完全吞掉。谓词只读 state()，别在里面做可能失败的解析。

六、用 Joiner 做限流。 allUntil 的谓词是「够不够」，不是「最多几个并发」。要控制同时在跑的子任务数量，用 Semaphore（14.3）或分批 fork，Joiner 管不了这个。

小结

  • Joiner 决定 join() 何时返回、返回什么、是否取消其余；五个内置实现覆盖全成功、任一成功、条件停止三类语义。
  • awaitAll 不取消、不传播失败：失败子任务留在 FAILED 状态，get() 只抛 ISE，原始异常在 exception()（本机实测 314ms vs 140ms 的差距）。
  • allSuccessfulOrThrow 在首个失败后取消其余，本机实测取消延迟 2ms，总耗时 134ms，FailedException.getCause() 保留原始异常。
  • anySuccessfulResultOrThrow 取最快成功者并取消其余（实测 154ms 全部结束）；全失败时抛 FailedException，且只保留一个 cause（实测 suppressed=0）。
  • allUntil 由谓词触发停止，返回的 Stream 含全部子任务，被取消者状态为 UNAVAILABLE。
  • 取消基于中断，子任务不响应中断就停不下来；owner 被中断时 join 抛 InterruptedException 但此刻尚未取消，取消发生在 close()。

到这里「怎么写」已经清楚，剩下一件事：已有的 ExecutorService 代码怎么迁过来，哪些能迁、哪些迁了会更糟。15.3 用同一段扇出逻辑做前后对照，并给出迁移的硬边界。

阅读导航：上一节 15.1 StructuredTaskScope 的作用域 ｜ 下一节 15.3 从 ExecutorService 迁移的边界

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「books」更多文章

  1. 《Java 消息中间件·事件驱动架构》附录 B 架构决策记录模板
  2. 《Java 消息中间件·事件驱动架构》附录 A 术语表
  3. 《Java 消息中间件·事件驱动架构》15.3 全书验收