SpringBoot+WebSocket实现即时通讯（一）_springboot聊天功能

作者：神奇cpp | 2024-07-28 03:26:08

踩

springboot聊天功能

前言

本博客姊妹篇

传统方式

背景：即时通讯过程中，解决传统网站使用HTTP轮询方式请求获取最新的数据（如每3秒请求一次）。

缺点：

Web客户端反复发出请求消耗服务器资源
请求包含较长的头部，浪费很多的带宽资源
只能由Web客户端发送请求到服务端获取数据
实时性不高

WebSocket

WebSocket：WebSocket是一种在单个TCP连接上进行全双工通信的协议。

优势：

一个Web客户端和服务端只建立一个TCP连接
请求包含轻量级的头部，减少了数据传输量
服务端可以主动推送数据到Web客户端
实时性高

一、功能描述

即时通讯：发送、接收消息
用户管理：业务自己实现，暂从数据库添加
好友管理：添加好友、删除好友、修改备注、好友列表等
群组管理：新建群、解散群、编辑群、变更群主、拉人进群、踢出群等
聊天模式：私聊、群聊
消息类型：系统、文本、语音、图片、视频
聊天管理：删除聊天、置顶聊天、查看聊天记录等

二、WebSocket服务

2.1 引入依赖

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

2.2 配置WebSocket扫描

package com.qiangesoft.im.config;

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

/**
 * WebSocket配置类
 */
@Configuration
public class WebSocketConfig {

    /**
     * bean注册：会自动扫描带有@ServerEndpoint注解声明的Websocket Endpoint(端点)，注册成为Websocket bean。
     * 注意：如果项目使用外置的servlet容器，而不是直接使用springboot内置容器的话，就不要注入ServerEndpointExporter，因为它将由容器自己提供和管理。
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

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

2.3 WebSocket服务类

package com.qiangesoft.im.core;

import com.alibaba.fastjson2.JSONObject;
import com.qiangesoft.im.core.constant.ChatTypeEnum;
import com.qiangesoft.im.core.constant.ImBodyEnum;
import com.qiangesoft.im.pojo.bo.ImMessageBO;
import com.qiangesoft.im.pojo.dto.PingDTO;
import com.qiangesoft.im.pojo.vo.ImMessageVO;
import com.qiangesoft.im.pojo.vo.PongVO;
import com.qiangesoft.im.pojo.vo.SysUserVo;
import com.qiangesoft.im.service.IImGroupUserService;
import com.qiangesoft.im.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

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;
import java.util.stream.Collectors;

/**
 * 聊天会话
 *
 * @author qiangesoft
 * @date 2023-08-30
 */
@Slf4j
@ServerEndpoint("/ws/im/{userId}")
@Component
public class ImWebSocketServer {

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的session
     */
    private static final ConcurrentHashMap<Long, Session> WEBSOCKET_MAP = new ConcurrentHashMap<>();

    /**
     * 连接成功：用map存客户端对应的session
     */
    @OnOpen
    public void onOpen(Session session, @PathParam("userId") Long userId) {
        log.info("User [{}] connection opened=====>", userId);

        // 关闭之前的
        if (WEBSOCKET_MAP.containsKey(userId)) {
            Session oldSession = WEBSOCKET_MAP.get(userId);
            close(oldSession, userId);
        }

        // 存储session
        WEBSOCKET_MAP.put(userId, session);

        // 在线人数
        log.info("User connection add 1, online num is [{}]", WEBSOCKET_MAP.size());

        // 响应
        PongVO pongVO = new PongVO();
        pongVO.setType(ImBodyEnum.PONG.getCode());
        pongVO.setContent("连接成功");
        pongVO.setTimestamp(System.currentTimeMillis());
        doSendMessage(session, pongVO);
    }

    /**
     * 收到客户端消息
     */
    @OnMessage
    public void onMessage(Session session, @PathParam("userId") Long userId, String message) {
        log.info("User [{}] send a message, content is [{}]", userId, message);

        PingDTO pingDTO = null;
        try {
            pingDTO = JSONObject.parseObject(message, PingDTO.class);
        } catch (Exception e) {
            log.error("消息解析失败");
            e.printStackTrace();
        }
        if (pingDTO == null || !ImBodyEnum.PING.getCode().equals(pingDTO.getType())) {
            sendInValidMessage(session);
            return;
        }

        // 响应
        PongVO pongVO = new PongVO();
        pongVO.setType(ImBodyEnum.PONG.getCode());
        pongVO.setContent("已收到消息~");
        pongVO.setTimestamp(System.currentTimeMillis());
        doSendMessage(session, pongVO);
    }

