Springboot集成websocket

Springboot集成websocket

1. WebSocket概述

我们再日常的web应用开发过程中，常见的是前端向后端发起请求，但有的时候需要后端主动给前端发送消息，这个时候全双工的websocket即时通讯就闪亮登场了。

• 全双工通讯模式允许双方同时进行双向通讯，如手机通话。
• 半双工通讯模式允许双方交替发送和接收信息，但不能同时通讯，如对讲机。
• 单工通讯模式只能单向传输信息，不能回复，如广播电台。

2. WebSocket原理

WebSocket是一种在单个TCP连接上进行全双工通信的协议。它通过一个简单的握手过程来建立连接，然后在连接上进行双向数据传输。与传统的HTTP请求不同，WebSocket连接一旦建立，就可以在客户端和服务器之间保持打开状态，直到被任何一方关闭。

WebSocket协议的核心特点包括：

• 全双工通信：客户端和服务器可以同时发送和接收消息。
• 持久连接：一旦建立连接，就可以持续进行数据交换，无需像HTTP那样频繁地建立新的连接。
• 低延迟：由于连接是持久的，数据可以几乎实时地发送和接收。
• 轻量级协议：WebSocket协议的头部信息非常简单，减少了数据传输的开销。

3. Spring Boot集成WebSocket

在Spring Boot中集成WebSocket非常简单，Spring提供了对WebSocket的原生支持。以下是一个基本的集成步骤：

3.1 添加依赖

在pom.xml中添加Spring Boot的WebSocket依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
1
2
3
4

3.2 创建WebSocket配置类

创建一个配置类来启用和配置WebSocket：

package com.jiayuan.common.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;
import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;

/**
 * @Title: WebSocketConfig
 * @Package: com.jiayuan.common.config
 * @Description: websocket配置
 * @Author: xmc
 * @Date: 创建时间 2024-04-24
 */
@Configuration
public class WebSocketConfig {

    /**
     * 自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
     *
     * @return
     */

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

    /**
     * 通信文本消息和二进制缓存区大小
     * 避免对接 第三方 报文过大时，Websocket 1009 错误
     *
     * @return
     */

    @Bean
    public ServletServerContainerFactoryBean createWebSocketContainer() {
        ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
        // 在此处设置bufferSize
        container.setMaxTextMessageBufferSize(10240000);
        container.setMaxBinaryMessageBufferSize(10240000);
        container.setMaxSessionIdleTimeout(15 * 60000L);
        return container;
    }
}
 
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

3.3 创建消息处理器

package com.jiayuan.common.config;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.jiayuan.common.redis.RedisCache;
import com.jiayuan.modules.critical.dto.CvRecordExtraDTO;
import com.jiayuan.modules.critical.dto.SyncRecordDTO;
import com.jiayuan.modules.critical.service.CvRecordService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Title: WebSocketServer
 * @Package: com.jiayuan.common.config
 * @Description: websocket的服务端
 * @Author: xmc
 * @Date: 创建时间 2024-04-24
 */
@Component
@Slf4j
@ServerEndpoint("/api/pushMessage/{userId}")
public class WebSocketServer {

    /**
     * 静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
     */
    private static int onlineCount = 0;
    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的WebSocket对象。
     */
    private static ConcurrentHashMap<String, WebSocketServer> webSocketMap = new ConcurrentHashMap<>();
    /**
     * 与某个客户端的连接会话，需要通过它来给客户端发送数据
     */
    private Session session;
    /**
     * 接收userId
     */
    private String userId = "";

    /**
     * 连接建立成
     * 功调用的方法
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        this.session = session;
        this.userId = userId;
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            //加入set中
            webSocketMap.put(userId, this);
        } else {
            //加入set中
            webSocketMap.put(userId, this);
            //在线数加1
            addOnlineCount();
        }
        log.info("用户连接:" + userId + ",当前在线人数为:" + getOnlineCount());
        sendMessage("连接成功");
    }

    /**
     * 连接关闭
     * 调用的方法
     */
    @OnClose
    public void onClose() {
        if (webSocketMap.containsKey(userId)) {
            webSocketMap.remove(userId);
            //从set中删除
            subOnlineCount();
        }
        log.info("用户退出:" + userId + ",当前在线人数为:" + getOnlineCount());
    }

    /**
     * 收到客户端消
     * 息后调用的方法
     *
     * @param message 客户端发送过来的消息
     **/
    @OnMessage
    public void onMessage(String message, Session session) {
        log.info("用户消息:" + userId + ",报文:" + message);
        //可以群发消息
        //消息保存到数据库、redis
        if (StringUtils.isNotBlank(message)) {
            try {
                //解析发送的报文
                JSONObject jsonObject = JSON.parseObject(message);
                //追加发送人(防止串改)
                jsonObject.put("fromUserId", this.userId);
                String toUserId = jsonObject.getString("toUserId");
                //传送给对应toUserId用户的websocket
                if (StringUtils.isNotBlank(toUserId) && webSocketMap.containsKey(toUserId)) {
                    webSocketMap.get(toUserId).sendMessage(message);
                } else {
                    //否则不在这个服务器上，发送到mysql或者redis
                    log.error("请求的userId:" + toUserId + "不在该服务器上");
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {

        log.error("用户错误:" + this.userId + ",原因:" + error.getMessage());
        error.printStackTrace();
    }

    /**
     * 实现服务
     * 器主动推送
     */
    public void sendMessage(String message) {
        try {
            this.session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送自定
     * 义消息
     **/
    public static void sendInfo(String message, String userId) {
        log.info("发送消息到:" + userId + "，报文:" + message);
        if (StringUtils.isNotBlank(userId) && webSocketMap.containsKey(userId)) {
            webSocketMap.get(userId).sendMessage(message);
        } else {
            log.error("用户" + userId + ",不在线！");
        }
    }

    /**
     * 获得此时的
     * 在线人数
     *
     * @return
     */
    public static synchronized int getOnlineCount() {
        return onlineCount;
    }

    /**
     * 在线人
     * 数加1
     */
    public static synchronized void addOnlineCount() {
        WebSocketServer.onlineCount++;
    }

    /**
     * 在线人
     * 数减1
     */
    public static synchronized void subOnlineCount() {
        WebSocketServer.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
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186

上述代码可能会有疑问：session和userId两个字段是不是安全的？多人连接，后面的会不会覆盖掉前面session和userId。

答案是不会的，关于处理器WebSocketServer我们要明确以下几点：

1. websocket是原型模式，@ServerEndpoint每次建立双向通信的时候都会创建一个实例
2. 为什么每次都@OnOpen都要检查webSocketMap.containsKey(userId) ，实际使用的时候发现偶尔会出现重连失败或者其他原因导致之前的session还存在，这时候就需要一个清除动作

3.4 服务器主动给客户端发送消息

 WebSocketServer.sendInfo("服务器主动给客户端发送消息test", "zhangsan");
1

4. 使用ApiPost测试WebSocket

以下是如何使用ApiPost进行测试的步骤：

1. 新建一个websocket测试

2. 填写URLws://localhost:8080/cvms-api/api/pushMessage/3

注意以下几点：

• 协议是ws，加密方式请选择wss
• 选择Raw
• URL的拼接公式如下

# servlet.context-path 这个是application.yml中的配置
ws://ip:port//${servlet.context-path}/注解@ServerEndpoint的值
1
2

• 有权限验证的，比如说shiro权限验证，URL就需要加入白名单

 filterMap.put("/api/pushMessage/*", "anon");
1

5. 参考和感谢

SpringBoot2.0集成WebSocket，实现后台向前端推送信息