微服務學習----Spring Cloud 使用 webSocket

Spring Cloud 使用 webSocket

網關配置

spring cloud 的網關組件有zuul和getway

getway

base:
  config:
    nacos:
      nacoshost: localhost
      port: 8848

spring:
  application:
    name: gateway
  main:
    allow-bean-definition-overriding: true
  cloud:
    nacos:
      discovery:
        server-addr: ${base.config.nacos.nacoshost}:${base.config.nacos.port}
    gateway:
      discovery:
        locator:
          enabled: true
      routes:
        #  websocket
        - id: CLOUD-WEBSOCKET
          uri: lb:ws://cloud-websocket
          predicates:
            - Path=/cloud-websocket/**
server:
  port: 8888

配置網關的時候注意添加ws協議。

zuul

zuul只能管理http請求，不推薦使用zuul管理websocket連接，推薦直連。

服務端

• 添加maven依賴

  • <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-websocket</artifactId>
            </dependency>
    

• 添加webSocket 配置

  • @Configuration
    @EnableWebSocket
    public class WebsocketConfiguration implements WebSocketConfigurer {
        @Override
        public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
            // webSocket通道
            // 指定處理器和路徑
            registry.addHandler(new WebSocketHandler(), "/websocket")
                    // 指定自定義攔截器
                    .addInterceptors(new WebSocketInterceptor())
                    // 允許跨域
                    .setAllowedOrigins("*");
            // sockJs通道
            registry.addHandler(new WebSocketHandler(), "/sock-js")
                    .addInterceptors(new WebSocketInterceptor())
                    .setAllowedOrigins("*")
                    // 開啟sockJs支持
                    .withSockJS();
        }
    }
    

• 添加處理器

  • package com.auexpress.cloud.handler;
    
    import com.alibaba.fastjson.JSONObject;
    import org.apache.commons.lang3.StringUtils;
    import org.springframework.web.socket.*;
    import org.springframework.web.socket.handler.AbstractWebSocketHandler;
    
    import java.io.IOException;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /**
     * @Description
     * @ClassName WebSocketHandler
     * @Author HYSong
     * @date 2020.04.14 10:08
     */
    public class WebSocketHandler extends AbstractWebSocketHandler {
        /**
         *  存儲sessionId和webSocketSession
         *  需要注意的是，webSocketSession沒有提供無參構造，不能進行序列化，也就不能通過redis存儲
         *  在分布式系統中，要想別的辦法實現webSocketSession共享
         */
        private static Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();
        private static Map<String, String> userMap = new ConcurrentHashMap<>();
    
        /**
         * webSocket連接創建后調用
         */
        @Override
        public void afterConnectionEstablished(WebSocketSession session) {
            // 獲取參數
            String user = String.valueOf(session.getAttributes().get("user"));
            userMap.put(user, session.getId());
            sessionMap.put(session.getId(), session);
        }
    
        /**
         * 接收到消息會調用
         */
        @Override
        public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
            JSONObject jsonObject = JSONObject.parseObject(message.getPayload().toString());
            String content = jsonObject.getString("content");
            String targetAdminId = jsonObject.getString("targetId");
            if("0".equals(targetAdminId)){
                //  推送給所有人
                userMap.forEach((key,value)->{
                    try {
                        this.sendMessage(key,content);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                });
            }else{
                sendMessage("1", content);
            }
        }
    
        /**
         * 連接出錯會調用
         */
        @Override
        public void handleTransportError(WebSocketSession session, Throwable exception) {
            sessionMap.remove(session.getId());
        }
    
        /**
         * 連接關閉會調用
         */
        @Override
        public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
            sessionMap.remove(session.getId());
        }
    
        @Override
        public boolean supportsPartialMessages() {
            return false;
        }
    
        /**
         * 后端發送消息
         */
        public void sendMessage(String user, String message) throws IOException {
            String sessionId = userMap.get(user);
            if (StringUtils.isEmpty(sessionId)) {
                return;
            }
            WebSocketSession session = sessionMap.get(sessionId);
            if (session == null) {
                return;
            }
            session.sendMessage(new TextMessage(message));
        }
    }
    

• 添加攔截器

  • package com.auexpress.cloud.interceptor;
    
    import org.springframework.http.server.ServerHttpRequest;
    import org.springframework.http.server.ServerHttpResponse;
    import org.springframework.http.server.ServletServerHttpRequest;
    import org.springframework.web.socket.WebSocketHandler;
    import org.springframework.web.socket.server.HandshakeInterceptor;
    
    import java.util.Map;
    
    /**
     * @Description
     * @ClassName WebSocketInterceptor
     * @Author HYSong
     * @date 2020.04.14 10:09
     */
    public class WebSocketInterceptor implements HandshakeInterceptor {
        /**
         * handler處理前調用,attributes屬性最終在WebSocketSession里,
         * 可能通過webSocketSession.getAttributes().get(key值)獲得
         */
        @Override
        public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) {
            if (request instanceof ServletServerHttpRequest) {
                ServletServerHttpRequest serverHttpRequest = (ServletServerHttpRequest) request;
                // 獲取請求路徑攜帶的參數
                String user = serverHttpRequest.getServletRequest().getParameter("user");
                attributes.put("user", user);
                return true;
            } else {
                return false;
            }
        }
    
        @Override
        public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
    
        }
    }