    /**
     * 连接关闭调用的方法
     */
    @OnClose
    public void onClose(Session session, @PathParam("userId") Long userId) {
        close(session, userId);

        // 在线人数减1
        if (!WEBSOCKET_MAP.containsKey(userId)) {
            log.info("User connection reduce 1, online num is [{}]", WEBSOCKET_MAP.size());
        }

        log.info("User [{}] connection is closed<=====", userId);
    }

    /**
     * 报错
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, @PathParam("userId") Long userId, Throwable error) {
        log.info("User [{}] connection is error!", userId);
        error.printStackTrace();
    }

    /**
     * 指定的userId服务端向客户端发送消息
     */
    public static void sendMessage(ImMessageBO message) {
        String chatType = message.getChatType();
        if (ChatTypeEnum.GROUP.getCode().equals(chatType)) {
            sendGroupMessage(message);
        }

        if (ChatTypeEnum.PERSON.getCode().equals(chatType)) {
            sendPersonMessage(message);
        }
    }

    /**
     * 被挤下线
     */
    public static void offline(Long userId) {
        Session session = WEBSOCKET_MAP.get(userId);
        if (session != null) {
            // 设备下线
            PongVO pongVO = new PongVO();
            pongVO.setType(ImBodyEnum.OFFLINE.getCode());
            pongVO.setContent("设备被挤下线");
            pongVO.setTimestamp(System.currentTimeMillis());
            doSendMessage(session, pongVO);

            // 关闭
            close(session, userId);
        }
    }

    /**
     * 自定义关闭
     *
     * @param session
     * @param userId
     */
    public static void close(Session session, Long userId) {
        try {
            session.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        WEBSOCKET_MAP.remove(userId);
    }

    /**
     * 发送无效消息
     */
    private static void sendInValidMessage(Session session) {
        PongVO pongVO = new PongVO();
        pongVO.setType(ImBodyEnum.PONG.getCode());
        pongVO.setContent("无效消息");
        pongVO.setTimestamp(System.currentTimeMillis());
        doSendMessage(session, pongVO);
    }

    /**
     * 发送群组消息
     *
     * @param message
     */
    private static void sendGroupMessage(ImMessageBO message) {
        MessageHandlerService messageHandlerService = SpringUtil.getBean(MessageHandlerService.class);
        ImMessageVO messageVO = messageHandlerService.buildVo(message);

        PongVO pongVO = new PongVO();
        pongVO.setType(ImBodyEnum.MESSAGE.getCode());
        pongVO.setContent(messageVO);
        pongVO.setTimestamp(System.currentTimeMillis());

        // 发送给群成员
        IImGroupUserService groupUserService = SpringUtil.getBean(IImGroupUserService.class);
        List<Long> userIdList = groupUserService.listGroupUser(message.getTargetId()).stream().map(SysUserVo::getId).collect(Collectors.toList());
        for (Long userId : userIdList) {
            Session session = WEBSOCKET_MAP.get(userId);
            doSendMessage(session, pongVO);
        }
    }

    /**
     * 发送私聊消息
     *
     * @param message
     */
    private static void sendPersonMessage(ImMessageBO message) {
        MessageHandlerService messageHandlerService = SpringUtil.getBean(MessageHandlerService.class);
        ImMessageVO messageVO = messageHandlerService.buildVo(message);

        PongVO pongVO = new PongVO();
        pongVO.setType(ImBodyEnum.MESSAGE.getCode());
        pongVO.setContent(messageVO);
        pongVO.setTimestamp(System.currentTimeMillis());

        // 发送给好友
        Session session = WEBSOCKET_MAP.get(message.getTargetId());
        doSendMessage(session, pongVO);
    }

    /**
     * 发送消息
     *
     * @param session
     * @param message
     */
    private static void doSendMessage(Session session, PongVO message) {
        try {
            if (session != null) {
                session.getBasicRemote().sendText(JSONObject.toJSONString(message));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

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
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237

三、源码地址

源码地址：https://gitee.com/qiangesoft/boot-business/tree/master/boot-business-im

后续内容见下章

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/神奇cpp/article/detail/892639