一、WebSocket相关介绍

在线测试websocket网站:http://www.easyswoole.com/wstool.html

1、WebSocket基本概念

1.1 基本概念

基本概念原理这里就不细讲了,一查一大把,推荐大佬的这篇博客(偏代码实践一些)以及知乎的这篇高赞回答(偏寓教于乐一些)。需要重点说明的几点

  • Websocket作为标准的通信协议在Web C/S使用时,需要浏览器和Web服务容器的支持
  • Websocket依靠Http来完成握手,握手完成后才是完全走Websocket协议
  • 资源描述符的前缀为ws,加密通道传输则为wss,例如ws://example.com:80/some/path
  • Websocket没有规定发送内容的格式,支持文本、二进制

1.2 Http握手

为什么要使用Http来进行握手而不是完全独立采用自有协议,主要原因有:

  • Websocket主要还是作为Http的一种补充,与Http紧密结合是合情合理的,并且能够较好地融入Http生态
  • 提供了良好的兼容性处理,可以通过Http来获取兼容性支持反馈以及使用Http来在不支持websocket的客户端上模拟兼容Websocket

1.3 SockJS

SockJS是一个JavaScript库,主要用于应对浏览器缺失websocket支持的情况。它提供了连贯的、跨浏览器的JavaScript API,它首先尝试使用原生Websocket,在失败时能够使用各种浏览器特定的传输协议来模拟Websocket的行为

1.4 Java规范

Java发布提供了Websocket的标准API接口JSR-356,作为Java EE7标准的一部分。大部分标准的Java web容器都已经实现了对Websocket的支持,同时也是兼容这个标准接口的,例如Tomcat 7.0.47+, Jetty 9.1+, GlassFish 4.1+, WebLogic 12.1.3+, Undertow 1.0+ (WildFly 8.0+)等。

1.5 Websocket使用场景

这里主要参考Spring文档中的叙述:Websocket虽然可以使网页变得动态以及更加有交互性,但是在很多情况下Ajax结合Http Streaming或者长轮询可以提供简单高效的解决方案。例如新闻、邮件、社交订阅等需要动态更新,但是在这些情景下每隔几分钟更新一次是完全没有问题的。而另一方面,协作、游戏以及金融应用则需要更加接近实时更新。注意延迟本身并不是决定性因素,如果信息量相对较少(例如监控网络故障),Http Streaming或轮询同样也可以高效地解决。低延迟、高频率以及高信息量的组合情况下,Websocket才是最佳选择。

2、STOMP协议

2.1 概念介绍

STOMP是一个简单的互操作协议,它被设计为常用消息传递模式的最小子集,定义了一种基于文本的简单异步消息协议,它最初是为脚本语言(如 Ruby、 Python 和 Perl)创建的,用于连接企业消息代理。STOMP已经广泛使用了好几年,并且得到了很多客户端(如stomp.js、Gozirra、stomp.py、stompngo等)、消息代理端(如ActiveMQ、RabbitMQ等)工具库的支持,目前最新的协议版本为1.2。

STOMP是一种基于’Frame’的协议,Frame基于Http建模,每个Frame由一个命令(Command)、一组头部(Headers)和可选的正文(Body)组成,如下是一个STOMP frame的基本结构示例:

COMMAND
header1:value1
header2:value2

Body^@

可以看到STOMP本身的结构是非常简单明了的。

2.2 STOMP相关命令

STOMP同样有客户端和服务端的概念,服务端被认为是可以接收和发送消息的一组目的地;而客户端则是用户代理,可以进行两种操作:发送消息(SEND)、发送订阅(SUBSCRIBE),为此,STOMP的命令有如下几种。

客户端命令:

  • CONNECT:用于初始化信息流或TCP连接,是客户端第一个需要发送的命令
  • SEND:表示向目的地发送消息,必须要包含一个名为destination的头部
  • SUBSCRIBE:用于注册监听一个目的地,必须包含一个名为destination的头部
  • BEGIN:用于启动事务,必须包含一个名为transaction的头部
  • COMMIT:用于提交事务,必须包含一个名为transaction的头部
  • ABORT:用于回滚事务,必须包含一个名为transaction的头部
  • DISCONNECT:告知服务端关闭连接

