SpringBoot +WebSocket应用

我们今天不研究原理，只看应用。

什么是WebSocket

WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket通信协议于2011年被IETF定为标准RFC 6455，并由RFC7936补充规范。WebSocket API也被W3C定为标准。
WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

以上是百度百科的WebSocket定义。其实我们就可以理解WebSocket是一种全双工通讯协议，也就是服务端和客户端都可以发送、接收信息，对比Http协议想一下就很容易理解：Http协议是无状态的单向通讯协议，只能是客户端发送请求给服务端、服务端返回响应给客户端，之后连接就关闭了，服务端是不可以主动发送消息给客户端的。

应用场景

为什么需要WebSocket？

WebSocket是为了解决Web应用场景下特定需求的，Web应用使用的是http协议，服务端是没有办法主动发送消息给客户端的，但是某些场景下恰恰需要服务端主动发送消息给客户端，比如实时数据监控的场景：服务端定时（比如每分钟）获取到被监控设备的实时数据，前端页面实时展示监控值，当服务器端获取到新的数据的时候，要求前端页面相应的进行实时展示。

如果不采用WebSocket，一般会采用如下两种方式实现以上需求：

1. 轮询：前端js定时轮询，发起查询请求，如果有新数据的话当然就可以获取到，但是即使没有新数据，前端也不会知道，只能机械式的不断轮询。
2. 长轮询（long Polling）：阻塞式轮询，拿不到数据就不返回一直等待，其实是一种更傻的轮询方式。

轮询方式是基于Http协议实现的，主要缺点是开销大，因为每次轮询都需要发起http请求。

而WebSocket是该应用场景下以上两种轮询方式的理想替代方案，WebSocket仅在首次与服务器通讯的时候采用http协议，建立链接之后立即升级协议为WebSocket协议，建立长连接，之后的通讯都会工作在这个长连接下，所以，可以极大减少及时通讯业务场景下的连接开销。

SpringBoot集成WebSocket

SpringBoot对WebSocket做了无缝支持，使用起来也非常方便。

第一步：引入依赖。

        
            org.springframework.boot
            spring-boot-starter-websocket
        
1
2
3
4

第二步：使用ServerEndpoint注解创建WebSocket服务端：

package com.example.service;

import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@ServerEndpoint(value = "/websocket")
@Component
public class MyWebSocket {

    //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
    private static int onlineCount = 0;

    //concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet();

    //与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;

    //记录客户端特性：比如页面查询参数等等
    private String params;
    private HashMap frontEndData = new HashMap<>();

    /**
     * 连接建立成功调用的方法*/
    @OnOpen
    public void onOpen(Session session) {
        this.session = session;
        webSocketSet.add(this);     //加入set中
        addOnlineCount();           //在线数加1
        System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
        try {
            sendMessage(session.getId()+ ":connected...");
        } catch (IOException e) {
            System.out.println("IO异常");
        }
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);  //从set中删除
        subOnlineCount();           //在线数减1
        System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message 客户端发送过来的消息*/
    @OnMessage
    public void onMessage(String message, Session session) {
        System.out.println("来自客户端的消息:" + message);
        this.params = message;
        try{
            sendMessage(message);
        }catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 发生错误时调用
     **/
     @OnError
     public void onError(Session session, Throwable error) {
         System.out.println("发生错误");
         error.printStackTrace();
     }


     public void sendMessage(String message) throws IOException {
        this.session.getBasicRemote().sendText(message);
         //this.session.getAsyncRemote().sendText(message);
         //群发消息
     }

    //给所有客户端发送消息
     public void sendMessageToAll(String message,MonitorService monitorService){
         for (MyWebSocket item : webSocketSet) {
             try {
                 if(monitorService.matchMessage(item.params,message))
                    item.sendMessage(message);
             } catch (IOException e) {
                 e.printStackTrace();
             }
         }
     }

     /**
      * 群发自定义消息
      * */
    public static void sendInfo(String message) throws IOException {
        for (MyWebSocket item : webSocketSet) {
            try {
                item.sendMessage(message);
            } catch (IOException e) {
                continue;
            }
        }
    }

    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    public static synchronized void addOnlineCount() {
        MyWebSocket.onlineCount++;
    }

    public static synchronized void subOnlineCount() {
        MyWebSocket.onlineCount--;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

第三部：创建一个MonitorService类，通过前台页面模拟设备数据提交给MonitorService，MonitorService接收到数据之后调用MyWebSocket 的sendMessageToAll方法，将设备监控数据发送给前端进行展示：

package com.example.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class MonitorService {
    @Autowired
    private MyWebSocket myWebSocket;
    public void getMonitorValue(String value){
        sendValue(value);
    }

    private void sendValue(String value) {
        try{
            myWebSocket.sendMessageToAll(value,this);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    public boolean matchMessage(String params,String message){
        //信息匹配
        if(message.startsWith(params)){
            return true;
        }
        return false;
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

matchMessage方法做一个简单的业务场景的模拟：前端页面通过WebSocket发送一个订阅信息上来，表示当前页面订阅某一种类型的监测数据，matchMessage方法暂定一个特别简单的逻辑：以订阅信息打头的监测数据仅发送给该订阅客户端。比如客户端订阅“TTT”信息，则监测数据为TTTabce123将发送给该客户端，其他非TTT打头的监测数据将不会发送给该客户端。

第四步：创建Controller，调用MonitorService的getMonitorValue方法，模拟设备监控数据的获取、并发送给订阅该数据的客户端。

package com.example.controller;

import com.example.service.MonitorService;
import com.example.service.TestProperties;
import com.example.service.UserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.*;

@RestController
@RequestMapping("/hello")
@Slf4j
public class HelloWorldController {

    @Autowired
    private MonitorService monitorService;
 
    @GetMapping ("/sendmessage/{msg}")
    public String sendMessage(@PathVariable String msg){
        monitorService.getMonitorValue(msg);
        return "hello";
    }


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

第五步：客户端代码，写入websocket.htm文件中：

    



Welcome

Send    Close






1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

运行测试

启动SpringBoot项目，使用浏览器打开websocket.htm，文本框中输入ttt，表示订阅以ttt开头的实时监测数据，点击send按钮，页面回显ttt，表示客户端与服务端通过WebSocket成功发送、接收了信息：

浏览器再打开两个websocket.htm页面，分别发送ccc，AAA：


浏览器访问我们编写的模拟发送监测数据的服务，发送ttt12323，页面收到返回信息hello，表示发送成功：

查看已经打开的websocket.htm页面，发现订阅ttt服务的页面收到了服务器端通过websocket协议推送的信息：

另外两个页面没有收到信息，按照我们的设计，另外两个页面分别定制的是ccc以及AAA信息所以不会收到ttt信息。因此我们分别再次发送ccc及AAA信息：


检查websocket.htm页面：


ccc以及AAA客户端分别接收到了对应的信息，测试成功！

总结

在SpringBoot项目中使用WebSocket其实非常简单：

1. 引入依赖。
2. 通过@ServerEndpoint注解编写WebScoket服务端。
3. 前端页面通过js代码编写WebSocket客户端。

以上三步，一个简单的WebSocket应用就可以正常工作了。

OK，Many thanks！