Spring Boot WebFlux 中 WebSocket 生命周期解析

Spring Boot WebFlux 中的 WebSocket 提供了一种高效、异步的方式来处理客户端与服务器之间的双向通信。WebSocket 连接的生命周期包括连接建立、消息传输、连接关闭以及资源清理等过程。此外，为了确保 WebSocket 连接的稳定性和可靠性，我们可以加入重试机制，以处理断开或网络问题时自动重新连接。

1. WebSocket 连接建立

WebSocket 的连接是通过 HTTP 的 Upgrade 机制从普通的 HTTP/HTTPS 请求升级而来的。具体流程如下：

1.1 客户端请求 WebSocket 连接

客户端通过 ws:// 或 wss:// 协议来访问 WebSocket 服务器，并发送 HTTP Upgrade 请求头，要求服务器将连接升级为 WebSocket 协议：

GET /ws HTTP/1.1
Host: example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: random-generated-key
Sec-WebSocket-Version: 13

1.2 服务器端处理 WebSocket 连接

Spring WebFlux 通过 WebSocketHandler 来处理 WebSocket 请求。以下是一个简单的 WebSocketHandler 实现：

@Component
public class MyWebSocketHandler implements WebSocketHandler {@Overridepublic Mono<Void> handle(WebSocketSession session) {return session.receive().doOnNext(message -> System.out.println("Received: " + message.getPayloadAsText())).then();}
}

当服务器收到 HTTP Upgrade 请求后，它会检查 Sec-WebSocket-Key 并返回 Sec-WebSocket-Accept 进行握手，建立连接。

1.3 握手成功，连接建立

如果握手成功，服务器会返回 101 Switching Protocols 响应，表示 WebSocket 连接已建立：

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: (calculated key)

2. WebSocket 消息处理

连接建立后，WebSocket 进入消息传输阶段，包括消息的接收和发送。

2.1 消息接收

服务器端可以通过 WebSocketSession.receive() 方法来接收客户端发送的消息：

session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(msg -> System.out.println("Received: " + msg)).then();

session.receive() 返回一个 Flux<WebSocketMessage>，可以处理流式消息，每次接收到新消息时执行 doOnNext() 中的处理逻辑。

2.2 消息发送

服务器端可以通过 WebSocketSession.send() 方法发送消息给客户端：

Flux<String> messages = Flux.interval(Duration.ofSeconds(1)).map(i -> "Message " + i);
return session.send(messages.map(session::textMessage));

send() 方法接收一个 Publisher<WebSocketMessage>，可以使用 Flux 来生成消息流。textMessage() 方法用于创建文本消息。

3. WebSocket 连接关闭

WebSocket 连接可以由客户端、服务器或网络异常等原因主动关闭。连接关闭的主要方式如下：

3.1 正常关闭

  • 客户端主动关闭：客户端可以通过调用 WebSocket.close() 发送 Close Frame，服务器接收到后会关闭连接。
  • 服务器主动关闭：服务器通过 WebSocketSession.close() 关闭连接：
    session.close(CloseStatus.NORMAL);

3.2 异常关闭

  • 网络异常：如网络断开或客户端崩溃等，连接会被强制关闭。
  • 心跳超时：如果使用 ping/pong 机制检测 WebSocket 是否存活，超时未收到 pong 响应时，连接会关闭。
    session.send(Flux.just(session.pingMessage(ByteBuffer.wrap(new byte[0]))));

3.3 连接关闭后的处理

服务器可以使用 session.receive().doOnTerminate() 监听连接关闭事件，执行清理操作：

session.receive().doOnTerminate(() -> System.out.println("WebSocket connection closed")).then();

4. WebSocket 生命周期总结

阶段说明
连接建立客户端发起 WebSocket 连接请求，服务器接受并返回 101 Switching Protocols 响应，连接建立。
消息传输服务器和客户端可以双向传输文本或二进制消息。
连接关闭连接可由客户端、服务器、网络异常等原因关闭。
资源清理连接关闭后需要进行资源清理操作，如取消订阅、清理状态等。

5. 完整示例：WebFlux WebSocket 服务器

以下是一个完整的 WebSocket 服务器配置示例，展示了如何在 Spring Boot WebFlux 中配置 WebSocket：

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter;
import org.springframework.web.reactive.socket.server.support.WebSocketHandlerMapping;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import reactor.core.publisher.Flux;
import java.time.Duration;
import java.util.Map;@Configuration
public class WebSocketConfig {@Beanpublic WebSocketHandler webSocketHandler() {return session -> {Flux<String> output = Flux.interval(Duration.ofSeconds(1)).map(time -> "Server time: " + time);return session.send(output.map(session::textMessage));};}@Beanpublic WebSocketHandlerMapping handlerMapping(WebSocketHandler handler) {return new WebSocketHandlerMapping(Map.of("/ws", handler));}@Beanpublic WebSocketHandlerAdapter handlerAdapter() {return new WebSocketHandlerAdapter();}
}

