SpringBoot整合WebSocket的两种方式及微服务网关Gateway配置
作者:mmseoamin日期:2023-12-13

一、说明

项目中后台微服务需要向前端页面推送消息,因此不可避免的需要用到WebSocket技术。SpringBoot已经为WebSocket的集成提供了很多支持,只是WebSocket消息如何通过微服务网关Spring Cloud Gateway向外暴露接口,实际开发过程中遇到了很多问题。微服务框架本身是作为一个平台为各种服务提供支撑的,所以对常用的两种WebSocket实现方式都要能够适配,特别是用Stomp方式实现时要考虑WebSocket接口与Rest API接口共存时的跨域问题。查了很多资料,也稍微浏览了一下源码,总算成功的解决了问题。下面着重讲实现的过程,展示代码,原理就不详细介绍了,网上一大堆。

二、WebSocket基本原理

WebSocket 协议是基于 TCP 的一种新的网络协议,它实现了浏览器与服务器全双工(full-duplex)通信—允许服务器主动发送信息给客户端,这样就可以实现从客户端发送消息到服务器,而服务器又可以转发消息到客户端,这样就能够实现客户端之间的交互。对于 WebSocket 的开发,Spring 也提供了良好的支持,目前很多浏览器已经实现了 WebSocket 协议,但是依旧存在着很多浏览器没有实现该协议,为了兼容那些没有实现该协议的浏览器,往往还需要通过 STOMP 协议来完成这些兼容。

SpringBoot整合WebSocket的两种方式及微服务网关Gateway配置,第1张

 三、常规实现方式

1、后台微服务代码

POM引入


	org.springframework.boot
	spring-boot-starter-websocket
	2.7.2


	org.springframework.boot
	spring-boot-starter-web
	2.7.2

WebSocket配置类

@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}

 WebSocket服务实现类

@ServerEndpoint("/websocket/{sid}")
@Component
@Slf4j
public class WebSocketServer {
    //静态变量,用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;
    //concurrent包的线程安全Set,用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet();
    //与某个客户端的连接会话,需要通过它来给客户端发送数据
    private Session session;
    //接收sid
    private String sid="";
/**
 * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session, @PathParam("sid") String sid) {
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        log.info("有新窗口开始监听:"+sid+",当前在线人数为" + getOnlineCount());
        this.sid=sid;
        try {
            sendMessage("连接成功");
        } catch (IOException e) {
            log.error("websocket IO异常");
        }
    }
/**
 * 连接关闭调用的方法
 */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        log.info("有一连接关闭!当前在线人数为" + getOnlineCount());
    }
/**
 * 收到客户端消息后调用的方法
 *
 * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("收到来自窗口"+sid+"的信息:"+message);
        //群发消息
        for (WebSocketServer item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
/**
 *
 * @param session
 * @param error
 */
    @OnError
    public void onError(Session session, Throwable error) {
        log.error("发生错误");
        error.printStackTrace();
    }
/**
 * 实现服务器主动推送
 */
    public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
    }
/**
 * 群发自定义消息
 * */
    public static void sendInfo(String message,@PathParam("sid") String sid) throws IOException {
        log.info("推送消息到窗口"+sid+",推送内容:"+message);
        for (WebSocketServer item : webSocketSet) {
            try {
                //这里可以设定只推送给这个sid的,为null则全部推送
                if(sid==null) {
                    item.sendMessage(message);
                }else if(item.sid.equals(sid)){
                    item.sendMessage(message);
                }
            } catch (IOException e) {
                continue;
            }
        }
    }
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }
    public static synchronized void subOnlineCount() {
        WebSocketServer.onlineCount--;
    }
    public static CopyOnWriteArraySet getWebSocketSet() {
        return webSocketSet;
    }
}

应用程序配置信息 application.yml

server:
  port: 9009
spring:
  application:
    name: test-service

2、Spring Cloud Gateway配置

这种方式的配置相对简单一些,只要把websocket的路径配置上就可以了。需要和REST API的路由信息分开进行配置。另外对于REST API有安全校验机制,在网关验证http请求携带的Token信息,对WebSocket连接就不检查了。

