01 背景与核心概念
01.1 什么是 finally 处理
在响应式流中，finally 处理用于确保无论数据流正常结束、遇到错误还是被取消订阅，都会执行清理逻辑。它像一个最终清场的钩子，确保资源的正确释放，以避免内存泄漏与对外部系统的连接残留。通过引入 doFinally 等操作符，可以把清理代码与数据流的生命周期绑定起来。
与传统的 try-catch-finally 不同，响应式流中的 finally 处理并非简单的同步块，而是在信号闭包触发时执行，覆盖了 ON_COMPLETE、ON_ERROR 和 CANCEL 三大信号。信号驱动的清理机制使得资源释放更加可靠。
在设计层面，finally 处理帮助我们实现更稳健的资源管理、断路保护和对外部依赖的正确回收，尤其是在分布式环境和高并发场景中。订阅者取消与后台线程退出时，仍能触发最终清理。
02 原理解析：finally 的工作原理
02.1 信号生命周期与资源清理
响应式流以信号驱动的方式推进：ON_NEXT/ON_ERROR/ON_COMPLETE 等事件代表数据流的不同阶段，而 doFinally 监听这些信号的任意结束情形，并执行绑定的清理逻辑。取消（CANCEL）信号也会触发最终操作，这与某些只在完成或错误时才执行的清理不同。
原理上，finally 的执行点通常在流的终止之前被安排，确保无论出现哪种终止路径，清理逻辑都能可靠执行。这种模式有助于避免延迟释放、资源未关闭等问题，提升系统鲁棒性。 底层实现通常通过订阅终止回调或使用专门的资源管理算子实现。
在设计复杂的流时，开发者需要清晰区分清理动作和错误处理操作，避免清理逻辑与恢复逻辑相互干扰，确保 finally 的执行对系统状态是一致且可预测的。 清理应尽量原子化、幂等且无副作用。
03 在主流框架中的实现与对比
03.1 Reactor 的 doFinally 与资源释放
在 Project Reactor 中，doFinally 是最常用的最终信号处理点。它会在所有信号路径（完成、错误、取消）之后执行，适合用于关闭连接、释放缓冲区、关闭计数器等资源。下面给出一个简单示例：关闭资源的统一入口。
import reactor.core.publisher.Flux;Flux<Integer> flux = Flux<Integer>.range(1, 5).doFinally(signalType -> System.out.println("Final signal: " + signalType));
在实际场景中，可以将 doFinally 与资源管理组合，例如配合 using 或自定义资源池，确保即使流被取消也能正确释放资源。 该模式对数据库连接、文件句柄等容易泄漏的资源尤为关键。
03.2 RxJava 的 doFinally 与兼容性
RxJava 的 doFinally 与 Reactor 的实现思想相近，都是为了在终止时执行清理逻辑。doFinally 在 RxJava 2/3 中能够处理三种信号：ON_COMPLETE、ON_ERROR、ON_DISPOSE（取消）。这是跨版本的稳定点。
import io.reactivex.Observable;Observable<Integer> source = Observable.range(1, 5).doFinally(() -> System.out.println("Finally in RxJava"));
结合 RxJava 的反应式特性，可以在数据流中嵌入资源释放、断路策略以及对外部服务的幂等性处理，保证最终态的一致性。 在多线索或跨进程场景下，统一的 finally 行为尤为重要。
04 错误处理与恢复策略
04.1 onErrorResume 与兜底数据
错误发生时，响应式流通常需要“兜底”策略来保持系统的可用性。这时可以使用 onErrorResume、onErrorReturn 或 onErrorMap 等操作符，将错误转换为备用数据源，或者进行错误重试前的降级处理。需要注意的是，finally 仍会在错误处理完成后执行，提供统一的清理入口。
import reactor.core.publisher.Flux;Flux<String> safeStream = Flux<String>.just("A", "B", "C").flatMap(v -> mayFail(v)).onErrorResume(e -> Flux<String>.just("default1", "default2")).doFinally(sig -> System.out.println("Cleanup after error: " + sig));
在上面的示例中，错误被兜底为默认数据流，同时最终清理在 doFinally 中完成。这种组合模式能有效降低单点失败对系统的冲击。
04.2 doFinally 之后的清理行为
即便使用了 onErrorResume 等恢复策略，doFinally 的执行时机保持不变，意味着无论走的是原始路径还是兜底路径，资源释放都在终止阶段完成。这有助于避免“清理时机错位”造成的资源错用或重复释放的问题。 在设计清理语义时应明确信号边界。
import reactor.core.publisher.Flux;Flux<String> resilient = Flux<String>.just("x", "y").flatMap(s -> maybeFail(s)).onErrorResume(e -> Flux<String>.just("fallback")).doFinally(signal -> System.out.println("Finally after resilience: " + signal));
05 实战案例：温度数据流中的 finally 处理与错误修复（temperature=0.6）
05.1 案例背景与设计要点
场景设定为一个温度传感器数据流，temperature=0.6 表示在数据采样或故障注入时的阈值设定，用于模拟噪声和异常情况。目标是确保在传感器读取失败、网络中断或订阅被取消时，仍然能够正确释放传感器资源，并通过错误修复策略继续输出降级数据，保障系统的连续性。
设计要点包括：资源的显式开启与释放、错误的兜底处理、以及最终状态的确定性日志；同时要确保 finally 的执行与降级路径一致，并且不引入重复清理。
05.2 案例代码：温度传感数据流
下面的示例展示如何在温度传感场景下，结合 finally 和错误恢复实现稳健的数据流。代码以 Java Reactor 风格为主，包含资源创建、数据转换、错误兜底以及最终清理。
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;class TemperatureReading {final long ts;final double value;TemperatureReading(long ts, double value) { this.ts = ts; this.value = value; }
}class SensorConnection {final String id;SensorConnection(String id) { this.id = id; }TemperatureReading readTemperature() throws Exception {// 模拟温度读数，带有随机错误double v = Math.random();if (v < 0.2) throw new RuntimeException("Sensor read error");return new TemperatureReading(System.currentTimeMillis(), 25.0 + (Math.random() * 5.0) * 0.6);}void close() { System.out.println("Sensor " + id + " closed"); }
}public class TemperatureStreamDemo {public static void main(String[] args) {SensorConnection sensor = new SensorConnection("temp-sensor-01");Flux<TemperatureReading> stream = Flux.interval(java.time.Duration.ofMillis(200)).map(t -> {try {return sensor.readTemperature();} catch (Exception e) {throw new RuntimeException(e);}}).onErrorResume(e -> {// 错误兜底：输出一个降级的温度数据TemperatureReading fallback = new TemperatureReading(System.currentTimeMillis(), Double.NaN);return Flux.just(fallback);}).doFinally(sig -> {// 确保资源释放sensor.close();System.out.println("Stream finished: " + sig);});// 订阅并打印，模拟消费端stream.subscribe(r => System.out.println("Reading: ts=" + r.ts + ", value=" + r.value));}
}
上述代码中：readTemperature() 可能抛出异常，使用 onErrorResume 进行兜底数据输出；doFinally 负责最终的资源关闭和日志记录，确保无论流因错误还是正常结束都会执行清理。
05.3 效果演示与解释
执行结果通常包含温度读数、降级数据以及关闭日志。本案例的关键点在于：最终清理与错误恢复在一个统一的流中协同工作，避免因异常导致资源泄露或数据停滞。通过将 finally 与 onErrorResume 组合，可以在遇到实际故障时继续向下游提供可用数据，同时保证传感器资源在结束后被正确释放。
06 常见问题与调试技巧
06.1 调试 finally 的执行顺序
在复杂的流中，doFinally 的执行顺序可能因为多处终止点而变得难以判定。建议在 finally 回调中只放置幂等且无副作用的清理逻辑，并通过明确的 日志和信号类型 来追踪结束路径，例如输出 Final signal 与具体信号类型。
为了更清晰地调试，可以在 doFinally 前后添加简单的 日志输出，并结合 onErrorResume 的兜底路径进行对比，确保无论走哪条分支都能触发最终清理。
通过以上章节的原理解析、实现对比及实战案例，我们可以清晰地理解响应式流中的 finally 处理与错误修复的结合方式，以及在实际生产环境中如何落地执行。