说明：

  • WebSocketHandler 处理 WebSocket 连接，发送定时消息。
  • WebSocketHandlerMapping 将 /ws 端点映射到 WebSocketHandler。
  • WebSocketHandlerAdapter 用于适配 WebSocket 处理器。

6. 服务器端发起 WebSocket 连接

如果你希望服务器主动连接到其他 WebSocket 服务器，可以使用 WebSocketClient。Spring WebFlux 提供了 ReactorNettyWebSocketClient 来发起 WebSocket 连接。

6.1 示例：服务器端发起 WebSocket 连接

import org.springframework.stereotype.Service;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
import reactor.core.publisher.Mono;
import java.net.URI;@Service
public class WebSocketClientService {private final ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();public Mono<Void> connectToWebSocketServer() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).subscribe();return sendMessage;});}
}

6.2 在 Spring Boot 启动时自动连接

通过在 @PostConstruct 中调用连接方法，可以确保 WebSocket 客户端在 Spring Boot 启动时自动连接：

import jakarta.annotation.PostConstruct;
import org.springframework.stereotype.Component;@Component
public class WebSocketClientInitializer {private final WebSocketClientService webSocketClientService;public WebSocketClientInitializer(WebSocketClientService webSocketClientService) {this.webSocketClientService = webSocketClientService;}@PostConstructpublic void init() {webSocketClientService.connectToWebSocketServer().subscribe();}
}

7. WebSocket 连接重试机制

在 WebSocket 的生命周期中，由于网络问题或服务器错误，WebSocket 连接可能会中断。为了提高 WebSocket 连接的可靠性，我们可以为 WebSocket 客户端添加重试机制，以确保断开后能够重新连接。

7.1 使用 retry() 方法重试连接

WebFlux 提供了 retry() 方法来自动重试操作。以下是一个简单的重试机制示例：

import reactor.core.publisher.Mono;public class WebSocketClientService {private final ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();public Mono<Void> connectToWebSocketServer() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).subscribe();return sendMessage;}).retry(5);  // 最大重试5次}
}

在这个例子中，retry(5) 表示如果 WebSocket 连接失败，最多会重试 5 次。

7.2 使用 retryWhen() 实现自定义重试逻辑

我们还可以通过 retryWhen() 来实现更复杂的重试策略，例如设置重试间隔时间或根据错误类型决定是否重试：

import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;
import java.time.Duration;public class WebSocketClientService {private final ReactorNettyWebSocketClient client = new ReactorNettyWebSocketClient();public Mono<Void> connectToWebSocketServer() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).subscribe();return sendMessage;}).retryWhen(errors ->errors.zipWith(Flux.range(1, 5), (error, count) -> count)  // 重试次数.flatMap(retryCount -> Mono.delay(Duration.ofSeconds(retryCount)))  // 增加重试间隔);}
}

在这个例子中，retryWhen() 会根据错误进行自定义重试逻辑，设置每次重试间隔递增。

8. 连接关闭后的重试机制

为了确保连接在关闭后重新建立，我们可以监听连接关闭事件并尝试重试：

