解决websocket不定时出现1005错误

后台抛出异常如下:

Operator called default onErrorDropped
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalArgumentException: WebSocket close status code does NOT comply with RFC-6455: 1005
Caused by: java.lang.IllegalArgumentException: WebSocket close status code does NOT comply with RFC-6455: 1005

分析原因是:

spring cloud gateway 转发websocket请求无法监听到 close 事件 没有收到预期的状态码

解决方案:

在gateway进行请求拦截

代码如下:

@Slf4j
@Component
public class CustomWebsocketRoutingFilter implements GlobalFilter, Ordered {//Sec-Websocket protocolpublic static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol";//Sec-Websocket headerpublic static final String SEC_WEBSOCKET_HEADER = "sec-websocket";//http header schemapublic static final String HEADER_UPGRADE_WebSocket = "websocket";public static final String HEADER_UPGRADE_HTTP = "http";public static final String HEADER_UPGRADE_HTTPS = "https";private final WebSocketClient webSocketClient;private final WebSocketService webSocketService;private final ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider;// 不直接使用 headersFilters 用该变量代替private volatile List<HttpHeadersFilter> headersFilters;public CustomWebsocketRoutingFilter(WebSocketClient webSocketClient, WebSocketService webSocketService, ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider) {this.webSocketClient = webSocketClient;this.webSocketService = webSocketService;this.headersFiltersProvider = headersFiltersProvider;}/* for testing *///http请求转为ws请求static String convertHttpToWs(String scheme) {scheme = scheme.toLowerCase();return "http".equals(scheme) ? "ws" : "https".equals(scheme) ? "wss" : scheme;}@Overridepublic int getOrder() {// Before NettyRoutingFilter since this routes certain http requests//修改了这里 之前是-1 降低优先级return Ordered.LOWEST_PRECEDENCE - 2;}@Overridepublic Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {changeSchemeIfIsWebSocketUpgrade(exchange);URI requestUrl = exchange.getRequiredAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);String scheme = requestUrl.getScheme();if (ServerWebExchangeUtils.isAlreadyRouted(exchange) || (!"ws".equals(scheme) && !"wss".equals(scheme))) {return chain.filter(exchange);}ServerWebExchangeUtils.setAlreadyRouted(exchange);HttpHeaders headers = exchange.getRequest().getHeaders();HttpHeaders filtered = HttpHeadersFilter.filterRequest(getHeadersFilters(), exchange);List<String> protocols = getProtocols(headers);return this.webSocketService.handleRequest(exchange, new ProxyWebSocketHandler(requestUrl, this.webSocketClient, filtered, protocols));}/* for testing *///获取请求头里的协议信息List<String> getProtocols(HttpHeaders headers) {List<String> protocols = headers.get(SEC_WEBSOCKET_PROTOCOL);if (protocols != null) {ArrayList<String> updatedProtocols = new ArrayList<>();for (int i = 0; i < protocols.size(); i++) {String protocol = protocols.get(i);updatedProtocols.addAll(Arrays.asList(StringUtils.tokenizeToStringArray(protocol, ",")));}protocols = updatedProtocols;}return protocols;}/* for testing */List<HttpHeadersFilter> getHeadersFilters() {if (this.headersFilters == null) {this.headersFilters = this.headersFiltersProvider.getIfAvailable(ArrayList::new);// remove host header unless specifically asked not tothis.headersFilters.add((headers, exchange) -> {HttpHeaders filtered = new HttpHeaders();filtered.addAll(headers);filtered.remove(HttpHeaders.HOST);boolean preserveHost = exchange.getAttributeOrDefault(ServerWebExchangeUtils.PRESERVE_HOST_HEADER_ATTRIBUTE, false);if (preserveHost) {String host = exchange.getRequest().getHeaders().getFirst(HttpHeaders.HOST);filtered.add(HttpHeaders.HOST, host);}return filtered;});this.headersFilters.add((headers, exchange) -> {HttpHeaders filtered = new HttpHeaders();for (Map.Entry<String, List<String>> entry : headers.entrySet()) {if (!entry.getKey().toLowerCase().startsWith(SEC_WEBSOCKET_HEADER)) {filtered.addAll(entry.getKey(), entry.getValue());}}return filtered;});}return this.headersFilters;}static void changeSchemeIfIsWebSocketUpgrade(ServerWebExchange exchange) {// 检查版本是否适合URI requestUrl = exchange.getRequiredAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);String scheme = requestUrl.getScheme().toLowerCase();String upgrade = exchange.getRequest().getHeaders().getUpgrade();// change the scheme if the socket client send a "http" or "https"if (HEADER_UPGRADE_WebSocket.equalsIgnoreCase(upgrade) && (HEADER_UPGRADE_HTTP.equals(scheme) || HEADER_UPGRADE_HTTPS.equals(scheme))) {String wsScheme = convertHttpToWs(scheme);boolean encoded = ServerWebExchangeUtils.containsEncodedParts(requestUrl);URI wsRequestUrl = UriComponentsBuilder.fromUri(requestUrl).scheme(wsScheme).build(encoded).toUri();exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, wsRequestUrl);if (log.isTraceEnabled()) {log.trace("changeSchemeTo:[" + wsRequestUrl + "]");}}}//自定义websocket处理方式private static class ProxyWebSocketHandler implements WebSocketHandler {private final WebSocketClient client;private final URI url;private final HttpHeaders headers;private final List<String> subProtocols;ProxyWebSocketHandler(URI url, WebSocketClient client, HttpHeaders headers, List<String> protocols) {this.client = client;this.url = url;this.headers = headers;if (protocols != null) {this.subProtocols = protocols;} else {this.subProtocols = Collections.emptyList();}}@Overridepublic List<String> getSubProtocols() {return this.subProtocols;}@Overridepublic Mono<Void> handle(WebSocketSession session) {return this.client.execute(this.url, this.headers, new WebSocketHandler() {private CloseStatus adaptCloseStatus(CloseStatus closeStatus) {int code = closeStatus.getCode();if (code > 2999 && code < 5000) {return closeStatus;}switch (code) {case 1000://正常关闭return closeStatus;case 1001://服务器挂了或者页面跳转return closeStatus;case 1002://协议错误return closeStatus;case 1003://收到了不能处理的数据类型return closeStatus;case 1004:// 预留关闭状态码return CloseStatus.PROTOCOL_ERROR;case 1005:// 预留关闭状态码 期望收到状态码但是没有收到return CloseStatus.PROTOCOL_ERROR;case 1006:// 预留关闭状态码 连接异常关闭return CloseStatus.PROTOCOL_ERROR;case 1007://收到的数据与实际的消息类型不匹配return closeStatus;case 1008://收到不符合规则的消息return closeStatus;case 1009://收到太大的不能处理的消息return closeStatus;case 1010://client希望server提供多个扩展,server没有返回相应的扩展信息return closeStatus;case 1011://server遇到不能完成的请求return closeStatus;case 1012:// Not in RFC6455// return CloseStatus.SERVICE_RESTARTED;return CloseStatus.PROTOCOL_ERROR;case 1013:// Not in RFC6455// return CloseStatus.SERVICE_OVERLOAD;return CloseStatus.PROTOCOL_ERROR;case 1015:// 不能进行TLS握手 如:server证书不能验证return CloseStatus.PROTOCOL_ERROR;default:return CloseStatus.PROTOCOL_ERROR;}}/*** send      发送传出消息* receive   处理入站消息流* doOnNext  对每条消息做什么* zip       加入流* then      返回接收完成时完成的Mono<Void>*/@Overridepublic Mono<Void> handle(WebSocketSession proxySession) {Mono<Void> serverClose = proxySession.closeStatus().filter(__ -> session.isOpen()).map(this::adaptCloseStatus).flatMap(session::close);Mono<Void> proxyClose = session.closeStatus().filter(__ -> proxySession.isOpen()).map(this::adaptCloseStatus).flatMap(proxySession::close);// Use retain() for Reactor NettyMono<Void> proxySessionSend = proxySession.send(session.receive().doOnNext(WebSocketMessage::retain));Mono<Void> serverSessionSend = session.send(proxySession.receive().doOnNext(WebSocketMessage::retain));// Ensure closeStatus from one propagates to the otherMono.when(serverClose, proxyClose).subscribe();// Complete when both sessions are donereturn Mono.zip(proxySessionSend, serverSessionSend).then();}@Overridepublic List<String> getSubProtocols() {return CustomWebsocketRoutingFilter.ProxyWebSocketHandler.this.subProtocols;}});}}}

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

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

相关文章

vue 路由动态加载

在 Vue.js 中&#xff0c;可以使用 webpack 的动态导入语法来实现路由动态加载。下面是一个简单的示例&#xff1a; const Home () > import(/* webpackChunkName: "home" */ ./views/Home.vue); const About () > import(/* webpackChunkName: "about…

Flink的checkpoint是怎么实现的?

分析&回答 Checkpoint介绍 Checkpoint容错机制是Flink可靠性的基石,可以保证Flink集群在某个算子因为某些原因(如 异常退出)出现故障时,能够将整个应用流图的状态恢复到故障之前的某一状态,保证应用流图状态的一致性。Flink的Checkpoint机制原理来自“Chandy-Lamport alg…

第一百三十三回 StreamProvier

文章目录 概念介绍使用方法示例代码 我们在上一章回中介绍了通道相关的内容&#xff0c;本章回中将介绍 StreamProvider组件.闲话休提&#xff0c;让我们一起Talk Flutter吧。 概念介绍 在Flutter中Stream是经常使用的组件&#xff0c;对该组件的监听可以StremBuilder&#x…

Docker 的快速使用

ubuntu安装 centos安装 安装完毕之后执行一下这条命令&#xff0c;可以避免每次使用docker命令都需要sudo权限 sudo usermod -aG docker $USER阿里云docker镜像加速 DockerHub 遇到不懂或者不会使用的命令可以使用docker --help查看文档 docker --help 如&#xff1a; dock…

golang中map赋值

众所周知&#xff0c;golang中map是一个指针&#xff0c;既然是一个指针&#xff0c;那么参数传递、赋值应该都是指针传递&#xff0c;而下面的例子也印证了我的想法 package mainimport "fmt"func test_map2(m map[string]string) {fmt.Printf("inner: %v, %p…

OFDM 系统在 AWGN 信道下对不同载波频率偏移 (CFO) 的 BER 灵敏度研究(Matlab代码实现)

&#x1f4a5;&#x1f4a5;&#x1f49e;&#x1f49e;欢迎来到本博客❤️❤️&#x1f4a5;&#x1f4a5; &#x1f3c6;博主优势&#xff1a;&#x1f31e;&#x1f31e;&#x1f31e;博客内容尽量做到思维缜密&#xff0c;逻辑清晰&#xff0c;为了方便读者。 ⛳️座右铭&a…

互联网摸鱼日报(2023-09-01)

互联网摸鱼日报(2023-09-01) 36氪新闻 暑期剧综营销复盘&#xff1a;要么小单快返&#xff0c;要么长线绑定 ESG管理商「Wint」融资3500万美元、WeWork启动债务重组、北京规划机器人产业园&#xff5c;PropTech周刊73期 小米应用商店关闭红包专场&#xff0c;羊毛党遭遇痛击…

Metinfo6.0.0任意文件读取【漏洞复现】

文章目录 1.1、漏洞描述1.2、漏洞等级1.3、影响版本1.4、漏洞复现代码审计漏洞点 1.5、深度利用EXP编写 1.6、漏洞挖掘1.7修复建议 1.1、漏洞描述 漏洞名称&#xff1a;MetInfo任意文件读取 漏洞简介&#xff1a;MetInfo是一套使用PHP和MySQL开发的内容管理系统&#xff0c;其…

2023年Java核心技术第十二篇(篇篇万字精讲)

目录 22. AtomicInteger 底层实现原理是什么&#xff1f;如何在自己的项目代码中应用CAS操作&#xff1f; 22.1 典型回答 22.1.1 CAS详细解释&#xff1a; 22.1.1.1 预期值的选取&#xff1a; 22.1.2 CAS的弊端 22.1.2.1 ABA问题&#xff1a; 22.1.2.2 自旋次数限制&#xff1a…

Java实现根据商品ID获取京东商品详情数据,1688商品详情接口,1688API接口封装方法

要通过京东的API获取商品详情数据&#xff0c;您可以使用京东开放平台提供的接口来实现。以下是一种使用Java编程语言实现的示例&#xff0c;展示如何通过京东开放平台API获取商品详情&#xff1a; 首先&#xff0c;确保您已注册成为京东开放平台的开发者&#xff0c;并创建一…

基于网卡序号双网卡数据共享(网卡转发)

基于网卡序号&#xff1a;ifr.ifr_ifindex; 实现网卡之间的数据转发 #include <stdio.h> #include <stdlib.h> #include <string.h> #include <unistd.h> #include <net/if.h> #include <sys/ioctl.h> #include <sys/types.h> #in…

经管博士科研基础【12】包络定理

当我们知道一个函数的最优解时&#xff0c;我们要求解这一个函数的值函数关于函数中某一个参数的导数&#xff0c;那么就可以使用包络定理。 1. 无约束条件下的包络定理 函数在其极值点处对一个参数&#xff08;参数不是自变量&#xff09;取偏导数的结果&#xff0c;等价于这…

【List】List集合有序测试案例:ArrayList,LinkedList,Vector(123)

List是有序、可重复的容器。 有序&#xff1a; List中每个元素都有索引标记。可以根据元素的索引标记(在List中的位置)访问 元素&#xff0c;从而精确控制这些元素。 可重复&#xff1a; List允许加入重复的元素。更确切地讲&#xff0c;List通常允许满足 e1.equals(e2) 的元素…

Node.js crypto模块 加密算法

背景 微信小程序调用飞蛾热敏纸打印机&#xff0c;需要进行参数sig签名校验&#xff0c;使用的是sha1进行加密 // 通过crypto.createHash()函数&#xff0c;创建一个hash实例&#xff0c;但是需要调用md5&#xff0c;sha1&#xff0c;sha256&#xff0c;sha512算法来实现实例的…

小兔鲜商02

npm i vueuse/core -fvue插件使用&#xff1a; 许多公用的全局组件&#xff0c;&#xff0c;可以通过插件注册进去&#xff0c;就不用一个一个导入组件&#xff0c;&#xff0c; import XtxSkeleton from /components/library/xtx-skeletonexport default {install (app) {// …

C++并发编程:使用C++实现线程安全的栈

C并发编程&#xff1a;使用C实现线程安全的栈 引言 在多线程编程中&#xff0c;数据结构的线程安全性是至关重要的。本文将详细介绍如何使用C20标准库中的一些新特性来实现一个线程安全的栈。 什么是线程安全的栈&#xff1f; 简单来说&#xff0c;一个线程安全的栈是一个可…

linux操作系统中的动静态库(未完)

1. 静态库与动态库 静态库&#xff08;.a&#xff09;&#xff1a;程序在编译链接的时候把库的代码链接到可执行文件中。程序运行的时候将不再需要静态库动态库&#xff08;.so&#xff09;&#xff1a;程序在运行的时候才去链接动态库的代码&#xff0c;多个程序共享使用库的…

为什么Java接口可以多继承,而类不可以?

个人主页&#xff1a;金鳞踏雨 个人简介&#xff1a;大家好&#xff0c;我是金鳞&#xff0c;一个初出茅庐的Java小白 目前状况&#xff1a;22届普通本科毕业生&#xff0c;几经波折了&#xff0c;现在任职于一家国内大型知名日化公司&#xff0c;从事Java开发工作 我的博客&am…

unity 跨屏显示

1.代码 /*Type:设置分辨率*/ using System.Collections; using System.Collections.Generic; using UnityEngine; using System; using System.Runtime.InteropServices;public class ScreenManager : MonoBehaviour {[HideInInspector]//导入设置窗口函数 [DllImport("…

Redis之MoreKey问题及Scan命令解读

目录 MoreKey问题讨论 Scan命令 Sscan命令 Hscan命令 Zscan命令 MoreKey问题讨论 keys * 查看当前库所有key 对于海量数据执行key *会造成严重服务卡顿、影响业务。在实际环境中最好不要使用。生产制造过程中keys * / flushdb/flushall等危险命令以防止误删误用。 大量的…