java websocket教程

定义

websocket是什么

WebSocket是一种在单个TCP连接上进行全双工通讯的协议.简单来说就是客户端与服务端建立起长连接可以相互发送消息.

websocket使用场景

主要用在对消息实时性比较高的场景.用来替代轮询方案

• 实时在线聊天
• 浏览器之间的协同编辑工作
• 多人在线游戏

浏览器支持websocket的版本

WebSocket通信协议于2011年被修订为RFC 6455的标准.所以对浏览器、后端服务器是有要求的.以下是被支持的版本

image.png

tomcat支持websocket的版本

http://tomcat.apache.org/（7.0.27支持websocket，建议用tomcat8，7.0.27中的接口已经过时）

浏览器与服务器之间连接如何建立(通信协议)

Websocket 通过HTTP/1.1 协议的101状态码进行握手,升级成websocket连接

• 请求

# Websocket使用ws或wss统一资源标志符(必填)
GET ws://localhost:8090/ws/stomp/561/abkkwlke/websocket HTTP/1.1
# 升级成websocket协议(必填)
Upgrade: websocket
# Connection必须设置Upgrade，表示客户端希望连接升级(必填)
Connection: Upgrade
# Origin字段是可选的，通常用来表示在浏览器中发起此Websocket连接所在的页面
Origin: http://example.com
# Sec-WebSocket-Key 服务端会用来验证该请求是否是websocket请求,尽量避免与http请求被误认为websocket(必填)
Sec-WebSocket-Key: sN9cRrP/n9NdMgdcy2VJFQ==
# Websocket支持的版本(必填)
Sec-WebSocket-Version: 13

• 响应

# 响应的状态码,必须是101
HTTP/1.1 101 // 
# 升级的协议
Upgrade: websocket
# 表示客户端希望连接升级
Connection: upgrade
# 服务端根据Sec-WebSocket-Key生成,用来验证该请求是websocket请求
Sec-WebSocket-Accept: V395OugSb9uYXr6dA44VGcn/oAM=

浏览器与服务器之间数据如何传输(数据协议)

STOMP 是基于 WebSocket的上层协议,提供了一个基于帧的线路格式层，用来定义消息语义.提供了一套完整websocket数据传输的api.让前后端能够快速变现.

• 消息发送的格式

# stomp命令
SEND
# 服务端接口
destination:/ws/broadcast
content-length:87

# 内容 可以是json格式
{"destination":"/topic","payload":"1231231","onErrorDestination":"/topic"}

• 支持的命令
  • SEND
  • SUBSCRIBE
  • UNSUBSCRIBE
  • BEGIN
  • COMMIT
  • ABORT
  • ACK
  • NACK
  • DISCONNECT

浏览器与服务器之间如何实现消息的广播、点对点传输

主要通过发布/订阅的模式来实现

• 广播思路

  1. 浏览器订阅主题: /topic
  2. 服务器发送消息到主题/topic
  3. 所有订阅的浏览器都能收到消息

• 点对点的思路(浏览器A->B)

  1. 浏览器B订阅主题: /user/B/topic
  2. 浏览器A发送消息到主题: /user/B/topic
  3. 浏览器B就能收到消息

如何使用

后端使用

spring boot整合websocket

• pom

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

• stomp的配置

@Configuration
@ComponentScan("com.websocket.test")
@EnableConfigurationProperties(value = {WebSocketProperties.class})
@EnableWebSocketMessageBroker
public class WebSocketConfigurer extends AbstractWebSocketMessageBrokerConfigurer {

    @Autowired
    private WebSocketProperties webSocketProperties;

