行业知识
Spring Boot WebFlux:实现web（Server-Sent Events）事件异步推送
WebFlux
在Spring Boot中，Flux是一个重要的概念，它是Spring Framework 5.0以后引入的响应式编程框架WebFlux的核心组件之一。Flux是Reactor项目的一部分，它实现了Reactive Streams规范，用于处理异步、非阻塞的数据流。
与传统的Spring MVC不同，WebFlux不需要Servlet API，而是通过Reactor项目提供的Flux和Mono来构建响应式应用程序。Flux用于表示包含0到N个元素的异步序列，而Mono则表示包含0到1个元素的异步序列。这使得WebFlux可以在有限的资源下提高系统的吞吐量和伸缩性，从而在处理大量请求时表现出更好的性能。
在WebFlux中，所有的操作都是非阻塞的，并且基于Reactor的背压机制来控制数据流的速度，从而防止系统资源的过度消耗。这使得WebFlux在处理高并发请求时具有更好的稳定性和可扩展性。
在Spring Boot中使用Flux时，通常需要在控制器方法中返回一个Flux对象，以表示一个异步的数据流。然后，可以使用各种操作符来对这个数据流进行转换、过滤、聚合等操作，以满足业务需求。例如，可以使用map操作符将数据流中的每个元素转换为另一种形式，使用filter操作符过滤掉不符合条件的元素，使用reduce操作符将数据流中的所有元素聚合成一个结果等。
总之，Flux是Spring Boot中实现响应式编程的重要组件之一，它使得应用程序能够更好地处理高并发请求，提高系统的吞吐量和伸缩性。
Flux 的主要特性包括：
-
异步非阻塞：Flux 旨在以非阻塞的方式处理数据，这意味着它不会阻塞调用线程，而是立即返回一个表示异步计算结果的对象。这使得它能够在高并发环境下提供更好的性能和响应能力。
-
背压（Backpressure）：背压是响应式编程中的一个重要概念，它允许接收方控制发送方的数据发送速率，以防止数据过快产生而导致的资源耗尽。Flux 提供了背压机制，使得接收方可以通知发送方其处理能力，从而避免数据丢失或资源耗尽。
-
冷流与热流：Flux 中的流可以是冷的（Cold）或热的（Hot）。冷流会为每个订阅者重新生成数据，而热流则会广播数据给所有订阅者，无论订阅者何时订阅。
-
操作符：Flux 提供了一系列的操作符，这些操作符可以链式调用，用于对流中的数据进行转换、过滤、聚合等操作。这些操作符使得在流上执行复杂的逻辑变得简单且直观。
-
错误处理：Flux 支持错误处理机制，可以在流中出现错误时进行适当的处理，如重试、忽略错误或传播错误。
-
与其他响应式库的集成：Flux 可以与其他遵循 Reactive Streams 规范的响应式库无缝集成，如 RxJava、Rx.NET 等。
在 Spring Boot 应用程序中，Flux 通常用于构建响应式 Web 服务。你可以使用 Spring WebFlux（基于 Project Reactor）来创建异步的、非阻塞的 Web 控制器，这些控制器可以返回 Flux 或 Mono 对象作为响应。这种模型非常适合处理大量并发请求，特别是在微服务架构中。
WebFlux和sse
Flux是Spring Framework 5.0中引入的响应式编程库Project Reactor的核心组件之一。它是一个用于表示异步、非阻塞数据流的类型，你可以在其上应用各种操作符来进行数据转换、过滤和聚合等操作。在Spring Boot应用程序中，Flux通常用于构建响应式Web服务，返回给客户端的是一个数据流而不是一次性的数据包。这使得客户端可以持续接收服务器的更新，而不需要频繁地发起新的请求。
SSE（Server-Sent Events）是另一种在服务器端和客户端之间建立异步通信的机制。与WebSocket不同，SSE是一种单向通信协议，只允许服务器向客户端发送数据。当服务器有新的数据产生时，它会通过SSE连接将数据发送给客户端。SSE基于HTTP协议，这意味着它可以在现有的HTTP基础设施上工作，而不需要额外的配置或协议支持。此外，SSE使用简单的文本格式来表示传输的数据，这使得它在浏览器端的支持非常广泛，除了IE浏览器外，其他现代浏览器都支持SSE。
在Spring Boot应用程序中，你可以使用WebFlux框架来实现SSE。在WebFlux中，你可以返回一个类型为Flux的对象来创建一个SSE端点。ServerSentEvent是一个特殊的类型，用于表示服务器发送给客户端的事件。当这个Flux对象发出事件时，它们将作为SSE事件发送给客户端。
总结起来，Flux和SSE都是用于在服务器端和客户端之间建立异步通信的技术。Flux是一个更通用的响应式编程库，可以用于构建各种异步场景，而SSE则是一种专门用于服务器向客户端推送事件的机制。在Spring Boot应用程序中，你可以使用Flux来实现SSE端点，从而利用SSE的优势来提供实时数据更新给客户端。
教程
springboot 项目引入webflux依赖
org.springframework.boot spring-boot-starter-webflux
创建flux控制器，FluxController 类，一下代码是接入文心一言大模型web流式接口，并使用flux返回给前端。
import com.alibaba.fastjson.JSON;import lombok.extern.slf4j.Slf4j;import org.json.JSONArray;import org.json.JSONObject;import org.springframework.http.HttpHeaders;import org.springframework.http.MediaType;import org.springframework.web.bind.annotation.PostMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import org.springframework.web.reactive.function.client.WebClient;import reactor.core.publisher.Flux;import java.io.IOException;/** * @author tarzan */@RestController@RequestMapping@Slf4jpublic class FluxController { public static final String API_KEY = ""; public static final String SECRET_KEY = ""; private static final String ACCESS_TOKEN_URL = "https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_&client_secret=" + SECRET_KEY; private static final String CHAT_URL = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/eb-instant"; @PostMapping(path = "/event-stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux helloWorld() throws IOException { /* return Flux.fromArray("hello world".split(" ")) // 每个元素延迟500毫秒，以模拟逐字效果 .delayElements(Duration.ofMillis(500));*/ // 构造聊天请求参数 JSONObject chatPayload = new JSONObject(); JSONArray messagesArray = new JSONArray(); JSONObject message = new JSONObject(); message.put("role", "user"); message.put("content", "为什么现在的农村出来的大学生，混的都不好？"); messagesArray.put(message); chatPayload.put("messages", messagesArray); chatPayload.put("stream", true); StringBuffer answer=new StringBuffer(); return WEB_CLIENT.post().uri("rpc/2.0/ai_custom/v1/wenxinworkshop/chat/eb-instant?access_token=" + getAccessToken()).contentType(MediaType.APPLICATION_JSON) .bodyValue(chatPayload.toString()) .retrieve() .bodyToFlux(String.class) // 可能需要其他流处理，比如map、filter等 .map(data -> { String result=JSON.parseObject(data).getString("result"); answer.append(result); return result; }).doOnComplete(() -> { // 当Flux完成时，输出结束消息 System.out.println("处理完毕，流已关闭。"); System.out.println(answer); }); } private static final WebClient WEB_CLIENT = WebClient.builder().baseUrl("https://aip.baidubce.com").defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_STREAM_JSON_VALUE).build(); static String getAccessToken() { // 构建请求体 String requestBody = "grant_type=client_credentials" + "&client_&client_secret=" + SECRET_KEY; return WEB_CLIENT.post().uri("/oauth/2.0/token").bodyValue(requestBody).retrieve() // 假设服务器返回的是 JSON 格式的字符串 .bodyToMono(String.class) .map(data -> JSON.parseObject(data).getString("access_token") ).block(); }}前端代码如下：
es
Sorry，浏览器不支持WebSocket
$(function() { /* const eventSource = new EventSource("/event-stream"); const dataDiv = $('#data'); eventSource.onmessage = function(event) { console.log('Received message:'+event.data); dataDiv.append(event.data); }; eventSource.onerror = function(error) { console.error('EventSource failed:'+error); dataDiv.append(error); eventSource.close(); };*/ const dataDiv = $('#data'); // 准备要发送的数据 const data = new FormData(); data.append('key1', 'value1'); data.append('key2', 'value2');// 发送 POST 请求并处理响应 fetch('/event-stream', { method: 'POST', body: data }).then(response => { // 检查响应是否为 SSE 流 if (response.headers.get('Content-Type') === 'text/event-stream') { // 返回一个可读流以读取响应体 return response.body.getReader(); } else { throw new Error('Response is not an SSE stream'); } }).then(reader => { // 递归处理 SSE 事件的函数 function processSSEEvent(reader) { return reader.read().then(({done, value}) => { if (done) { console.log('SSE stream ended'); return; } // value 是一个包含数据的 Uint8Array const eventStr = new TextDecoder().decode(value).trim(); // const event = eventStr.trim(); // 处理事件（例如，解析为 JSON，如果事件是 JSON 格式） console.log('Received SSE event:', eventStr); const event = eventStr.replace(/^data:/, ''); dataDiv.append(event); // 继续读取下一个事件 return processSSEEvent(reader); }); } // 开始处理 SSE 事件 return processSSEEvent(reader); }).catch(error => { console.error('Error fetching SSE stream:', error); }) });前端代码接收flux流式返回结果，将响应结果不断输入到前端页面上。代码中注释部分EventSource，只支持get请求，如果是调用get请求的接口可以使用注释的代码，实现flux返回的结果。
补充
上面代码只适合接收纯文本的数据，如果是json数据请修改代码如下：
let aiType = chatType.val(); const data = new FormData(); data.append('qa', qa); fetch(`/chat/${user.id}/${aiType}`, { method: 'POST', body: data, // 你的POST数据 }).then(response => { if (!response.body || !response.body.getReader) { chatNormal(); throw new Error('Response does not support streaming'); } const reader = response.body.getReader(); let buffer = ''; let fullAnswer = ''; let isStart = true; return new ReadableStream({ start(controller) { function readChunk() { return reader.read().then(async ({done, value}) => { if (done) { controller.close(); chatNormal(); // 复制功能 copy(); // 分享功能 share(); return; } const chunk = new TextDecoder().decode(value); buffer += chunk; // 分割并处理每一行 const lines = buffer.split('\n'); for (let i = 0; i setTimeout(() => { fullAnswer = fullAnswer + word; subscribe(res, fullAnswer); resolve(); }, 25)); } } readChunk(); }, }); });EventSource简介
EventSource是HTML5中引入的一种新的API，它允许服务器向客户端推送实时事件。这种推送是基于HTTP协议的，并且使用一种特殊的MIME类型，即"text/event-stream"，这使得服务器能够发送一系列的事件到客户端。
EventSource的主要用途是实现服务器和客户端之间的实时通信。客户端通过创建一个EventSource对象并指定一个URL，就可以开始监听服务器在该URL上发送的事件。一旦服务器有新的事件数据要发送，它就会将这些数据以特定的格式（即"event: data"的形式）发送给客户端。客户端在接收到这些数据后，可以通过注册的事件处理函数来处理这些数据。
EventSource具有一些重要的特性，包括实时性、低延迟和易用性。由于它使用长连接的方式进行数据传输，因此相比于传统的轮询方式，它能够更加高效地传输数据。此外，由于它是基于HTTP协议的，因此它可以在各种不同的场景下使用，并且与WebSocket和长轮询等方式兼容。
总的来说，EventSource是一种非常有用的技术，它使得服务器能够实时地向客户端推送事件，从而提高了Web应用程序的实时性和响应性。
实现效果
下一篇：揭秘转运仓，概念解析与运作原理！
