若依集成WebSocket实现消息通知功能

实现效果

• A用户和B用户同时在线时，A用户提交申请后，B用户可以收到消息通知
  
• A用户在线，B用户不在线时，A用户可以提交申请，B用户上线后可以收到消息通知
  
• WebSocket刷新不会断开连接
  

先看效果，使用admin用户向test用户发送通知进行测试：

官方文档

我使用的是前后端分离版本的若依，前端使用的是vue3,官方文档中提供了集成WebSocket的方法：https://doc.ruoyi.vip/ruoyi-vue/document/cjjc.html#集成websocket实现实时通信 ，详细内容如下：
1、ruoyi-framework/pom.xml文件添加websocket依赖。

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

复制代码

2、配置匿名访问（可选）

1. // 如果需要不登录也可以访问，需要在`SecurityConfig.java`中设置匿名访问
2. ("/websocket/**").permitAll()

复制代码

3、下载插件相关包和代码实现覆盖到工程中

1. 提示
2. 插件相关包和代码实现ruoyi-vue/集成websocket实现实时通信.zip
3. 链接: https://pan.baidu.com/s/1y1g8NkelRT_pS0fIbmyP8g 提取码: mjs7

复制代码

4、测试验证
如果要测试验证可以把websocket.vue内容复制到login.vue，点击连接发送消息测试返回结果。
代码实现

官方文档中提供了SemaphoreUtils,WebSocketConfig,WebSocketServer和WebSocketUsers四个后端java文件用于集成WebSocket，我的代码仅在WebSocketServer中做修改，其余文件没有修改，修改后的代码如下。
此处为WebSocket实现消息通知的简易版思路，仅为玩具，不考虑性能，应该会存在bug，整体思路如下：

1、用户登陆成功后，向后端发送当前用户名，并在本地localstorage中存储值（用户刷新时使用）
2、后端接收到前端发送的用户名后，判断该用户是否在后端维护的用户列表中，如果不在，则新建用户，如果在，则说明该用户在前端进行了刷新，那么就不新建用户，而是更新现有用户列表中的信息
3、A用户向B用户发送消息，如果B用户在线，则通过后端直接发送消息给B用户，如果B用户不在线，则在后端维护一个缓存消息列表，保存B用户的用户名和需要发送的消息
4、当后端接收到前端新建WebSocket连接的请求后，判断该用户是否在后端维护的用户列表中，如果在，接着查找是否有发送给该用户的消息，如果有则发送消息通知，如果没有则更新之前用户列表中维护的信息，如果该用户不在已有用户列表中，则新建用户

修改后的WebSocketServer内容如下：