服务端命令:

  • CONNECTED:服务器响应客户的段的CONNECT请求,表示连接成功
  • MESSAGE:用于将订阅的消息发送给客户端,头部destination的值应与SEND frame中的相同,且必须包含一个名为message-id的头部用于唯一标识这个消息
  • RECIPT:收据,表示服务器成功处理了一个客户端要求返回收据的消息,必须包含头部message-id表明是哪个消息的收据
  • ERROR:出现异常时,服务端可能会发送该命令,通常在发送ERROR后将关闭连接

可以说STOMP主要就是提供了发送消息、订阅消息的语义,同时还能够支持事务的处理。

二、WebSocket的应用

官网关于websocket的介绍:https://docs.spring.io/spring-framework/docs/current/reference/html/web.html#websocket

1、WebSocket原生注解

原生websocket入门实践:https://blog.csdn.net/lemon_TT/article/details/113263443

pom.xml

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

配置类WebSocketConfig,这里开启了配置之后springboot才会去扫描对应的注解

@Configuration
@EnableWebSocket
public class WebSocketConfig {
//如果使用了springboot启动项目,则需要bean注入,而如果使用了外置tomcat容器,则并不要bean注入,否侧会报错
  @Bean
  public ServerEndpointExporter serverEndpoint() {
    return new ServerEndpointExporter();
  }
}

处理消息类WsServerEndpoint

@ServerEndpoint("/myWs")
@Component
public class WsServerEndpoint {
    /**
     * 连接成功
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        System.out.println("连接成功");
    }

    /**
     * 连接关闭
     * @param session
     */
    @OnClose
    public void onClose(Session session) {
        System.out.println("连接关闭");
    }

    /**
     * 接收到消息
     * @param text
     */
    @OnMessage
    public String onMsg(String text) throws IOException {
        return "servet 发送:" + text;
    }
}

这些注解都是属于jdk自带的,并不是spring提供的,具体位置是在javax.websocket下,需要注意的是接收参数中的session,这是我们需要保存的,后面如果要对客户端发送消息的话使用session.getBasicRemote().sendText(XXX)

@ServerEndpoint

  • 通过这个 spring boot 就可以知道你暴露出去的 ws 应用的路径,有点类似我们经常用的@RequestMapping。比如你的启动端口是8080,而这个注解的值是ws,那我们就可以通过 ws://127.0.0.1:8080/ws 来连接你的应用。

@OnOpen

  • 当 websocket 建立连接成功后会触发这个注解修饰的方法。

@OnClose

  • 当 websocket 建立的连接断开后会触发这个注解修饰的方法。

@OnMessage

  • 当客户端发送消息到服务端时,会触发这个注解修改的方法,它有一个 String 入参表明客户端传入的值。

@OnError

  • 当 websocket 建立连接时出现异常会触发这个注解修饰的方法。

2、Spring封装的WebSocket

spring同样也为我们提供了WebSocket的封装,这种方式可以自己配置拦截器,在tcp握手之前对请求进行一次处理,可以避免一些恶意的连接

2.1 pom.xml

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

2.2 WsSessionManager

这里简单通过 ConcurrentHashMap来实现了一个 session 池,用来保存已经登录的WebSocket 的 session。服务端发送消息给客户端必须要通过这个 session。

@Slf4j
public class WsSessionManager {
    /**
     * 保存连接 session 的地方
     */
    private static ConcurrentHashMap<String, WebSocketSession> SESSION_POOL = new ConcurrentHashMap<>();

    /**
     * 添加 session
     * @param key
     */
    public static void add(String key, WebSocketSession session) {
        // 添加 session
        SESSION_POOL.put(key, session);
    }

    /**
     * 删除 session,会返回删除的 session
     * @param key
     * @return
     */
    public static WebSocketSession remove(String key) {
        // 删除 session
        return SESSION_POOL.remove(key);
    }

    /**
     * 删除并同步关闭连接
     * @param key
     */
    public static void removeAndClose(String key) {
        WebSocketSession session = remove(key);
        if (session != null) {
            try {
                // 关闭连接
                session.close();
            } catch (IOException e) {
                // todo: 关闭出现异常处理
                e.printStackTrace();
            }
        }
    }
    /**
     * 获得 session
     * @param key
     * @return
     */
    public static WebSocketSession get(String key) {
        // 获得 session
        return SESSION_POOL.get(key);
    }
}

