Spring WebFlux之流式输出

🎉🎉🎉🎉🎉🎉
欢迎访问的个人博客：https://swzbk.site/，加好友，拉你入福利群
🎉🎉🎉🎉🎉🎉

流式输出（Streaming Output）是指将数据分块逐步发送给客户端，而不是一次性发送所有数据。这种方式特别适合处理大文件、实时数据或需要逐步展示的场景（如deepseek响应、语音、视频、日志等）。在springboot中通过Spring WebFlux实现。

1. Flux是什么？

• 定义：Flux是一个异步数据流处理库，用于生成、操作和消费数据流（类似集合，但支持异步和非阻塞操作）。
• 核心特点：
  • 背压（Backpressure）：消费者可以控制生产者的速度，避免数据过载。
  • 函数式编程：通过链式操作（如map、filter、flatMap）处理数据流。
  • 异步非阻塞：基于Reactor Netty实现高性能I/O，适合高并发场景。

2. Flux在后端Java中的作用

（1）处理异步与高并发

• 场景：微服务通信、实时数据处理（如消息队列、日志监控）、长连接（WebSocket）。
• 优势：通过异步非阻塞方式减少线程资源消耗，提升系统吞吐量。

（2）响应式Web开发

• 与Spring WebFlux结合：构建响应式REST API，支持HTTP/2和Server-Sent Events（SSE）。
• 示例代码：

  @GetMapping("/events")
  public Flux<Event> getEvents() {return Flux.interval(Duration.ofSeconds(1)).map(sequence -> new Event("Event " + sequence));
  }
  

  （每秒推送一个事件给客户端）

（3）背压管理

• 避免内存溢出：当消费者处理速度慢于生产者时，Flux通过背压机制暂停生产，确保系统稳定。

（4）数据流操作

• 丰富的操作符：支持过滤、转换、合并、重试等复杂逻辑。

  Flux.just(1, 2, 3).map(n -> n * 2).filter(n -> n % 3 == 0).subscribe(System.out::println); // 输出：6
  

3. 为什么选择Flux？

• 与Spring生态集成：无缝衔接Spring Boot、Spring Cloud，适合企业级应用。
• 轻量高效：相比传统阻塞IO，资源占用更少，适合云原生环境。
• 对比RxJava：Flux是Spring官方推荐的响应式库，更注重与Java生态的兼容性。

4. 案例实现

在 Java 中使用 OkHttp3 发送请求，并通过 Project Reactor 的 Flux 获取实时响应，通常适用于处理流式数据，比如服务器发送的实时更新或者大型数据块的逐步传输。以下为你详细介绍实现步骤并给出示例代码。

实现思路

1. 引入依赖：需要在项目中引入 OkHttp3 和 Project Reactor 的依赖。
2. 发送请求：使用 OkHttp3 发送 HTTP 请求。
3. 处理响应：将 OkHttp3 的响应流转换为 Flux 进行响应式处理。

4.1 添加依赖

如果你使用的是 Maven 项目，在 pom.xml 中添加以下依赖：

<dependencies><!-- OkHttp3 --><dependency><groupId>com.squareup.okhttp3</groupId><artifactId>okhttp</artifactId><version>4.9.3</version></dependency><!--1： webflux 会默认引入下面的单独reactor-core--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency><!-- 2：Project Reactor Core、单独引入 --><dependency><groupId>io.projectreactor</groupId><artifactId>reactor-core</artifactId><version>3.4.16</version></dependency>
</dependencies>

4.2 编写代码

你提供的代码片段是结合了 OkHttp3 发起异步请求和 Project Reactor 的 Flux 来处理响应结果。下面为你完善这个示例，并详细解释代码逻辑。

示例代码

import okhttp3.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;import java.io.IOException;public class OkHttpFluxExample {private static final OkHttpClient client = new OkHttpClient();public static Flux<String> makeApiCall(Request apiRequest) {return Flux.create(emitter -> {try {// 发起异步请求client.newCall(apiRequest).enqueue(new Callback() {@Overridepublic void onFailure(Call call, IOException e) {// 处理请求失败的情况System.err.println("请求 API 失败: " + e.getMessage());emitter.error(e);}@Overridepublic void onResponse(Call call, Response response) throws IOException {try (ResponseBody responseBody = response.body()) {if (response.isSuccessful() && responseBody != null) {// 假设响应是逐行文本数据，逐行发布到 Flux 中String[] lines = responseBody.string().split("\n");for (String line : lines) {emitter.next(line);}// 数据发布完成，发送完成信号emitter.complete();} else {// 处理响应不成功的情况String errorMessage = "请求 API 失败，响应码: " + response.code();System.err.println(errorMessage);emitter.error(new IOException(errorMessage));}}}});} catch (Exception e) {// 处理请求过程中发生的异常System.err.println("请求 API 时发生异常: " + e.getMessage());emitter.error(e);}});}public static void main(String[] args) {// 构建请求Request apiRequest = new Request.Builder().url("https://example.com/api") // 替换为实际的 API 地址.build();// 调用 makeApiCall 方法获取 FluxFlux<String> responseFlux = makeApiCall(apiRequest);// 订阅 Flux 并处理响应数据responseFlux.subscribe(// 处理每个接收到的元素line -> System.out.println("Received: " + line),// 处理错误error -> System.err.println("Error: " + error.getMessage()),// 处理完成信号() -> System.out.println("API 请求处理完成"));}
}

代码解释

1. makeApiCall 方法：
   • 该方法接收一个 Request 对象作为参数，返回一个 Flux<String> 对象。
   • 使用 Flux.create 创建 Flux，在其回调中使用 OkHttp3 的 enqueue 方法发起异步请求。
   • onFailure 方法：当请求失败时，打印错误信息并调用 emitter.error 方法将错误信号发送给 Flux 的订阅者。
   • onResponse 方法：
     • 若响应成功且响应体不为空，将响应体按行分割，逐行调用 emitter.next 方法将每行数据发布到 Flux 中。
     • 数据发布完成后，调用 emitter.complete 方法发送完成信号。
     • 若响应不成功，打印错误信息并调用 emitter.error 方法发送错误信号。
2. main 方法：
   • 构建一个 Request 对象，指定要请求的 API 地址。
   • 调用 makeApiCall 方法获取 Flux。
   • 使用 subscribe 方法订阅 Flux，处理接收到的元素、错误和完成信号。

注意事项

• 请将 https://example.com/api 替换为实际的 API 地址。
• 该示例假设响应是逐行文本数据，你可以根据实际情况调整数据处理逻辑。
• 在实际应用中，需要注意资源管理，例如关闭 OkHttp 客户端等。

🎉🎉🎉🎉🎉🎉
欢迎访问的个人博客：https://swzbk.site/，加好友，拉你入福利群
🎉🎉🎉🎉🎉🎉