session.receive().doOnTerminate(() -> {System.out.println("WebSocket connection closed");reconnect();  // 重新连接}).then();private void reconnect() {connectToWebSocketServer().retry(3)  // 重试3次.subscribe();
}

8.1 完整的客户端重试代码

public Mono<Void> connectWithRetry() {return client.execute(URI.create("ws://example.com/socket"), session -> {Mono<Void> sendMessage = session.send(Mono.just(session.textMessage("Hello Server!")));session.receive().map(WebSocketMessage::getPayloadAsText).doOnNext(System.out::println).doOnTerminate(() -> reconnect())  // 连接关闭后重试.subscribe();return sendMessage;}).retryWhen(errors ->errors.zipWith(Flux.range(1, 5), (error, count) -> count).flatMap(retryCount -> Mono.delay(Duration.ofSeconds(retryCount))));
}

9. 结论

Spring Boot WebFlux 中 WebSocket 的生命周期包括：

  1. 连接建立：通过 HTTP Upgrade 握手建立 WebSocket 连接。
  2. 消息收发：服务器和客户端之间通过 receive() 和 send() 方法进行消息交换。
  3. 连接关闭：连接可以通过正常关闭、异常关闭或主动关闭的方式结束。
  4. 资源清理：连接关闭后需要进行资源清理操作，确保系统稳定。
  5. 重试机制：通过 retry() 和 retryWhen() 方法为 WebSocket 连接添加自动重试机制，提高连接的可靠性。

通过 WebSocket，Spring Boot WebFlux 提供了高效的异步通信方式，特别适合用于实时数据流应用。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/bicheng/72632.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【数据挖掘】异构图与同构图

【数据挖掘】异构图与同构图

在图论&#xff08;Graph Theory&#xff09;中&#xff0c;异构图&#xff08;Heterogeneous Graph&#xff09;和同构图&#xff08;Homogeneous Graph&#xff09;是两种不同的图结构概念&#xff0c;它们的主要区别在于节点和边的类型是否单一。 1. 异构图&#xff08;Hete…
阅读更多...
Golang实践录：go发布版本信息收集

Golang实践录：go发布版本信息收集

go发布版本信息收集。 背景 本文从官方、网络资料收罗有关go的发布历史概况。主要目的是能快速了解golang不同版本的变更。鉴于官方资料为英文&#xff0c;为方便阅读&#xff0c;使用工具翻译成中文&#xff0c;重要特性参考其它资料补充/修改。由于发布版本内容较多&#xf…
阅读更多...
【C++】： STL详解 —— set和map类

【C++】： STL详解 —— set和map类

目录 关联式容器 键值对 set set的概念 set的构造函数 set的使用 map map的概念 map的构造函数 map的使用 multiset multimap 关联式容器 C标准库提供了多种容器&#xff0c;用于高效管理和操作数据集合。这些容器可分为以下几类&#xff1a; 顺序容器&#xff08;…
阅读更多...
DeepSeek：构筑大数据平台底座的最优解

DeepSeek：构筑大数据平台底座的最优解

一、大数据平台底座的重要性 在数字化浪潮席卷全球的当下，数据已成为企业乃至整个社会最具价值的资产之一 。大数据平台底座作为数据处理和业务支撑的核心枢纽，其重要性不言而喻，犹如大厦的基石，关乎整个数据生态系统的稳定与发展。 从数据处理角度来看，随着互联网、物联…
阅读更多...
Minix OS的配置 SSH C程序编译

Minix OS的配置 SSH C程序编译

Minix3的下载 官网&#xff1a;https://www.minix3.org/ 安装 平台&#xff1a;VMware 开机后进入系统使用setup命令来配置和安装尽量配置一个DNS服务器&#xff0c;比如8.8.8.8 SSH 安装&#xff1a;pkgin install openssh 修改配置文件&#xff0c;需要&#xff1a; 修…
阅读更多...
ubuntu20 安装python2

ubuntu20 安装python2

1. 确保启用了 Universe 仓库 在某些情况下&#xff0c;python2-minimal 包可能位于 Universe 仓库中。你可以通过以下命令启用 Universe 仓库并更新软件包列表&#xff1a; bash复制 sudo add-apt-repository universe sudo apt update 然后尝试安装&#xff1a; bash复制…
阅读更多...
STM32---FreeRTOS中断管理试验

STM32---FreeRTOS中断管理试验

一、实验 实验目的&#xff1a;学会使用FreeRTOS的中断管理 创建两个定时器&#xff0c;一个优先级为4&#xff0c;另一个优先级为6&#xff1b;注意&#xff1a;系统所管理的优先级范围 &#xff1a;5~15 现象&#xff1a;两个定时器每1s&#xff0c;打印一段字符串&#x…
阅读更多...
docker利用docker-compose-gpu.yml启动RAGFLOW，文档解析出错【亲测已解决】

docker利用docker-compose-gpu.yml启动RAGFLOW，文档解析出错【亲测已解决】

0.问题说明 想要让RAGFLOW利用GPU资源跑起来&#xff0c;可以选择docker-compose-gpu.yml启动。&#xff08;但是官网启动案例是86平台的不是NVIDIA GPU的&#xff0c;docker-compose-gpu.yml又是第三方维护&#xff0c;所以稍有问题&#xff09; 1.问题 docker利用docker-c…
阅读更多...
【AI深度学习网络】卷积神经网络（CNN）入门指南：从生物启发的原理到现代架构演进

【AI深度学习网络】卷积神经网络（CNN）入门指南：从生物启发的原理到现代架构演进

深度神经网络系列文章 【AI深度学习网络】卷积神经网络&#xff08;CNN&#xff09;入门指南&#xff1a;从生物启发的原理到现代架构演进【AI实践】基于TensorFlow/Keras的CNN&#xff08;卷积神经网络&#xff09;简单实现&#xff1a;手写数字识别的工程实践 引言 在当今…
阅读更多...
【ThreeJS Basics 06】Camera

【ThreeJS Basics 06】Camera

文章目录 Camera 相机PerspectiveCamera 透视相机正交相机用鼠标控制相机大幅度转动&#xff08;可以看到后面&#xff09; 控制组件FlyControls 飞行组件控制FirstPersonControls 第一人称控制PointerLockControls 指针锁定控制OrbitControls 轨道控制TrackballControls 轨迹球…
阅读更多...
Linux | Ubuntu 与 Windows 双系统安装 / 高频故障 / UEFI 安全引导禁用

Linux | Ubuntu 与 Windows 双系统安装 / 高频故障 / UEFI 安全引导禁用

注&#xff1a;本文为 “buntu 与 Windows 双系统及高频故障解决” 相关文章合辑。 英文引文&#xff0c;机翻未校。 How to install Ubuntu 20.04 and dual boot alongside Windows 10 如何将 Ubuntu 20.04 和双启动与 Windows 10 一起安装 Dave’s RoboShack Published in…
阅读更多...
在 C++ 中，通常会使用 `#define` 来定义宏，并通过这种方式发出警告或提示。

在 C++ 中，通常会使用 `#define` 来定义宏，并通过这种方式发出警告或提示。

在 C++ 中，通常会使用 #define 来定义宏，并通过这种方式发出警告或提示。 如何实现 GBB_DEPRECATED_MSG 宏： 你可以通过以下方式定义一个宏，显示弃用警告： #include <iostream>// 定义一个宏，用来打印弃用警告 #define GBB_DEPRECATED_MSG(msg
阅读更多...
el-tree右键节点动态位置展示菜单；el-tree的节点图片动态根据节点属性color改变背景色；加遮罩层（opacity）

el-tree右键节点动态位置展示菜单；el-tree的节点图片动态根据节点属性color改变背景色；加遮罩层（opacity）

一、el-tree右键节点动态位置展示菜单 关键：@node-contextmenu="handleRightClick"与@node-click=“handleNodeClick” <div class="content"><el-tabs class="tabs" @tab-click="handleClick" v-model="Modal"…
阅读更多...
Leetcode 378-有序矩阵中第 K 小的元素

Leetcode 378-有序矩阵中第 K 小的元素

给你一个 n x n 矩阵 matrix &#xff0c;其中每行和每列元素均按升序排序&#xff0c;找到矩阵中第 k 小的元素。 请注意&#xff0c;它是 排序后 的第 k 小元素&#xff0c;而不是第 k 个 不同 的元素。 你必须找到一个内存复杂度优于 O(n2) 的解决方案。 示例 1&#xff1…
阅读更多...
【二.提示词工程与实战应用篇】【3.Prompt调优：让AI更懂你的需求】

【二.提示词工程与实战应用篇】【3.Prompt调优：让AI更懂你的需求】

最近老张在朋友圈秀出用AI生成的国风水墨画，隔壁王姐用AI写了份惊艳全场的年终总结，就连楼下小卖部老板都在用AI生成营销文案。你看着自己跟AI对话时满屏的"我不太明白您的意思"，是不是怀疑自己买了台假电脑？别慌，这可能是你的打开方式不对。今天咱们就聊聊这个…
阅读更多...
UNIAPP前端配合thinkphp5后端通过高德API获取当前城市天气预报

UNIAPP前端配合thinkphp5后端通过高德API获取当前城市天气预报

如何通过 UniApp 前端项目与 ThinkPHP5 后端结合高德天气 API 获取天气预报信息。我们将分为前端和后端两部分进行实现。以下是一个完整的代码. 一、项目结构 project/ ├── frontend/ (UniApp 项目) │ ├── pages/ │ │ └── weather/ │ │ ├── in…
阅读更多...
蓝桥杯C组真题——巧克力

蓝桥杯C组真题——巧克力

题目如下 思路 代码及解析如下 谢谢观看
阅读更多...
CSDN博客写作教学（五）：从写作到个人IP的体系化构建（完结篇）

CSDN博客写作教学（五）：从写作到个人IP的体系化构建（完结篇）

导语 （第一篇）Markdown编辑器基础 （第二篇）Markdown核心语法 （第三篇）文章结构化思维 （第四篇）标题优化与SEO实战 通过前四篇教程，你已掌握技术写作的“术”——排版、标题、流量与数据。但真正的价值在于将技能升维为“道”：用技术博客为支点，撬动个人品牌与职业发…
阅读更多...
Elasticsearch简单学习

Elasticsearch简单学习

1、依赖的导入 <!--ES依赖--> <dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-high-level-client</artifactId> </dependency>2、客户端链接 RestHighLevelClient client new RestHigh…
阅读更多...
macOS Sequoia 15.3 M3 Pro芯片 iOS 开发环境配置记录（最新）

macOS Sequoia 15.3 M3 Pro芯片 iOS 开发环境配置记录（最新）

进行如下工作之前首先确保终端已翻墙&#xff0c;在ClashX选择“复制终端代理命令”&#xff0c;在终端进行粘附并执行。 安装 homebrew Homebrew 是 Mac 平台的一个包管理工具&#xff0c;提供了许多Mac下没有的Linux工具等。 /bin/bash -c "$(curl -fsSL https://raw…
阅读更多...
最新文章

Copyright @ 2022~2023