JAVA WebFlux 响应数据丢失？Publisher 流关闭时机错误导致的坑
大家好，今天我们来聊聊在使用 Spring WebFlux 时，可能遇到的一个比较隐蔽但又非常棘手的问题：响应数据丢失，以及它背后常见的罪魁祸首——Publisher 流关闭时机错误。
WebFlux 凭借其非阻塞、响应式的特性，在处理高并发、IO 密集型场景时展现出强大的优势。然而，正因为其异步非阻塞的特性，也增加了开发的复杂性，稍有不慎就会掉入“坑”里。其中，Publisher 的管理，尤其是其关闭时机，就是经常被忽略但至关重要的一点。
WebFlux 的响应式流基础
在深入探讨数据丢失问题之前，我们先回顾一下 WebFlux 的核心概念：响应式流。 WebFlux 基于 Reactive Streams 规范构建，该规范定义了四个核心接口：
- Publisher: 数据源，负责发布数据。
- Subscriber: 数据消费者，负责接收和处理数据。
- Subscription: Publisher 和 Subscriber 之间的连接，负责控制数据流的速率。
- Processor: 同时实现了 Publisher 和 Subscriber 接口，可以对数据进行转换和处理。
WebFlux 使用 Reactor 作为其响应式库，Reactor 提供了 Mono 和 Flux 两种 Publisher 实现：
- Mono: 表示包含零个或一个元素的异步序列。
- Flux: 表示包含零个或多个元素的异步序列。
在 WebFlux 中，Controller 方法通常返回 Mono 或 Flux，WebFlux 框架负责将这些响应式流转换为 HTTP 响应。例如：
@RestController
public class MyController {
@GetMapping("/data")
public Flux<String> getData() {
return Flux.just("Data 1", "Data 2", "Data 3");
}
}在这个例子中，getData 方法返回一个 Flux<String>，它会发布三个字符串元素。WebFlux 框架会自动将这些元素转换为 HTTP 响应体。
数据丢失的场景分析
数据丢失通常发生在以下几种场景：
- 过早关闭 Publisher: Publisher 在所有数据发布完成之前就被关闭，导致 Subscriber 只能接收到部分数据。
- 异常未处理导致流中断: Publisher 在数据发布过程中发生异常，如果没有正确处理，会导致流中断，Subscriber 无法接收到后续数据。
- 背压处理不当: Subscriber 的处理速度慢于 Publisher 的发布速度，导致背压机制触发，如果没有正确处理背压，可能会导致数据丢失。
- 中间操作符的副作用: 使用
map,filter等中间操作符时，如果操作符内部的逻辑抛出异常或者有副作用，可能会导致数据丢失。
接下来，我们将针对前两种场景，进行详细的分析和示例。
场景一：过早关闭 Publisher
这是最常见的数据丢失原因。通常发生在以下情况：
- try-with-resources 错误使用: 在 try-with-resources 块中创建 Publisher，导致 try 块结束后 Publisher 被关闭。
- 手动调用
dispose()或cancel(): 在不应该关闭 Publisher 的时候，错误地调用了dispose()或cancel()方法。 - 异步操作完成时未正确处理 Publisher: 在异步操作完成时，没有确保 Publisher 已经发布完所有数据，就提前关闭了 Publisher。
示例代码 (错误示范):
@GetMapping("/data")
public Flux<String> getData() {
try (Flux<String> flux = Flux.just("Data 1", "Data 2", "Data 3")) {
return flux; // 错误：flux 在 try 块结束后会被关闭
}
}在这个例子中，Flux.just 创建的 flux 对象在 try-with-resources 块中，当 try 块结束时，flux 会被自动关闭。这意味着，虽然 getData 方法返回了 flux，但是当 WebFlux 框架尝试从 flux 中读取数据时，flux 已经关闭，导致数据丢失。
正确做法:
不要在 try-with-resources 中创建 Publisher，或者确保 Publisher 在 try 块结束后仍然可以访问。
@GetMapping("/data")
public Flux<String> getData() {
Flux<String> flux = Flux.just("Data 1", "Data 2", "Data 3");
return flux; // 正确：flux 不会被过早关闭
}更复杂的异步场景：
假设我们需要从数据库中异步读取数据，并将读取到的数据作为 Flux 返回。
@Autowired
private ReactiveMongoTemplate reactiveMongoTemplate;
@GetMapping("/dbData")
public Flux<MyData> getDbData() {
return reactiveMongoTemplate.findAll(MyData.class);
}在这个例子中，reactiveMongoTemplate.findAll 方法返回一个 Flux<MyData>，它会异步地从数据库中读取数据。只要 reactiveMongoTemplate 的生命周期正确管理，就不会出现过早关闭 Publisher 的问题。
但是，如果我们在异步操作中手动创建 Publisher，就需要格外小心。例如：
@GetMapping("/asyncData")
public Flux<String> getAsyncData() {
return Flux.create(emitter -> {
CompletableFuture.runAsync(() -> {
try {
emitter.next("Async Data 1");
emitter.next("Async Data 2");
Thread.sleep(100); // 模拟耗时操作
emitter.next("Async Data 3");
emitter.complete(); // 正确：在所有数据发布完成后调用 complete()
} catch (Exception e) {
emitter.error(e); // 发生异常时调用 error()
}
});
});
}在这个例子中，我们使用 Flux.create 方法手动创建一个 Publisher。Flux.create 接收一个 Consumer<FluxSink<T>>，我们可以在这个 Consumer 中手动发布数据。
关键点：
emitter.next(data): 发布数据。emitter.complete(): 表示所有数据已经发布完成，Publisher 正常关闭。emitter.error(e): 表示发生异常，Publisher 异常关闭。
如果忘记调用 emitter.complete()，会导致什么问题？
Subscriber 会一直等待，直到超时或者连接断开。WebFlux 默认的超时时间是 30 秒。
如果调用了 emitter.complete()，但是在 emitter.next() 之后，会导致什么问题？
emitter.complete() 之后的 emitter.next() 调用会被忽略，Subscriber 只能接收到部分数据。
因此，在使用 Flux.create 或 Mono.create 手动创建 Publisher 时，一定要确保在所有数据发布完成后调用 complete()，或者在发生异常时调用 error()。
场景二：异常未处理导致流中断
在响应式流中，如果 Publisher 在数据发布过程中发生异常，如果没有正确处理，会导致流中断，Subscriber 无法接收到后续数据。
示例代码 (错误示范):
@GetMapping("/errorData")
public Flux<String> getErrorData() {
return Flux.just("Data 1", "Data 2", "Data 3")
.map(data -> {
if (data.equals("Data 2")) {
throw new RuntimeException("Simulated Error");
}
return data;
});
}在这个例子中，map 操作符会抛出一个 RuntimeException，如果没有正确处理这个异常，会导致流中断，Subscriber 只能接收到 "Data 1"。
正确做法:
使用 onErrorReturn, onErrorResume, onErrorContinue 等操作符来处理异常。
onErrorReturn(fallbackValue): 当发生异常时，返回一个默认值。onErrorResume(fallbackPublisher): 当发生异常时，切换到另一个 Publisher。onErrorContinue(consumer): 当发生异常时，忽略当前元素，继续处理下一个元素。
示例代码 (使用 onErrorReturn):
@GetMapping("/errorData")
public Flux<String> getErrorData() {
return Flux.just("Data 1", "Data 2", "Data 3")
.map(data -> {
if (data.equals("Data 2")) {
throw new RuntimeException("Simulated Error");
}
return data;
})
.onErrorReturn("Error occurred");
}在这个例子中，当 map 操作符抛出异常时，onErrorReturn 操作符会返回 "Error occurred"，Subscriber 会接收到 "Data 1" 和 "Error occurred"。
示例代码 (使用 onErrorResume):
@GetMapping("/errorData")
public Flux<String> getErrorData() {
return Flux.just("Data 1", "Data 2", "Data 3")
.map(data -> {
if (data.equals("Data 2")) {
throw new RuntimeException("Simulated Error");
}
return data;
})
.onErrorResume(e -> Flux.just("Recovered Data 1", "Recovered Data 2"));
}在这个例子中，当 map 操作符抛出异常时，onErrorResume 操作符会切换到另一个 Flux，Subscriber 会接收到 "Data 1", "Recovered Data 1" 和 "Recovered Data 2"。
示例代码 (使用 onErrorContinue):
@GetMapping("/errorData")
public Flux<String> getErrorData() {
return Flux.just("Data 1", "Data 2", "Data 3")
.map(data -> {
if (data.equals("Data 2")) {
throw new RuntimeException("Simulated Error");
}
return data;
})
.onErrorContinue((e, data) -> {
System.err.println("Error processing data: " + data + ", error: " + e.getMessage());
});
}在这个例子中，当 map 操作符抛出异常时，onErrorContinue 操作符会忽略 "Data 2"，继续处理 "Data 3"，Subscriber 会接收到 "Data 1" 和 "Data 3"。同时，错误信息会被打印到控制台。
选择哪种异常处理方式取决于具体的业务需求。onErrorReturn 适合于提供一个默认值，onErrorResume 适合于切换到另一个数据源，onErrorContinue 适合于忽略错误数据。
背压处理
在响应式编程中，背压（Backpressure）是一个重要的概念。它指的是当 Subscriber 的处理速度慢于 Publisher 的发布速度时，Subscriber 向 Publisher 发出信号，要求 Publisher 降低发布速度。
WebFlux 提供了多种背压策略：
| 背压策略 | 描述 |
|---|---|
BUFFER |
缓存所有数据，直到 Subscriber 能够处理。可能会导致内存溢出。 |
DROP |
丢弃 Subscriber 无法处理的数据。 |
LATEST |
只保留最新的数据，丢弃旧的数据。 |
ERROR |
当 Subscriber 无法处理数据时，抛出异常。 |
IGNORE |
忽略 Subscriber 的背压请求，继续发布数据。可能会导致 Subscriber 崩溃。 |
示例代码:
@GetMapping("/backpressureData")
public Flux<Integer> getBackpressureData() {
return Flux.range(1, 1000)
.onBackpressureBuffer(100) // 使用 BUFFER 策略，缓存 100 个元素
.delayElements(Duration.ofMillis(1)); // 模拟 Publisher 发布速度快于 Subscriber 处理速度
}在这个例子中，onBackpressureBuffer(100) 指定使用 BUFFER 策略，缓存 100 个元素。如果 Publisher 发布速度过快，导致缓存满了，就会抛出 OverflowException。
选择哪种背压策略取决于具体的业务需求。BUFFER 适合于数据量不大，可以容忍一定的延迟的场景。DROP 和 LATEST 适合于对数据实时性要求高的场景。ERROR 适合于需要保证数据完整性的场景。IGNORE 一般不建议使用，除非你确定 Subscriber 能够处理所有数据。
中间操作符的副作用
在使用 map, filter 等中间操作符时，如果操作符内部的逻辑抛出异常或者有副作用，可能会导致数据丢失。
示例代码 (错误示范):
@GetMapping("/sideEffectData")
public Flux<String> getSideEffectData() {
return Flux.just("Data 1", "Data 2", "Data 3")
.map(data -> {
System.out.println("Processing data: " + data); // 副作用
return data.toUpperCase();
});
}在这个例子中，map 操作符内部的 System.out.println 语句有副作用。虽然这个副作用本身不会导致数据丢失，但是如果 map 操作符内部抛出异常，可能会导致副作用只执行了一部分，从而导致程序状态不一致。
正确做法:
尽量避免在中间操作符中使用副作用。如果必须使用副作用，需要确保副作用是幂等的，或者使用 doOnNext, doOnError, doOnComplete 等操作符来执行副作用。
doOnNext(consumer): 在每个元素发布之后执行 Consumer。doOnError(consumer): 在发生错误时执行 Consumer。doOnComplete(action): 在流完成时执行 Action。
示例代码 (使用 doOnNext):
@GetMapping("/sideEffectData")
public Flux<String> getSideEffectData() {
return Flux.just("Data 1", "Data 2", "Data 3")
.doOnNext(data -> System.out.println("Processing data: " + data)) // 使用 doOnNext 执行副作用
.map(data -> data.toUpperCase());
}在这个例子中，我们使用 doOnNext 操作符来执行副作用。doOnNext 操作符不会影响流的正常执行，即使 map 操作符抛出异常，doOnNext 操作符也会被执行。
总结和建议
今天我们讨论了 WebFlux 中响应数据丢失问题，以及 Publisher 流关闭时机错误导致的坑。 我们学习了如何避免过早关闭 Publisher，如何处理异常，如何处理背压，以及如何避免中间操作符的副作用。
以下是一些建议：
- 仔细阅读 Reactor 文档: Reactor 文档非常详细，包含了各种操作符的使用方法和注意事项。
- 编写单元测试: 编写单元测试可以帮助你发现潜在的问题，并确保你的代码能够正确处理各种情况。
- 使用调试工具: 使用调试工具可以帮助你跟踪数据流的执行过程，并找到数据丢失的原因。
- 监控你的应用程序: 监控你的应用程序可以帮助你及时发现问题，并采取相应的措施。
希望今天的分享能够帮助大家更好地理解 WebFlux，并避免数据丢失的问题。
最后的思考
WebFlux 的响应式编程模型带来了性能上的提升，但也增加了开发的复杂度。理解 Publisher 的生命周期管理，异常处理，以及背压处理是至关重要的。希望大家在实践中不断总结经验，避免踩坑，写出高质量的 WebFlux 应用。