1. package com.ruoyi.framework.websocket;
3. import java.util.concurrent.ConcurrentHashMap;
4. import java.util.concurrent.CopyOnWriteArrayList;
5. import java.util.concurrent.Semaphore;
6. import javax.websocket.OnClose;
7. import javax.websocket.OnError;
8. import javax.websocket.OnMessage;
9. import javax.websocket.OnOpen;
10. import javax.websocket.Session;
11. import javax.websocket.server.ServerEndpoint;
13. import com.fasterxml.jackson.databind.JsonNode;
14. import com.fasterxml.jackson.databind.ObjectMapper;
15. import org.slf4j.Logger;
16. import org.slf4j.LoggerFactory;
17. import org.springframework.stereotype.Component;
19. /**
20. * WebSocketServer 消息处理类，用于实现 WebSocket 服务端功能
21. *
22. * @author ruoyi
23. */
24. @Component
25. @ServerEndpoint("/websocket/message")
26. public class WebSocketServer {
27.     /**
28.      * WebSocketServer 日志控制器，用于记录日志信息
29.      */
30.     private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);
32.     /**
33.      * 默认最多允许同时在线人数 100
34.      */
35.     public static int socketMaxOnlineCount = 100;
37.     /**
38.      * 信号量，用于控制同时在线人数
39.      */
40.     private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);
42.     // 内部静态类 User，用于存储用户信息
43.     private static class User {
44.         private String id; // 用户 ID
45.         private String name; // 用户名
46.         private Session session; // 用户会话
48.         public User(String id, String name, Session session) {
49.             this.id = id;
50.             this.name = name;
51.             this.session = session;
52.         }
54.         // Getter 和 Setter 方法
55.         public String getId() {
56.             return id;
57.         }
59.         public void setId(String id) {
60.             this.id = id;
61.         }
63.         public String getName() {
64.             return name;
65.         }
67.         public void setName(String name) {
68.             this.name = name;
69.         }
71.         public Session getSession() {
72.             return session;
73.         }
75.         public void setSession(Session session) {
76.             this.session = session;
77.         }
78.     }
80.     /**
81.      * 存储用户信息的 ConcurrentHashMap，键为用户 ID，值为 User 对象
82.      */
83.     private static ConcurrentHashMap<String, User> users = new ConcurrentHashMap<>();
85.     /**
86.      * 消息缓存的 ConcurrentHashMap，键为用户名，值为缓存消息列表
87.      */
88.     private static ConcurrentHashMap<String, CopyOnWriteArrayList<String>> messageCache = new ConcurrentHashMap<>();
90.     /**
91.      * 连接建立成功时调用的方法
92.      *
93.      * @param session 用户会话
94.      * @throws Exception 可能抛出的异常
95.      */
96.     @OnOpen
97.     public void onOpen(Session session) throws Exception {
98.         boolean semaphoreFlag = false;
99.         // 尝试获取信号量
100.         semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);
101.         if (!semaphoreFlag) {
102.             // 未获取到信号量，说明在线人数已达上限
103.             LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);
104.             WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);
105.             session.close();
106.         } else {
107.             // 将用户会话存储到 WebSocketUsers 中
108.             WebSocketUsers.put(session.getId(), session);
109.             LOGGER.info("\n 建立连接 - {}", session);
110.             LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());
111.             WebSocketUsers.sendMessageToUserByText(session, "连接成功");
112.         }
113.     }
115.     /**
116.      * 连接关闭时调用的方法
117.      *
118.      * @param session 用户会话
119.      */
120.     @OnClose
121.     public void onClose(Session session) {
122.         LOGGER.info("\n 关闭连接 - {}", session);
123.         // 移除用户信息，移除用户时，将该用户的session值置为null
124.         // 用于判断该用户是否离线
125.         for (User user : users.values()) {
126.             if (user.getId().equals(session.getId())) {
127.                 user.setSession(null);
128.                 break;
129.             }
130.         }
132.         boolean removeFlag = WebSocketUsers.remove(session.getId());
133.         if (!removeFlag) {
134.             // 如果移除失败，则释放信号量
135.             SemaphoreUtils.release(socketSemaphore);
136.         }
137.     }
139.     /**
140.      * 抛出异常时调用的方法
141.      *
142.      * @param session     用户会话
143.      * @param exception   异常信息
144.      * @throws Exception  可能抛出的异常
145.      */
146.     @OnError
147.     public void onError(Session session, Throwable exception) throws Exception {
148.         if (session.isOpen()) {
149.             // 如果会话仍然开启，则关闭会话
150.             session.close();
151.         }
152.         String sessionId = session.getId();
153.         LOGGER.info("\n 连接异常 - {}", sessionId);
154.         LOGGER.info("\n 异常信息 - {}", exception);
155.         // 移除用户信息并释放信号量
156.         WebSocketUsers.remove(sessionId);
157.         SemaphoreUtils.release(socketSemaphore);
158.     }
160.     /**
161.      * 服务器接收到客户端消息时调用的方法
162.      *
163.      * @param message 消息内容
164.      * @param session 用户会话
165.      */
166.     @OnMessage
167.     public void onMessage(String message, Session session) {
168.         try {
169.             // 使用 Jackson 解析 JSON 消息
170.             ObjectMapper objectMapper = new ObjectMapper();
171.             JsonNode jsonNode = objectMapper.readTree(message);
172.             if (jsonNode.has("username") && !jsonNode.get("username").asText().isEmpty()) {
173.                 String username = jsonNode.get("username").asText();
174.                 // 检查用户是否已存在
175.                 boolean userExists = false;
176.                 for (User user : users.values()) {
177.                     if (user.getName().equals(username)) {
178.                         userExists = true;
179.                         // 更新用户的 ID 和会话
180.                         user.setId(session.getId());
181.                         user.setSession(session);
183.                         // 检查该用户是否有缓存消息
184.                         searchCache(username, user);
186.                         LOGGER.info("\n用户 {} 已更新", username);
187.                         LOGGER.info("\n当前用户列表 {}", users);
188.                         break;
189.                     }
190.                 }
191.                 if (!userExists) {
192.                     // 新建用户并存储到 users 中
193.                     User user = new User(session.getId(), username, session);
194.                     users.put(session.getId(), user);
195.                     LOGGER.info("\n用户 {} 已连接", username);
196.                     LOGGER.info("\n当前用户列表 {}", users);
198.                     // 检查是否有缓存消息
199.                     searchCache(username, user);
200.                 }
201.             } else if (jsonNode.has("targetusername") && jsonNode.has("sendmessage")) {
202.                 String targetusername = jsonNode.get("targetusername").asText();
203.                 String sendmessage = jsonNode.get("sendmessage").asText();
205.                 boolean userExists = false;
206.                 // 查找目标用户，如果用户的session为空，则表示用户离线
207.                 // 将通知存放到缓存消息列表中，如果不为空则表示用户在线，直接发送通知
208.                 for (User user : users.values()) {
209.                     if (user.getName().equals(targetusername)) {
210.                         if (user.getSession() != null) {
211.                             userExists = true;
212.                             WebSocketUsers.sendMessageToUserByText(user.getSession(), sendmessage);
213.                         }
214.                         break;
215.                     }
216.                 }
218.                 LOGGER.info("\n目标用户 {}", targetusername);
219.                 LOGGER.info("\n发送消息 {}", sendmessage);
220.                 LOGGER.info("\n当前用户列表 {}", users);
222.                 if (!userExists) {
223.                     // 如果目标用户不在线，则缓存通知
224.                     messageCache.computeIfAbsent(targetusername, k -> new CopyOnWriteArrayList<String>()).add(sendmessage);
225.                     LOGGER.info("\n消息已缓存，等待用户 {} 上线", targetusername);
226.                 }
227.             }
229.         } catch (Exception e) {
230.             // 如果解析失败，则按普通消息处理
231.             String msg = message.replace("你", "我").replace("吗", "");
232.             WebSocketUsers.sendMessageToUserByText(session, msg);
233.         }
234.     }
236.     /**
237.      * 检查一个用户的缓存通知并发送
238.      *
239.      * @param username 用户名
240.      * @param user     用户对象
241.      */
242.     public void searchCache(String username, User user) {
243.         if (messageCache.containsKey(username)) {
244.             CopyOnWriteArrayList<String> cachedMessages = messageCache.get(username);
245.             for (String cachedMessage : cachedMessages) {
246.                 WebSocketUsers.sendMessageToUserByText(user.getSession(), cachedMessage);
247.                 LOGGER.info("\n向用户 {} 发送缓存消息 - {}", username, cachedMessage);
248.             }
249.             // 清理缓存通知
250.             messageCache.remove(username);
251.             LOGGER.info("\n已清理用户 {} 的缓存消息", username);
252.         }
253.     }
254. }