    @Override
    public void registerStompEndpoints(StompEndpointRegistry stompEndpointRegistry) {
        // 注册一个Stomp的节点（endpoint）,并指定使用SockJS协议。
        stompEndpointRegistry
                .addEndpoint(webSocketProperties.getEndPoint())
                .setAllowedOrigins(webSocketProperties.getAllowedOrigins())
                .withSockJS();
    }

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {

        // 定义心跳线程
        ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
        taskScheduler.setThreadNamePrefix("wss-heartbeat-thread-");
        taskScheduler.setDaemon(true);
        taskScheduler.initialize();

        // 服务端发送消息给客户端的域,多个用逗号隔开
        registry.enableSimpleBroker(webSocketProperties.getEnableSimpleBroker())
                // 定义心跳间隔 单位(ms)
                .setHeartbeatValue(new long[]{webSocketProperties.getHeartBeatInterval(), webSocketProperties.getHeartBeatInterval()})
                .setTaskScheduler(taskScheduler);
        // 定义webSocket前缀
        registry.setApplicationDestinationPrefixes(webSocketProperties.getApplicationDestinationPrefixes());
    }

• yml

把stomp的相关配置做成配置文件,配置在yml中

commons.websocket:
  # 监听的节点
  endPoint: "/ws/stomp"
  # 跨域支持
  allowedOrigins: "*"
  # 可订阅的主题
  enableSimpleBroker:
   - "/topic"
   - "/queue"
   - "/user"
   - "/client"
  # 客户端向服务器发消息时的前缀
  applicationDestinationPrefixes: "/ws"

注册stomp节点

 stompEndpointRegistry.addEndpoint("/ws/stomp")

定义支持订阅的主题列表

  # 可订阅的主题
  enableSimpleBroker:
   - "/topic"
   - "/queue"
   - "/user"
   - "/client"

 registry.enableSimpleBroker(webSocketProperties.getEnableSimpleBroker());
 

定义跨域的支持

 stompEndpointRegistry.setAllowedOrigins("*")

定义心跳的支持

 @Override
public void configureMessageBroker(MessageBrokerRegistry registry) {

    // 定义心跳线程
    ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
    taskScheduler.setThreadNamePrefix("wss-heartbeat-thread-");
    taskScheduler.setDaemon(true);
    taskScheduler.initialize();

    // 服务端发送消息给客户端的域,多个用逗号隔开
    registry.enableSimpleBroker(webSocketProperties.getEnableSimpleBroker())
            // 定义心跳间隔 单位(ms)
            .setHeartbeatValue(new long[]{webSocketProperties.getHeartBeatInterval(), webSocketProperties.getHeartBeatInterval()})
            .setTaskScheduler(taskScheduler);
}

事件的监听

服务器可以监听到websocket的连接、已连接、订阅、退订、断开事件: .然后可以根据事件来做相应的业务处理.

• 例子

当某个客户端断开连接之后.发送消息到指定的topic

/**
 * 断开事件,当某个客户端断开连接之后.发送消息到指定的topic
 */
@Slf4j
@Component
public class WebSocketOnDisconnectEventListener implements ApplicationListener<SessionDisconnectEvent> {

    @Autowired
    private WebSocketService webSocketService;

    @Override
    public void onApplicationEvent(SessionDisconnectEvent sessionDisconnectEvent) {
        log.info("WebSocketOnDisconnectEventListener ... ");

        StompHeaderAccessor sha = StompHeaderAccessor.wrap(sessionDisconnectEvent.getMessage());

        if (sha.getSessionAttributes().get("onDisconnectTopic") != null) {
            String onDisconnectTopic = (String) sha.getSessionAttributes().get("onDisconnectTopic");
            String clientId = (String) sha.getSessionAttributes().get("clientId");

            webSocketService.send(
                    WebSocketMsgDefaultVo
                            .builder()
                            .payload(clientId + "断开连接")
                            .destination(onDisconnectTopic)
                            .build()
            );
        }
    }
}

session的获取

服务器可以监听浏览器连接成功事件,获取session信息,用来确定哪个浏览器

@Slf4j
@Component
public class WebSocketOnConnectedEventListener implements ApplicationListener<SessionConnectedEvent> {

    @Override
    public void onApplicationEvent(SessionConnectedEvent sessionConnectEvent) {
        String sessionId = (String) sessionConnectEvent.getMessage().getHeaders().get("simpSessionId");
        log.info("sessionId: {} ", sessionId);
        log.info("WebSocketOnConnectedEventListener ...");
    }
}

INFO  c.k.k.k.w.l.WebSocketOnConnectedEventListener - sessionId: 4gfxeh2z 
INFO  c.k.k.k.w.l.WebSocketOnConnectedEventListener - WebSocketOnConnectedEventListener ...

发送消息的接口

• spring boot中如何开启

浏览器发送消息给服务端,并且广播、点对点的发送给相应的其他浏览器.这里我们使用@MessageMapping注解来开启

• 自定义路由与封装的方法 例如 广播(broadcast)、点对点单播(unicast)

@Slf4j
@Controller
public class WebSocketController {

