java

5.3 超时与取消传播

讲清 timeout(Duration) 的真实语义——首元素超时与元素间隔超时，而不是整个流的总时长；对比 take(Duration) 的完成语义。用实测代码展示取消信号如何向上游传播、doFinally 放在不同位置看到的信号差异、Disposable/CompositeDisposable 的生命周期管理，以及 timeout 无法中断同步阻塞调用的原因与替代方案。示例基于 reactor-core 3.8.8。

Leeting Yan 6 分钟阅读 2638 字

timeout 的时钟从订阅开始

timeout(Duration) 经常被误解为「整个流最多执行这么久」。它实际管的是两段等待：

  1. 从订阅到第一个元素；
  2. 相邻两个元素之间。

每次收到 onNext，计时器重新开始。所以下面这段代码不会超时：

Flux.interval(Duration.ofMillis(500))
    .take(10)
    .timeout(Duration.ofSeconds(2))
    .subscribe(System.out::println,
        e -> System.out.println("Error: " + e.getClass().getSimpleName()));

interval 每 500ms 发一个元素，间隔远小于 2 秒。即使整个流要跑 5 秒，timeout 也不会触发。

反过来，如果第一个元素就迟到，或者两个元素之间卡住超过阈值，就会发出 TimeoutException：

Flux.never()
    .timeout(Duration.ofMillis(100))
    .subscribe(
        v -> System.out.println("onNext " + v),
        e -> System.out.println("onError " + e.getClass().getSimpleName()));

实测输出：

onError TimeoutException

Flux.never() 永远不会发元素，计时器从订阅时启动，100ms 后直接超时。

整个流的总时长用 take(Duration)

要限制「这条流总共最多跑 N 秒」，用 take(Duration)：

AtomicInteger count = new AtomicInteger();

Flux.interval(Duration.ofMillis(100))
    .take(Duration.ofMillis(550))
    .subscribe(
        v -> count.incrementAndGet(),
        e -> System.out.println("onError " + e),
        () -> System.out.println("onComplete, count=" + count.get()));

实测 550ms 窗口内发出了 5 个元素（0–4），然后正常完成：

onComplete, count=5

两者的终止语义完全不同：

需求操作符到达下游的终止信号
单个元素/元素间隔超时timeout(Duration)onError(TimeoutException)
整个流的总时长take(Duration)onComplete（正常完成）
超时后切换备用流timeout(Duration, fallback)备用流自己的信号

timeout 的第二个参数是超时后备，不是异常兜底：

Mono<String> result = fetchRemote()
    .timeout(Duration.ofSeconds(2), Mono.just("cached"));

它只在「超时」时生效。上游自己抛出的错误仍然走 onError，要处理得再加 onErrorResume。

取消如何向上游传播

timeout 触发时并不是只向下游发一个错误就结束。它会先取消上游订阅，再通知下游：

Flux.interval(Duration.ofSeconds(1))
    .doOnCancel(() -> System.out.println("upstream cancelled"))
    .timeout(Duration.ofMillis(250))
    .subscribe(
        v -> System.out.println("onNext " + v),
        e -> System.out.println("onError " + e.getClass().getSimpleName()));

实测输出：

upstream cancelled
onError TimeoutException

doOnCancel 在 timeout 的上游，所以它看到的是取消；下游收到的是超时错误。这个顺序（先 cancel、后 onError）是本次实测的观察结果，写清理逻辑时不要依赖「错误一定先到」。

doFinally 放在哪，看到的信号不一样

doFinally 报告的是「它所在位置的上游如何终止」。同样一条流，放在 timeout 前后结果不同：

// 放在 timeout 之前（上游）
Flux.never()
    .doFinally(s -> System.out.println("upstream doFinally " + s))
    .timeout(Duration.ofMillis(100))
    .subscribe();

// 放在 timeout 之后（下游）
Flux.never()
    .timeout(Duration.ofMillis(100))
    .doFinally(s -> System.out.println("downstream doFinally " + s))
    .subscribe();

