如何捕获响应式流的取消信号？解决doFinally取消时不生效问题

我明白你的问题了——当第三方方法foo内部直接取消订阅时，外层的doFinally没被触发，这是因为Reactor里取消信号默认是向上游传播的，外层的doFinally没接收到这个终止信号。

解决方案：使用doOnCancel()捕获取消事件 #

最直接的办法是专门使用doOnCancel()操作符，它会监听流的取消事件，不管取消信号来自哪里（上游、下游或者内部订阅），只要流被取消就会执行你的收尾逻辑。如果需要同时处理正常完成、错误和取消三种场景，可以结合doOnCancel()和doFinally()使用：

public Mono<Void> myFunction() {
    return Mono.just("hello")
            .flatMap(s -> foo(s))
            .doOnCancel(() -> {
                // 专门处理取消时的收尾操作
                System.out.println("流被取消，执行收尾逻辑");
            })
            .doFinally(signalType -> {
                // 处理正常完成或错误的情况
                System.out.println("最终信号类型：" + signalType);
            });
}

// 第三方库方法，无法修改
public Mono<Void> foo(String s) {
    return Mono.just(s)
            .doOnSubscribe(subscription -> subscription.cancel())
            .then();
}

进阶：用usingWhen做严谨的资源管理 #

如果你的收尾操作涉及资源管理（比如关闭数据库连接、释放文件句柄等），推荐使用usingWhen操作符，它可以分别指定正常完成、错误、取消三种场景下的处理逻辑，确保资源一定会被清理：

public Mono<Void> myFunction() {
    // 模拟需要管理的资源（比如数据库连接）
    return Mono.usingWhen(
            Mono.just("my-resource"), // 资源获取逻辑
            resource -> Mono.just("hello").flatMap(this::foo), // 业务流
            resource -> {
                // 正常完成时的资源清理
                System.out.println("正常完成，清理资源：" + resource);
                return Mono.empty();
            },
            (resource, throwable) -> {
                // 错误发生时的资源清理（可选，若不需要单独处理可以省略）
                System.out.println("发生错误，清理资源：" + resource);
                return Mono.empty();
            },
            resource -> {
                // 取消时的资源清理
                System.out.println("流被取消，清理资源：" + resource);
                return Mono.empty();
            }
    );
}

为什么doFinally没生效？ #

根据Reactor的设计，取消信号是向上游传播的。当foo内部在doOnSubscribe中立即取消订阅时，这个取消信号会传递给flatMap上游的Mono.just("hello")，但外层的doFinally属于整个流的下游，无法捕获到这个向上传播的取消信号。而doOnCancel()是专门监听流的取消状态，不管信号传播方向，因此能可靠触发。

内容的提问来源于stack exchange，提问作者chenchang