复制代码

为了维护WebSocket的状态，在pinia store中新建了websocket.js文件（路径为RuoYi-Vue3-master\src\store\modules\websocket.js）。
websocket.js内容如下：

1. import { ref } from 'vue';
2. import { ElNotification } from 'element-plus'; // 引入 Element UI 的通知功能
3. import 'element-plus/dist/index.css';          // 引入 Element UI 的通知功能样式
5. export const useWebSocketStore = defineStore('websocket', () => {
6.   const ws = ref(null);
7.   const connected = ref(false);
9.   function initWebSocket(username) {
11.     const wsuri = "ws://127.0.0.1:8080/websocket/message"; // WebSocket 服务器地址
12.     ws.value = new WebSocket(wsuri);
14.     ws.value.onopen = () => {
15.       console.log('WebSocket 已连接');
16.       connected.value = true;
17.       ws.value.send(JSON.stringify({ username: username }));
18.     };
20.     ws.value.onmessage = (event) => {
21.       console.log('收到消息:', event.data);
22.       // 在这里处理收到的消息
23.       if (event.data != "连接成功") {    // 用于测试，所以消息字符都是写死的
24.         ElNotification({
25.             title: '消息通知',
26.             message: '您收到一条消息',
27.             type: 'success',
28.             position: 'bottom-right',
29.         });
30.       }
31.     };
33.     ws.value.onclose = () => {
34.       console.log('WebSocket 已断开');
35.       connected.value = false;
36.     };
38.     ws.value.onerror = (event) => {
39.       console.error('WebSocket 错误:', event);
40.       connected.value = false;
41.     };
42.   }
44.   // 发送消息
45.   function sendmessage(username, message) {
46.     console.log(ws.value);
47.     console.log(message);
49.     if (message == "请求审批") {    // 用于测试，所以消息字符都是写死的
50.         ws.value.send(JSON.stringify({ targetusername: username, sendmessage: message }));
51.     } else {
52.         ws.value.send(JSON.stringify({ logoutname: username, logoutmessage: message }));
53.     }
54.   }
55.   
56.   function closeWebSocket() {    // 此处仅为测试，写了close，但是没有使用，
57.     if (ws.value) {
58.       ws.value.close();
59.       ws.value = null;
60.       connected.value = false;
61.     }
62.   }
64.   return { ws, initWebSocket, closeWebSocket, sendmessage };
65. });