2.3 HttpAuthHandler

HttpAuthHandler用于处理ws的消息,通过继承 TextWebSocketHandler 类并覆盖相应方法,可以对 websocket 的事件进行处理,这里可以同原生注解的那几个注解连起来看(可以创建多个session池管理不同的websocket)

  • afterConnectionEstablished 方法是在 socket 连接成功后被触发,同原生注解里的 @OnOpen 功能
  • afterConnectionClosed方法是在 socket 连接关闭后被触发,同原生注解里的 @OnClose 功能
  • handleTextMessage方法是在客户端发送信息时触发,同原生注解里的 @OnMessage 功能
@Component
public class HttpAuthHandler extends TextWebSocketHandler {
    /**
     * socket 建立成功事件
     * @param session
     * @throws Exception
     */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        //这里的值在拦截器中的域属性中复制,后面会自动添加进去
        Object token = session.getAttributes().get("token");
        if (token != null) {
            // 用户连接成功,放入在线用户缓存
            WsSessionManager.add(token.toString(), session);
        } else {
            throw new RuntimeException("用户登录已经失效!");
        }
    }
    /**
     * 接收消息事件
     * @param session
     * @param message
     * @throws Exception
     */
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        // 获得客户端传来的消息
        String payload = message.getPayload();
        Object token = session.getAttributes().get("token");
        System.out.println("server 接收到 " + token + " 发送的 " + payload);
        session.sendMessage(new TextMessage("server 发送给 " + token + " 消息 " + payload + " " +    LocalDateTime.now().toString()));
    }

    /**
     * socket 断开连接时
     * @param session
     * @param status
     * @throws Exception
     */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        Object token = session.getAttributes().get("token");
        if (token != null) {
            // 用户退出,移除缓存
            WsSessionManager.remove(token.toString());
        }
    }
}

2.4 Interceptor拦截器

MyInterceptor用来拦截ws请求,通过实现 HandshakeInterceptor 接口来定义握手拦截器,注意这里与上面 Handler 的事件是不同的,这里是建立握手时的事件,分为握手前与握手后,而 Handler 的事件是在握手成功后的基础上建立 socket 的连接。所以在如果把认证放在这个步骤相对来说最节省服务器资源。它主要有两个方法 beforeHandshakeafterHandshake,顾名思义一个在握手前触发,一个在握手后触发。

@Component
public class MyInterceptor implements HandshakeInterceptor {
    /**
     * 握手前
     * @param request
     * @param response
     * @param wsHandler
     * @param attributes
     * @return
     * @throws Exception
     */
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        System.out.println("握手开始");
        // 获得请求参数,这里用了hutools工具箱
        HashMap<String, String> paramMap = (HashMap<String, String>) HttpUtil.decodeParamMap(request.getURI().getQuery(), Charset.defaultCharset());
        String uid = paramMap.get("token");
        if (StrUtil.isNotBlank(uid)) {
            // 放入属性域,可以在HttpAuthHandler的session的attributes中获取
            attributes.put("token", uid);
            System.out.println("用户 token " + uid + " 握手成功!");
            return true;
        }
        System.out.println("用户登录已失效");
        return false;
    }
    /**
     * 握手后
     * @param request
     * @param response
     * @param wsHandler
     * @param exception
     */
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) {
        System.out.println("握手完成");
    }
}

2.5 PrincipalHandshakeHandler配置(可不设置)

用户登录系统后,才可以登录websocket,并重写MyPrincipal,MyPrincipalHandshakeHandlerDefaultHandshakeHandler的子类,处理websocket请求,这里我们只重写determineUser方法,生成我们自己的Principal ,这里我们使用loginName标记登录用户,而不是默认值

MyPrincipal 定义自己的Principal

import java.security.Principal;

public class MyPrincipal implements Principal {
    private String loginName;

    public MyPrincipal(String loginName){
        this.loginName = loginName;
    }

    @Override
    public String getName() {
        return loginName;
    }
}

生成MyPrincipalHandshakeHandler类

@Component
public class MyPrincipalHandshakeHandler extends DefaultHandshakeHandler {
    private static final Logger log = LoggerFactory.getLogger(MyPrincipalHandshakeHandler.class);