实测：

upstream doFinally cancel
downstream doFinally onError
场景doFinally 在操作符上游在下游
timeout 触发CANCELON_ERROR
take(Duration) 到期CANCELON_COMPLETE
dispose()CANCELCANCEL

这不是 bug，而是「取消」与「终止」是两件事：上游被取消，下游收到终止信号。做指标统计时先想清楚要数哪一侧。

Disposable：从外部取消

subscribe() 返回 Disposable，可以在任意线程调用 dispose()：

Disposable disposable = Flux.interval(Duration.ofMillis(100))
    .subscribe(System.out::println);

Thread.sleep(1000);
disposable.dispose();   // 停止接收，并向上游传播 cancel

dispose() 是幂等的，重复调用不会出错。取消后 disposable.isDisposed() 返回 true。

多个订阅可以放进一个组合对象统一管理：

import reactor.core.Disposable;
import reactor.core.Disposables;

Disposable.Composite composite = Disposables.composite();
composite.add(stream1.subscribe());
composite.add(stream2.subscribe());
composite.add(stream3.subscribe());

composite.dispose();   // 一次性取消全部

在长生命周期对象里，把 Disposable 存成字段，在 stop()/close() 中释放：

public final class DataPoller {

    private Disposable disposable;

    public void start() {
        disposable = Flux.interval(Duration.ofSeconds(5))
            .flatMap(i -> fetchData())
            .subscribe(
                this::process,
                error -> log.error("poll failed", error));
    }

    public void stop() {
        if (disposable != null) {
            disposable.dispose();
            disposable = null;
        }
    }
}

注意 start() 不应该重复调用；如果确实要支持重启，先 dispose() 旧订阅，避免两个轮询同时运行。

资源清理：using 与 doFinally

Flux.using 把资源的创建、使用、释放绑在一起，正常完成、出错、取消三种情况都会执行清理：

Flux.using(
        () -> {
            Connection conn = database.getConnection();
            System.out.println("open");
            return conn;
        },
        conn -> Flux.fromIterable(conn.query("SELECT * FROM users")),
        conn -> {
            conn.close();
            System.out.println("close");
        })
    .take(3)
    .subscribe(row -> System.out.println("row " + row));

实测输出：

open
row 1
row 2
row 3
close

take(3) 取满 3 行后取消上游，using 的清理函数仍然执行。这就是它比「手动在 doFinally 里 close」更可靠的地方：资源从创建到释放的配对关系由操作符保证。

如果清理本身是异步的（例如返回 Mono<Void> 的关闭 RPC），用 Flux.usingWhen(...)；它在清理完成之前不会结束下游。

doFinally 更适合做与资源无关的收尾：打点、日志、清空 ThreadLocal。

陷阱一：timeout 不能中断同步阻塞

下面的代码看起来「1 秒超时」，实际会阻塞到 callable 自己返回：

long t0 = System.currentTimeMillis();

Mono.fromCallable(() -> {
        Thread.sleep(1500);
        System.out.println("callable returned at +"
            + (System.currentTimeMillis() - t0) + "ms");
        return "done";
    })
    .timeout(Duration.ofMillis(200))
    .subscribe(
        v -> System.out.println("onNext " + v),
        e -> System.out.println("onError " + e.getClass().getSimpleName()));

实测（200ms 超时，1.5 秒的阻塞任务）：

callable returned at +1643ms
onNext done

TimeoutException 根本没有出现。原因在订阅顺序：timeout 先调用下游的 onSubscribe，下游默认在这里 request(Long.MAX_VALUE)，这个 request 一路传到 MonoCallableSubscription，同步执行了 callable.call()；而 timeout 要等 onSubscribe 返回之后才去订阅计时器。阻塞发生得比计时器启动还早，所以 200ms 的超时没有机会生效。

把阻塞任务交给 boundedElastic 后，timeout 能按时触发：