spring:
  cloud:
    gateway:
      globalcors:
        add-to-simple-url-handler-mapping: true
        corsConfigurations: #网关跨域配置
          '[/**]':
            allowedOriginPatterns: "*"
            allowedMethods: "*"
            allowedHeaders: "*"
            allowCredentials: true
            maxAge: 360000
      discovery:
        locator:
          lowerCaseServiceId: true
          enabled: true
      routes:
        - id: testWS2 #websocket 路由配置
          uri: lb:ws://test-service
          predicates:
            - Path=/test-ws/**
          filters:
            - StripPrefix=1
        - id: test2 #REST API路由配置
          uri: lb://test-service
          predicates:
            - Path=/test2/**
          filters:
            - StripPrefix=1
security:
  ignore:
    whites:
      - /test-ws/** #websocket不检查http头中的token

3、前端代码

写了个最简单的Html页面进行验证




    
    Title



hello world!

4、实现效果

打开网页 F12,可以看到效果。

SpringBoot整合WebSocket的两种方式及微服务网关Gateway配置,第2张

 

四、STOMP方式

STOMP即Simple (or Streaming) Text Orientated Messaging Protocol,简单(流)文本定向消息协议,它提供了一个可互操作的连接格式,允许STOMP客户端与任意STOMP消息代理(Broker)进行交互。STOMP协议由于设计简单,易于开发客户端,因此在多种语言和多种平台上得到广泛地应用。

为什么需要STOMP呢:常规的websocket连接和普通的TCP基本上没有什么差别的。所以STOMP在websocket上提供了一中基于帧线路格式(frame-based wire format)。简单一点来说,就是在websocket(TCP)上面加了一层协议,使双方遵循这种协议来发送消息。

1、后台微服务代码

POM引入,与上面相同


	org.springframework.boot
	spring-boot-starter-websocket
	2.7.2


	org.springframework.boot
	spring-boot-starter-web
	2.7.2

webSocket配置类

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketBrokerConfig implements WebSocketMessageBrokerConfigurer {
    //配置消息代理(Message Broker)
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 订阅
        registry.enableSimpleBroker("/toAll");
    }
    //注册STOMP协议的节点(endpoint)
    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 端点
        registry.addEndpoint("/device/point")
                //跨域,这儿必须加跨域,只在网关加不行
                .setAllowedOriginPatterns("*")
                //.setHandshakeHandler(customHandshakeHandler)
                .withSockJS(); // 使用sockJs
    }
    @Override
    public boolean configureMessageConverters(List messageConverters) { // 不要这个Bean好像也没问题
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
        MappingJackson2MessageConverter converter = new MappingJackson2MessageConverter();
        converter.setObjectMapper(objectMapper);
        DefaultContentTypeResolver resolver = new DefaultContentTypeResolver();
        resolver.setDefaultMimeType(MimeTypeUtils.APPLICATION_JSON);
        converter.setContentTypeResolver(resolver);
        messageConverters.add(new StringMessageConverter());
        messageConverters.add(new ByteArrayMessageConverter());
        messageConverters.add(converter);
        return false;
    }
}

Controller层代码

@Controller
public class ExampleWebSocketController {
    @MessageMapping("/welcome")
    @SendTo("/toAll/getResponse")
    public String sendTopicMessage(String str) {
        System.out.println("后台广播推送!");
        return str;
    }
}

2、Spring Cloud Gateway配置

这儿的配置是试了很多方法才确定了可行的配置的。对WebSocket,需要配置两个路由。一个是http方式,一个是ws方式。

网关本身加了跨域设置,后台服务的websocket也加了跨域配置,所以得在路由上加上DedupeResponseHeader的过滤器,只保留一个跨域请求头。这儿是关键。

spring:
  cloud:
    gateway:
      globalcors:
        add-to-simple-url-handler-mapping: true
        corsConfigurations: #网关跨域配置
          '[/**]':
            allowedOriginPatterns: "*"
            allowedMethods: "*"
            allowedHeaders: "*"
            allowCredentials: true
            maxAge: 360000
      discovery:
        locator:
          lowerCaseServiceId: true
          enabled: true
      routes:
        - id: model #REST API 路由配置
          uri: lb://model-service
          predicates:
            - Path=/model/**
          filters:
            - StripPrefix=1
        - id: websocket1  #websokcet 路由配置
          uri: lb://model-service
          predicates:
            - Path=/model-ws/device/point/info/**
          filters:
            - StripPrefix=1
            - DedupeResponseHeader=Access-Control-Allow-Credentials Access-Control-Allow-Origin
        - id: websocket1 #websokcet 路由配置
          uri: lb:ws://model-service
          predicates:
            - Path=/model-ws/device/point/**
          filters:
            - StripPrefix=1
            - DedupeResponseHeader=Access-Control-Allow-Credentials Access-Control-Allow-Origin
security:
  ignore:
    whites:
      - /model-ws/**

3、前端代码

再写一个简单的HTML页面展示效果




  
  Spring Boot+WebSoc+广播式


  
  
    
    
  
  
    
    
    

4、实现效果

打开网页,F12,可以看到效果

SpringBoot整合WebSocket的两种方式及微服务网关Gateway配置,第3张

SpringBoot整合WebSocket的两种方式及微服务网关Gateway配置,第4张