复制代码

我判断用户是否登陆的思路是，如果加载了（RuoYi-Vue3-master\src\layout）页面，那么用户肯定是登陆状态，所以就在localstorage中写入websocket:1,然后新建一个WebSocket连接，实现代码如下,如果用户刷新了，那么WebSocket连接会断开，但是用户刷新时，该组件会被重新加载，因此在onMounted中判断localstorage中是否存在标志，如果存在说明该用户在后端的用户列表中，那么直接将username发送给后端，后端在判断时会更新用户列表中的内容，从而实现用户刷新但是WebSocket没有断开。
修改后的layout代码如下：

1. ...省略...
3. ...省略...

复制代码

我在测试文件中添加了以一个申请按钮，并绑定了一个事件，实现代码如下：

1. <template>
2. ...省略...
3. // 按钮代码
4. <el-col :span="1.5">
5.   <el-button
6.     type="info"
7.     plain
8.     icon="DocumentAdd"
9.     @click="handleApply"
10.   >申请</el-button>
11. </el-col>
12. ...省略...
13. </template>

复制代码

至此，就实现了一个简易版本的用户消息通知功能，如果要实现不同用户之间的审批功能（玩具），只需要在前端根据用户角色的不同展示不同的页面就可以了，大致效果如下。

1. 1.前端通过import { checkRole } from "@/utils/permission";并在标签页中使用v-if来判断用户的角色以显示不同的页面。
2. 2.通过自定义监听事件实现收到消息后表格刷新。
3. （1）前端页面<script>中添加：
4. // 定义一个方法来刷新数据
5. function refreshData() {
6.   getList();
7. }
8. // 在组件挂载时监听自定义事件
9. onMounted(() => {
10.   window.addEventListener('refreshData', refreshData);
11. });
12. // 在组件卸载时移除事件监听器
13. onUnmounted(() => {
14.   window.removeEventListener('refreshData', refreshData);
15. });
17. （2）websocket.js中在ElNotification后添加：
18. ws.value.onmessage = (event) => {
19.   console.log('收到消息:', event.data);
20.   // 在这里处理收到的消息
21.   if (event.data != "连接成功") {
22.     ElNotification({
23.         title: '消息通知',
24.         message: '您收到一条消息',
25.         type: 'success',
26.         position: 'bottom-right',
27.     });
29.     // 收到消息后0.5秒刷新页面
30.     setTimeout(() => {
31.       // 收到消息后触发自定义事件来刷新数据
32.       const event = new Event('refreshData');
33.       window.dispatchEvent(event);
34.     }, 500);
35.   }
36. };

复制代码

来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！