Mono.fromCallable(() -> {
        Thread.sleep(1500);
        return "done";
    })
    .subscribeOn(Schedulers.boundedElastic())
    .timeout(Duration.ofMillis(200))
    .subscribe(...);

实测任务在约 350ms 被中断，同时下游收到 TimeoutException。但这不是通用保证：这里的 sleep 恰好响应中断；JDBC 查询、部分 socket 读写并不会因为线程被中断就立刻停止。timeout 取消的是响应式订阅，不是底层 I/O。

可靠的超时必须由客户端自己支持：给 HTTP 客户端设置连接/读取超时，给 JDBC 设置 query timeout，或者用支持取消的异步 API。Reactor 的 timeout 只能作为最后一道护栏。

Mono.fromFuture 的取消语义

Mono.fromFuture(cf) 在取消时会调用 cf.cancel(true)，但 CompletableFuture 的取消不会中断已经在运行的任务：

CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
    sleep(800);
    return "done";
});

Mono.fromFuture(cf)
    .timeout(Duration.ofMillis(200))
    .subscribe(...);

实测 200ms 后下游收到超时；再过 600ms 查询状态：

future done=true cancelled=true

isCancelled() 为 true，但任务已经跑完，supplyAsync 里的代码没有被中断。取消一个 Future 和停止它背后的工作不是一回事。

陷阱二：timeout 与 retry 的顺序

timeout 和 retryWhen 谁在外层，决定了超时覆盖的是「单次尝试」还是「整组尝试」：

// A：每次尝试各有 1 秒
source.timeout(Duration.ofSeconds(1))
    .retryWhen(Retry.backoff(3, Duration.ofMillis(200)));

// B：整组重试（含退避）总共只有 1 秒
source.retryWhen(Retry.backoff(3, Duration.ofMillis(200)))
    .timeout(Duration.ofSeconds(1));

A 中 retryWhen 重新订阅的是整条 timeout 链，计时器随每次订阅重置；B 中计时器只启动一次，退避等待也会吃掉预算。对外部调用做单次超时用 A；给整个业务操作设总预算用 B，或者直接用 Mono.delay(...) 做截止时间。

陷阱三：cancel 是协作式的

cancel() 只是把取消信号传给上游，是否立刻停下来由上游实现决定：

  • Flux.interval、Flux.range 这类内置源会立即停止调度。
  • Mono.fromCallable 正在执行的代码不会被打断。
  • 自己写的 Publisher 必须检查取消标志，并在 cancel() 后停止发射。
  • Flux.create 的 sink 被取消后，继续调用 sink.next() 会返回失败；应在 sink.onCancel(...) 里释放外部资源。

这也是 Reactive Streams 规范的立场：§3.12、§3.13 只要求 Publisher 最终停止发信号、最终丢弃订阅者引用，不要求立即完成底层清理；§2.8 还明确要求订阅者准备好在取消后仍收到已请求的迟到 onNext。取消是请求，不是抢占。

小结

操作符/方法管什么终止表现
timeout(Duration)首元素与元素间隔onError(TimeoutException)
timeout(Duration, fallback)同上，超时切备用流备用流的信号
take(Duration)整条流的总时长onComplete
takeUntil(predicate)条件满足即停（包含命中元素）onComplete
dispose()外部取消订阅上游 CANCEL
Flux.using资源创建/释放配对完成、错误、取消都释放
doFinally收尾钩子取决于它在上游还是下游

记住三点：

  1. timeout(Duration) 是元素间隔超时，不是总时长。
  2. 取消信号向上传播，但不会中断同步阻塞代码；真正的超时要靠底层客户端支持。
  3. doFinally 放在 timeout 前后看到的信号不同，统计和清理要选对位置。

下一章进入调度器——控制操作符链在哪个线程上执行。

阅读导航

上一篇：5.2 指数退避重试与熔断信号

下一篇：6.1 调度器：单线程、弹性、并行与立即

继续阅读

探索更多技术文章

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

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南