    @Autowired
    private WebSocketService webSocketService;

    @MessageMapping("/broadcast")
    public ResponseMessage broadcast(WebSocketMsgDefaultVo vo) throws Exception {
        log.info("/web_socket/broadcast test ... ", vo.toString());
        webSocketService.send(vo);
        return ResponseMessage.ok(vo.getPayload());
    }

    @MessageMapping("/unicast")
    public ResponseMessage unicast(WebSocketMsgDefaultVo vo) throws Exception {
        log.info("/web_socket/unicast test ... {} ", vo.toString());
        webSocketService.send(vo.getUserId(), vo);
        return ResponseMessage.ok(vo.getPayload());
    }
}

做成基础组件

可以把上面整合spring boot的示例.做成基础组件starter.给其他模块调用.这样别人使用就可以不考虑整合的细节.只要关注与业务的实现

pom

<dependency>
      <groupId>com.example</groupId>
      <artifactId>websocket-starter</artifactId>
</dependency>

yml配置

commons.websocket:
  # 监听的节点
  endPoint: "/ws/stomp"
  # 跨域支持
  allowedOrigins: "*"
  # 可订阅的主题
  enableSimpleBroker:
   - "/topic"
   - "/queue"
   - "/user"
   - "/client"
  # 客户端向服务器发消息时的前缀
  applicationDestinationPrefixes: "/ws"
  # 心跳的间隔
  heartBeatInterval: 10000

前端使用

使用stomp js 来操作websocket

官网api地址

https://stomp-js.github.io/stomp-websocket/codo/class/Client.html

引入

<script type="text/javascript" src="http://cdn.jsdelivr.net/sockjs/0.3.4/sockjs.min.js"></script>
<script type="text/javascript" src="http://cdnjs.cloudflare.com/ajax/libs/stomp.js/2.3.3/stomp.min.js"></script>

连接

// 开启socket连接
    function connect() {
        var socket = new SockJS('/ws/stomp');
        stompClient = Stomp.over(socket);
        stompClient.connect({"userId": "1", "onDisconnectTopic": "/topic", "clientId": "1"}, function (frame) {
            setConnected(true);
            subscribe();
        });
    }

订阅

function subscribe() {
        console.log("subscribe");
        stompClient.subscribe("/topic", function (data) {
            var message = data.body;
            messageList.append("<li>" + message + "</li>");
        });
    }

发送消息

// 向‘/ws/customizedcast’服务端发送消息
    function sendName() {
        var value = document.getElementById('name').value;
        stompClient.send("/ws/clientcast", {}, JSON.stringify({
            "destination": "/topic",
            "payload": "payload " + value,
            "clientId": "1",
            "onErrorDestination":"/topic"
        }));
    }

断开

// 断开socket连接
    function disconnect() {
        if (stompClient != null) {
            stompClient.disconnect(function (frame) {
                setConnected(false);
            }, {"userId": "1", "onDisconnectTopic": "/topic", "clientId": "1"});
        }
        console.log("Disconnected");
    }

心跳

为了使客户端与服务器的连接保活(若客户端、服务器长时间不通信,就会断开)定义了一套维护心跳的机制.就是客户端会起定时任务发送ping帧,服务端收到返回一个pong帧消息.来保证连接的存活

>>> PING stomp.min.js:8 
<<< PONG stomp.min.js:8 

例子

简易聊天室

1. 打开浏览器A,B
2. A广播消息 1
3. B广播消息 2
4. A发送消息a给B
5. B发送消息b给A

• 最后显示如下

image.png

思考题

• 客户端如何处理断线重连机制
• 客户端如何处理事务的发送机制
• 服务器如何处理统一异常

参考资料

• https://spring.io/guides/gs/messaging-stomp-websocket/
• https://zh.wikipedia.org/wiki/WebSocket#cite_note-4
• https://stomp-js.github.io/stomp-websocket/codo/class/Client.html