    protected Principal determineUser(ServerHttpRequest request, WebSocketHandler wsHandler, Map<String, Object> attributes) {

        HttpSession httpSession = getSession(request);
        String user = (String)httpSession.getAttribute("loginName");

        if(StrUtil.isEmpty(user)){
            log.error("未登录系统,禁止登录websocket!");
            return null;
        }
        log.info(" MyDefaultHandshakeHandler login = " + user);
        return new MyPrincipal(user);
    }

    private HttpSession getSession(ServerHttpRequest request) {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest serverRequest = (ServletServerHttpRequest) request;
            //如果之前登录了必定有session,如果没有登录就返回null
            return serverRequest.getServletRequest().getSession(false);
        }
        return null;
    }
}

2.6 配置类WebSocketConfig

通过实现 WebSocketConfigurer 类并覆盖相应的方法进行 websocket 的配置。我们主要覆盖 registerWebSocketHandlers 这个方法。通过向 WebSocketHandlerRegistry 设置不同参数来进行配置。其中 addHandler方法添加我们的 ws 的 handler 处理类,第二个参数是你暴露出的 ws 路径。addInterceptors添加我们写的拦截器。setAllowedOrigins这个是关闭跨域校验。

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {

    @Autowired
    private HttpAuthHandler httpAuthHandler;
    @Autowired
    private MyInterceptor myInterceptor;
    @Autowired
    private MyPrincipalHandshakeHandler myPrincipalHandshakeHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry
                //自定义的websocket服务,这里都可以定义多个
                .addHandler(httpAuthHandler, "ws")
                //设置拦截器
                .addInterceptors(myInterceptor)
                //设置登录用户检查
                //.setHandshakeHandler(myPrincipalHandshakeHandler)
                //关闭跨域校验
                .setAllowedOrigins("*");
    }
}

最后访问链接ws://localhost:8085/parentServer?token=shawn

3、STOMP

3.1 介绍

stomp是WebSocket的一个子协议,SpringBoot官方也有整合stomp的例子,这也是我现在所用到的整合方式,这种方式功能更加强大,可以使用消息代理,对于发送的消息可以使用类似springMvc的处理方式,同时消息的发送变成了订阅的模式,可以很方便的进行群发。

3.2 pom.xml

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

3.3 WebSocketConfig配置类

@Configuration
@EnableWebSocketMessageBroker
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Bean
    public WebSocketInterceptor getWebSocketInterceptor() {
        return new WebSocketInterceptor();
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        // 配置客户端尝试连接地址
        registry.
                addEndpoint("/ws").     // 设置连接节点,前端请求的建立连接的地址就是 http://ip:端口/ws
                //addInterceptors(getWebSocketInterceptor()).     // 设置握手拦截器
                setAllowedOrigins("*").     // 配置跨域
                withSockJS();       // 开启sockJS支持,这里可以对不支持stomp的浏览器进行兼容。
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        // 消息代理,这里配置自带的消息代理,也可以配置其它的消息代理
        // 一定要注意这里的参数,可以理解为开启通道,后面如果使用了"/XXX"来作为前缀,这里就要配置,同时这里的"/topic"是默认群发消息的前缀,前端在订阅群发地址的时候需要加上"/topic"
        registry.enableSimpleBroker("/user","/topic");  
        // 客户端向服务端发送消息需有的前缀,需要什么样的前缀在这里配置,但是不建议使用,这里跟下面首次订阅返回会有点冲突,如果不需要首次订阅返回消息,也可以加上消息前缀
        // registry.setApplicationDestinationPrefixes("/");
        // 配置单发消息的前缀 /user,前端订阅一对一通信的地址需要加上"/user"前缀
        registry.setUserDestinationPrefix("/user");
    }
}

3.4 SpringMVC控制器

WSController是ws的控制器,@SubscribeMapping注解可以在客户端首次订阅了对应的地址后直接返回一条消息,订阅地址支持路径参数,接收路径参数需要在参数前加上@DestinationVariable,下面有三种常用的订阅方式,这里一定要注意地址格式,通用群发消息/topic/hello,指定一部分人可以收到的群发消息/topic/state/{classId},一对一消息/user/{name}/hello,我这里的ResultWrapper.success就是一个封装类,跟springMVC中封装的返回对象完全一致,stomp会把对象解析为json字符串返回给前端。

