• 解决websocket不定时出现1005错误

    后台抛出异常如下：

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

    分析原因是：

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

    解决方案：

    在gateway进行请求拦截

    代码如下：

    1. @Slf4j
    2. @Component
    3. public class CustomWebsocketRoutingFilter implements GlobalFilter, Ordered {
    4.  
    5.     //Sec-Websocket protocol
    6.     public static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol";
    7.     //Sec-Websocket header
    8.     public static final String SEC_WEBSOCKET_HEADER = "sec-websocket";
    9.     //http header schema
    10.     public static final String HEADER_UPGRADE_WebSocket = "websocket";
    11.     public static final String HEADER_UPGRADE_HTTP = "http";
    12.     public static final String HEADER_UPGRADE_HTTPS = "https";
    13.     private final WebSocketClient webSocketClient;
    14.     private final WebSocketService webSocketService;
    15.     private final ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider;
    16.     // 不直接使用 headersFilters 用该变量代替
    17.     private volatile List<HttpHeadersFilter> headersFilters;
    18.  
    19.     public CustomWebsocketRoutingFilter(WebSocketClient webSocketClient, WebSocketService webSocketService, ObjectProvider<List<HttpHeadersFilter>> headersFiltersProvider) {
    20.         this.webSocketClient = webSocketClient;
    21.         this.webSocketService = webSocketService;
    22.         this.headersFiltersProvider = headersFiltersProvider;
    23.     }
    24.  
    25.     /* for testing */
    26.     //http请求转为ws请求
    27.     static String convertHttpToWs(String scheme) {
    28.         scheme = scheme.toLowerCase();
    29.         return "http".equals(scheme) ? "ws" : "https".equals(scheme) ? "wss" : scheme;
    30.     }
    31.  
    32.     @Override
    33.     public int getOrder() {
    34.         // Before NettyRoutingFilter since this routes certain http requests
    35.         //修改了这里 之前是-1 降低优先级
    36.         return Ordered.LOWEST_PRECEDENCE - 2;
    37.     }
    38.  
    39.     @Override
    40.     public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    41.         changeSchemeIfIsWebSocketUpgrade(exchange);
    42.  
    43.         URI requestUrl = exchange.getRequiredAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
    44.         String scheme = requestUrl.getScheme();
    45.  
    46.         if (ServerWebExchangeUtils.isAlreadyRouted(exchange) || (!"ws".equals(scheme) && !"wss".equals(scheme))) {
    47.             return chain.filter(exchange);
    48.         }
    49.         ServerWebExchangeUtils.setAlreadyRouted(exchange);
    50.  
    51.         HttpHeaders headers = exchange.getRequest().getHeaders();
    52.         HttpHeaders filtered = HttpHeadersFilter.filterRequest(getHeadersFilters(), exchange);
    53.  
    54.         List<String> protocols = getProtocols(headers);
    55.  
    56.         return this.webSocketService.handleRequest(exchange, new ProxyWebSocketHandler(requestUrl, this.webSocketClient, filtered, protocols));
    57.     }
    58.  
    59.     /* for testing */
    60.     //获取请求头里的协议信息
    61.     List<String> getProtocols(HttpHeaders headers) {
    62.         List<String> protocols = headers.get(SEC_WEBSOCKET_PROTOCOL);
    63.         if (protocols != null) {
    64.             ArrayList<String> updatedProtocols = new ArrayList<>();
    65.             for (int i = 0; i < protocols.size(); i++) {
    66.                 String protocol = protocols.get(i);
    67.                 updatedProtocols.addAll(Arrays.asList(StringUtils.tokenizeToStringArray(protocol, ",")));
    68.             }
    69.             protocols = updatedProtocols;
    70.         }
    71.         return protocols;
    72.     }
    73.  
    74.     /* for testing */
    75.     List<HttpHeadersFilter> getHeadersFilters() {
    76.         if (this.headersFilters == null) {
    77.             this.headersFilters = this.headersFiltersProvider.getIfAvailable(ArrayList::new);
    78.  
    79.             // remove host header unless specifically asked not to
    80.             this.headersFilters.add((headers, exchange) -> {
    81.                 HttpHeaders filtered = new HttpHeaders();
    82.                 filtered.addAll(headers);
    83.                 filtered.remove(HttpHeaders.HOST);
    84.                 boolean preserveHost = exchange.getAttributeOrDefault(ServerWebExchangeUtils.PRESERVE_HOST_HEADER_ATTRIBUTE, false);
    85.                 if (preserveHost) {
    86.                     String host = exchange.getRequest().getHeaders().getFirst(HttpHeaders.HOST);
    87.                     filtered.add(HttpHeaders.HOST, host);
    88.                 }
    89.                 return filtered;
    90.             });
    91.  
    92.             this.headersFilters.add((headers, exchange) -> {
    93.                 HttpHeaders filtered = new HttpHeaders();
    94.                 for (Map.Entry<String, List<String>> entry : headers.entrySet()) {
    95.                     if (!entry.getKey().toLowerCase().startsWith(SEC_WEBSOCKET_HEADER)) {
    96.                         filtered.addAll(entry.getKey(), entry.getValue());
    97.                     }
    98.                 }
    99.                 return filtered;
    100.             });
    101.         }
    102.  
    103.         return this.headersFilters;
    104.     }
    105.  
    106.     static void changeSchemeIfIsWebSocketUpgrade(ServerWebExchange exchange) {
    107.         // 检查版本是否适合
    108.         URI requestUrl = exchange.getRequiredAttribute(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR);
    109.         String scheme = requestUrl.getScheme().toLowerCase();
    110.         String upgrade = exchange.getRequest().getHeaders().getUpgrade();
    111.         // change the scheme if the socket client send a "http" or "https"
    112.         if (HEADER_UPGRADE_WebSocket.equalsIgnoreCase(upgrade) && (HEADER_UPGRADE_HTTP.equals(scheme) || HEADER_UPGRADE_HTTPS.equals(scheme))) {
    113.             String wsScheme = convertHttpToWs(scheme);
    114.             boolean encoded = ServerWebExchangeUtils.containsEncodedParts(requestUrl);
    115.             URI wsRequestUrl = UriComponentsBuilder.fromUri(requestUrl).scheme(wsScheme).build(encoded).toUri();
    116.             exchange.getAttributes().put(ServerWebExchangeUtils.GATEWAY_REQUEST_URL_ATTR, wsRequestUrl);
    117.             if (log.isTraceEnabled()) {
    118.                 log.trace("changeSchemeTo:[" + wsRequestUrl + "]");
    119.             }
    120.         }
    121.     }
    122.  
    123.     //自定义websocket处理方式
    124.     private static class ProxyWebSocketHandler implements WebSocketHandler {
    125.  
    126.         private final WebSocketClient client;
    127.         private final URI url;
    128.         private final HttpHeaders headers;
    129.         private final List<String> subProtocols;
    130.  
    131.         ProxyWebSocketHandler(URI url, WebSocketClient client, HttpHeaders headers, List<String> protocols) {
    132.             this.client = client;
    133.             this.url = url;
    134.             this.headers = headers;
    135.             if (protocols != null) {
    136.                 this.subProtocols = protocols;
    137.             } else {
    138.                 this.subProtocols = Collections.emptyList();
    139.             }
    140.         }
    141.  
    142.         @Override
    143.         public List<String> getSubProtocols() {
    144.             return this.subProtocols;
    145.         }
    146.  
    147.         @Override
    148.         public Mono<Void> handle(WebSocketSession session) {
    149.             return this.client.execute(this.url, this.headers, new WebSocketHandler() {
    150.                 private CloseStatus adaptCloseStatus(CloseStatus closeStatus) {
    151.                     int code = closeStatus.getCode();
    152.                     if (code > 2999 && code < 5000) {
    153.                         return closeStatus;
    154.                     }
    155.                     switch (code) {
    156.                         case 1000:
    157.                             //正常关闭
    158.                             return closeStatus;
    159.                         case 1001:
    160.                             //服务器挂了或者页面跳转
    161.                             return closeStatus;
    162.                         case 1002:
    163.                             //协议错误
    164.                             return closeStatus;
    165.                         case 1003:
    166.                             //收到了不能处理的数据类型
    167.                             return closeStatus;
    168.                         case 1004:
    169.                             // 预留关闭状态码
    170.                             return CloseStatus.PROTOCOL_ERROR;
    171.                         case 1005:
    172.                             // 预留关闭状态码 期望收到状态码但是没有收到
    173.                             return CloseStatus.PROTOCOL_ERROR;
    174.                         case 1006:
    175.                             // 预留关闭状态码 连接异常关闭
    176.                             return CloseStatus.PROTOCOL_ERROR;
    177.                         case 1007:
    178.                             //收到的数据与实际的消息类型不匹配
    179.                             return closeStatus;
    180.                         case 1008:
    181.                             //收到不符合规则的消息
    182.                             return closeStatus;
    183.                         case 1009:
    184.                             //收到太大的不能处理的消息
    185.                             return closeStatus;
    186.                         case 1010:
    187.                             //client希望server提供多个扩展，server没有返回相应的扩展信息
    188.                             return closeStatus;
    189.                         case 1011:
    190.                             //server遇到不能完成的请求
    191.                             return closeStatus;
    192.                         case 1012:
    193.                             // Not in RFC6455
    194.                             // return CloseStatus.SERVICE_RESTARTED;
    195.                             return CloseStatus.PROTOCOL_ERROR;
    196.                         case 1013:
    197.                             // Not in RFC6455
    198.                             // return CloseStatus.SERVICE_OVERLOAD;
    199.                             return CloseStatus.PROTOCOL_ERROR;
    200.                         case 1015:
    201.                             // 不能进行TLS握手 如：server证书不能验证
    202.                             return CloseStatus.PROTOCOL_ERROR;
    203.                         default:
    204.                             return CloseStatus.PROTOCOL_ERROR;
    205.                     }
    206.                 }
    207.  
    208.                 /**
    209.                  * send      发送传出消息
    210.                  * receive   处理入站消息流
    211.                  * doOnNext  对每条消息做什么
    212.                  * zip       加入流
    213.                  * then      返回接收完成时完成的Mono<Void>
    214.                  */
    215.                 @Override
    216.                 public Mono<Void> handle(WebSocketSession proxySession) {
    217.                     Mono<Void> serverClose = proxySession.closeStatus().filter(__ -> session.isOpen())
    218.                             .map(this::adaptCloseStatus)
    219.                             .flatMap(session::close);
    220.                     Mono<Void> proxyClose = session.closeStatus().filter(__ -> proxySession.isOpen())
    221.                             .map(this::adaptCloseStatus)
    222.                             .flatMap(proxySession::close);
    223.                     // Use retain() for Reactor Netty
    224.                     Mono<Void> proxySessionSend = proxySession
    225.                             .send(session.receive().doOnNext(WebSocketMessage::retain));
    226.                     Mono<Void> serverSessionSend = session
    227.                             .send(proxySession.receive().doOnNext(WebSocketMessage::retain));
    228.                     // Ensure closeStatus from one propagates to the other
    229.                     Mono.when(serverClose, proxyClose).subscribe();
    230.                     // Complete when both sessions are done
    231.                     return Mono.zip(proxySessionSend, serverSessionSend).then();
    232.                 }
    233.                 @Override
    234.                 public List<String> getSubProtocols() {
    235.                     return CustomWebsocketRoutingFilter.ProxyWebSocketHandler.this.subProtocols;
    236.                 }
    237.             });
    238.         }
    239.     }
    240.  
    241.  
    242. }

  • 相关阅读:
    Java中的dozer对象转换
    JDBC笔记
    Altassian | Bitbucket Python API使用及相关任务自动化的实现
    【C++】class的设计与使用（十）重载iostream运算符
    sql根据出生日期计算当前年龄 函数TIMESTAMPDIFF()
    广度优先搜索(Breadth First Search, BFS)算法
    编译nw-node版本的插件
    故障分析 | OceanBase Proxy 无法连接 OBserver 集群
    需求分析岗的一般工作流程
    sfm算法之三角化(三角测量)
  • 原文地址：https://blog.csdn.net/vigor512/article/details/132620369