Spring Boot整合WebSocket（stomp协议）

# 前言

场景：通过 websocket 实现实时返回用户的通知消息
难点：
- 如果发送给指定用户？
- 集群下如何发送？

# 依赖

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

# 配置

@Configuration(proxyBeanMethods = false)
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // ws://xxxxx/ws 连接
        registry
                .addEndpoint("/ws")
                .setAllowedOriginPatterns("*");
        
        // 启用WebSocket端点 前端即可通过 http://xxxx/ws与服务器建立通信
        registry
                .addEndpoint("/ws") 
                // .setAllowedOrigins("*") // 很多教程都说使用这个解决跨域问题，实际上要用下面的setAllowedOriginPatterns
                .setAllowedOriginPatterns("*") // 解决跨域问题
                .withSockJS(); // 启用SockJS支持
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 请求路径(destination)前缀
        registry.setApplicationDestinationPrefixes("/app");
        // 订阅路径前缀，即所有订阅路径都要以/topic作为前缀
        registry.enableSimpleBroker("/topic");
        // 发送给指定用户的前缀
        registry.setUserDestinationPrefix("/user");
        
    }
    
    /**
     * 配置客户端入站通道拦截器
     *
     * @param registration
     */
    @Override
    public void configureClientInboundChannel(ChannelRegistration registration) {
        registration.interceptors(new MyChannelInterceptor());
    }

}

# 拦截器

public class MyChannelInterceptor implements ChannelInterceptor {

    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);

        // 判断是否是连接，如果是，需要获取token，并且设置用户对象
        if (Objects.nonNull(accessor) && StompCommand.CONNECT.equals(accessor.getCommand())) {
            List<String> nativeHeader = accessor.getNativeHeader("Blade-Auth");
            String auth = nativeHeader.get(0);
            if (StrUtil.isNotBlank(auth)) {
                String token = JwtUtil.getToken(auth);
                Claims claims = AuthUtil.parseJWT(token);
                if (Objects.nonNull(claims)) {
                    String userId = (String) claims.get("user_id");
                    if (StringUtils.hasText(userId)) {
                        // 设置当前访问器的认证用户
                        accessor.setUser(() -> userId);
                    }
                    return message;
                }
            }
            return null;
        }
        return message;
    }
}

# 控制器

@Controller
public class TestController {

    @MessageMapping("/chat.sendMessage")
    @SendTo("/topic/public")
    public ChatMessage sendMessage(@Payload ChatMessage chatMessage) {
        return chatMessage;
    }

    @MessageMapping("/chat.addUser")
    @SendTo("/topic/public")
    public ChatMessage addUser(@Payload ChatMessage chatMessage,
                               SimpMessageHeaderAccessor headerAccessor) {
        // Add username in web socket session
        headerAccessor.getSessionAttributes().put("username", chatMessage.getSender());
        return chatMessage;
    }
    
    /**
     * 点对点发送消息，将消息发送到指定用户
     */
    @MessageMapping("/test")
    public String sendUserMessage(Principal principal, @Payload ChatMessage chatMessage) {
        // 订阅路径：/user/topic/public
        simpMessageSendingOperations.convertAndSendToUser(chatMessage.getSender(), "/topic/public",
                String.format("%s 发送给 %s", principal.getName(), chatMessage.getSender()));
        // 订阅路径：/topic/public
        simpMessageSendingOperations.convertAndSend("/topic/public",
                String.format("%s 发送给222 %s", principal.getName(), chatMessage.getSender()));
        return "success";
    }


}

@Data
public class ChatMessage {

	public enum MessageType {
		/**
		 * 聊天
		 */
		CHAT,
		/**
		 * 加入
		 */
		JOIN,
		/**
		 * 离开
		 */
		LEAVE
	}

	private MessageType type;
	private String content;
	private String sender;

}

# 调试

前端调试工具：https://gitee.com/Samler/sockjs-client-tool

在线地址：http://ws-tool.samler.cn/

注意

~~不建议使用 apifox 的 WebSocket 接口调试功能，至少截止软件版本 2.2.33 版的 beta 版功能会提示 无法连接到 http://localhost:8207/ws~~

实际上是没有开启 ws 连接

# 问题与进阶

前后端分离项目提示跨域

配置 setAllowedOriginPatterns("*")

经过网关调用报 401 未授权

将 /ws/** 设置白名单

如何鉴权

通过请求头的 token，参考拦截器代码

经过 Gateway 转发，返回两个 Access-Control-Allow-Origin

在网关配置

spring:
  cloud:
    gateway:
      default-filters:
        - DedupeResponseHeader=Access-Control-Allow-Origin Access-Control-Allow-Credentials, RETAIN_LAST

参考：Spring Cloud Gateway 2.X 跨域时出现重复 Origin 的 BUG - 明天，你好啊 - 博客园 (opens new window)

# 参考资料

上次更新: 2023/06/09, 22:19:39

← XXL-JOB快速上手 SpringBoot整合i18n（多语言）→