@MessageMapping是用来接收客户端对某个地址发送的消息,需要注意的是客户端发送的地址,如果在之前的配置类中配置了发送前缀,则必须携带前缀才能发送消息到客户端,如:/app/hello,但是服务器仍然只需要这样写@MessageMapping("/hello")

@SendTo是用来向客户端发送消息的注解,这里填写的参数就是订阅地址的全名/topic/hello不能省略/topic,返回消息只需要return消息对象即可。

除了注解的方式发送消息,还有一种灵活的方式使用消息模板来发送,simpMessagingTemplate.convertAndSendToUser(一对一)和simpMessagingTemplate.convertAndSend(群发),注意我参数中填写的方式,这种方式比较推荐使用,可以在任意地方对客户端发送消息,但是这个地方似乎有个坑,发送消息之后会阻塞在这里,不过可以开一个线程去发送消息。

@RestController
public class WSController {
    @Autowired
    private SimpMessagingTemplate simpMessagingTemplate;

    @SubscribeMapping({"/topic/hello"})
    public Result subscribeTime() {
        return ResultWrapper.success("hello!");
    }
    
    @SubscribeMapping({"/topic/info/{classId}"})
    public Result subscribeState(@DestinationVariable String classId) {
        return ResultWrapper.success("班级消息推送订阅成功!");
    }

    @SubscribeMapping({"/user/{name}/hello"})
    public Result subscribeParam(@DestinationVariable String name) {
        return ResultWrapper.success("你好!"+name);
    }
    
    @MessageMapping("/hello")
    @SendTo("/topic/hello")
    public Result hello(RequestMessage requestMessage) {
        System.out.println("接收消息:" + requestMessage);
        return ResultWrapper.success("服务端接收到你发的:"+requestMessage);
    }

    @GetMapping("/sendMsgToUser")
    public String sendMsgByUser(String name, String msg) {
        // /user/{name}/hello
        simpMessagingTemplate.convertAndSendToUser(name, "/hello", msg);
        return "success";
    }

    @GetMapping("/sendMsgToAll")
    public String sendMsgByAll(int classId, String msg) {
        // /topic/info/{classId}
        simpMessagingTemplate.convertAndSend("/topic/info/"+classId, msg);
        return "success";
    }
}

3.5 前端的订阅

一部分代码举例

function connect() {
        var socket = new SockJS('http://localhost:8092/simple');
        stompClient = Stomp.over(socket);
        stompClient.connect({}, function (frame) {
            setConnected(true);
            console.log('Connected:' + frame);
            stompClient.subscribe('/topic/say', function (response) {
                showResponse(JSON.parse(response.body).responseMessage);
            });
            // 另外再注册一下定时任务接受
            stompClient.subscribe('/topic/callback', function (response) {
                showCallback(response.body);
            });
        });
    }

    function disconnect() {
        if (stompClient != null) {
            stompClient.disconnect();
        }
        setConnected(false);
        console.log('Disconnected');
    }

三、常见问题

1、Session共享问题

上面反复提到一个问题就是,服务端如果要主动发送消息给客户端一定要用到 session。而大家都知道的是 session 这个东西是不跨 jvm 的。如果有多台服务器,在 http 请求的情况下,我们可以通过把 session 放入缓存中间件中来共享解决这个问题,通过 spring session 几条配置就解决了。但是 web socket 不可以。他的 session 是不能序列化的,当然这样设计的目的不是为了为难你,而是出于对 http 与 web socket 请求的差异导致的。

目前网上找到的最简单方案就是通过 redis 订阅广播的形式,主要代码跟第二种方式差不多,你要在本地放个 map 保存请求的 session。也就是说每台服务器都会保存与他连接的 session 于本地。然后发消息的地方要修改,并不是现在这样直接发送,而通过 redis 的订阅机制。服务器要发消息的时候,你通过 redis 广播这条消息,所有订阅的服务端都会收到这个消息,然后本地尝试发送。最后肯定只有有这个对应用户 session 的那台才能发送出去。


参考文章

【websocket】spring boot 集成 websocket 的四种方式

STMOP集成WebSocket实时通信

Spring WebSocket简析

SpringBoot项目整合WebSocket

Logo

为开发者提供学习成长、分享交流、生态实践、资源工具等服务,帮助开发者快速成长。

更多推荐