WebSocket+Http实现功能加成

WebSocket+Http实现功能加成

前言

首先，WebSocket和HTTP是两种不同的协议，它们在设计和用途上有一些显著的区别。以下是它们的主要特点和区别：

HTTP (HyperText Transfer Protocol):

1. 请求-响应模型： HTTP 是基于请求-响应模型的协议，客户端发送请求，服务器返回响应。这是一个单向通信的模式。

2. 无状态： HTTP 是无状态的协议，每个请求都是独立的，服务器不会记住之前的请求状态。为了维护状态，通常使用会话（Session）和 Cookie。

3. 连接短暂： 每个请求都需要建立一个新的连接，然后在响应后关闭。这种短暂的连接模型适用于传统的网页浏览，但对于实时通信则显得不够高效。

4. 端口： 默认使用端口80进行通信（HTTPS使用端口443）。

WebSocket:

1. 双向通信： WebSocket 提供全双工通信，允许在客户端和服务器之间建立持久性的双向连接，实现实时数据传输。

2. 状态： WebSocket 连接是持久性的，服务器和客户端之间可以保持状态。这减少了每次通信时的开销，使其适用于需要频繁交换数据的实时应用。

3. 协议升级： WebSocket 连接通常通过HTTP协议的升级头部进行初始化，然后升级到WebSocket连接。这样的设计允许在HTTP和WebSocket之间平稳切换，同时在需要时实现更高级的协议升级。

4. 端口： 默认使用端口80进行非安全通信，使用端口443进行安全通信。

5. 低开销： 相比于HTTP轮询或长轮询，WebSocket 通信的开销较低，因为它减少了头部信息的传输，同时避免了不必要的连接和断开。

HTTP适用于传统的请求-响应模型，而WebSocket适用于需要实时、双向通信的应用场景，如在线游戏、聊天应用和实时协作工具。在某些情况下，它们可以结合使用，以充分发挥各自的优势。

代码实现

在这里，我们首先确定一个事情，就是这个我们以若依为主要的参考，在若依的基础上进行演示和操作，所以，对于若依这套架子有成见的同志请移步！!我们先创建一个信号量相关处理的工具类，作为辅助工具!

这段 Java 代码是一个用于处理信号量（Semaphore）的工具类，其中包含了获取信号量和释放信号量的方法。信号量是一种用于控制同时访问特定资源的线程数的同步工具。

package com.ruoyi.framework.websocket;import java.util.concurrent.Semaphore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/*** 信号量相关处理* * @author ruoyi*/
public class SemaphoreUtils
{/*** SemaphoreUtils 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(SemaphoreUtils.class);/*** 获取信号量* * @param semaphore* @return*/public static boolean tryAcquire(Semaphore semaphore){boolean flag = false;try{// 尝试获取信号量，如果成功返回 true，否则返回 falseflag = semaphore.tryAcquire();}catch (Exception e){// 捕获异常并记录日志LOGGER.error("获取信号量异常", e);}return flag;}/*** 释放信号量* * @param semaphore*/public static void release(Semaphore semaphore){try{// 释放信号量，增加可用的许可数semaphore.release();}catch (Exception e){// 捕获异常并记录日志LOGGER.error("释放信号量异常", e);}}
}

解释每个部分的功能：

1. tryAcquire(Semaphore semaphore) 方法：该方法尝试从信号量中获取一个许可。如果成功获取许可，返回 true；如果无法获取许可，返回 false。在获取信号量时，可能会抛出异常，例如在等待获取信号量的过程中发生中断或超时，因此在方法内捕获异常并记录日志。

2. release(Semaphore semaphore) 方法：该方法释放一个许可到信号量中，增加信号量的可用许可数。同样，可能会在释放信号量时发生异常，捕获异常并记录日志。

这样的工具类通常用于协调多个线程对共享资源的访问，通过信号量可以限制同时访问某个资源的线程数量，以避免竞争和提高系统的稳定性。在这个类中，异常处理的目的是确保即使在获取或释放信号量的过程中发生异常，也能够进行适当的处理并记录相关信息。

然后呢，我们来编写一个websocket的配置文件，内容很简单，如下：

package com.ruoyi.framework.websocket;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** websocket 配置* * @author ruoyi*/
@Configuration
public class WebSocketConfig
{@Beanpublic ServerEndpointExporter serverEndpointExporter(){return new ServerEndpointExporter();}
/************************************************自己增加的内容*******************************************************************/@Beanpublic ServletServerContainerFactoryBean createWebSocketContainer() {ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();// 在此处设置bufferSizecontainer.setMaxTextMessageBufferSize(5120000);container.setMaxBinaryMessageBufferSize(5120000);container.setMaxSessionIdleTimeout(15 * 60000L);return container;}
}

这是一个WebSocket配置类，用于配置WebSocket相关的参数和组件。以下是对每个部分的详细介绍：

1. @Configuration 注解:

   • 表示这是一个Spring配置类，用于配置WebSocket相关的组件。

2. @Bean 方法 - serverEndpointExporter:

   • 创建并返回一个ServerEndpointExporter对象。
   • ServerEndpointExporter是Spring提供的用于注册和管理WebSocket端点的Bean，它会在Spring容器中查找所有使用@ServerEndpoint注解的WebSocket端点并注册它们。
   • 这个Bean的存在使得WebSocket端点能够被正确地注册并且可以被使用。

3. @Bean 方法 - createWebSocketContainer:

   • 创建并返回一个ServletServerContainerFactoryBean对象。
   • ServletServerContainerFactoryBean是Spring提供的用于配置WebSocket容器的工厂Bean。通过配置它，可以设置WebSocket容器的一些参数。
   • 在这个方法中，设置了以下WebSocket容器的参数：
     • setMaxTextMessageBufferSize(5120000): 设置文本消息的最大缓冲区大小为5 MB。
     • setMaxBinaryMessageBufferSize(5120000): 设置二进制消息的最大缓冲区大小为5 MB。
     • setMaxSessionIdleTimeout(15 * 60000L): 设置会话的最大空闲时间为15分钟。

简而言之，这个WebSocket配置类使用Spring的Java配置方式，通过@Bean注解创建了两个Bean，分别是ServerEndpointExporter和ServletServerContainerFactoryBean。前者用于注册WebSocket端点，后者用于配置WebSocket容器的一些参数。这样，你的Spring Boot应用就能够正确地支持WebSocket了。

然后呢，就到了最主要的内容，我们需要写一个websocket服务类，里面有众多方法

package com.ruoyi.framework.websocket;import java.util.concurrent.Semaphore;
import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;/*** websocket 消息处理* * @author ruoyi*/
@Component
@ServerEndpoint("/websocket/message")
public class WebSocketServer
{/*** WebSocketServer 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketServer.class);/*** 默认最多允许同时在线人数100*/public static int socketMaxOnlineCount = 100;private static Semaphore socketSemaphore = new Semaphore(socketMaxOnlineCount);/*** 连接建立成功调用的方法*/@OnOpenpublic void onOpen(Session session) throws Exception{boolean semaphoreFlag = false;// 尝试获取信号量semaphoreFlag = SemaphoreUtils.tryAcquire(socketSemaphore);if (!semaphoreFlag){// 未获取到信号量LOGGER.error("\n 当前在线人数超过限制数- {}", socketMaxOnlineCount);WebSocketUsers.sendMessageToUserByText(session, "当前在线人数超过限制数：" + socketMaxOnlineCount);session.close();}else{// 添加用户WebSocketUsers.put(session.getId(), session);LOGGER.info("\n 建立连接 - {}", session);LOGGER.info("\n 当前人数 - {}", WebSocketUsers.getUsers().size());WebSocketUsers.sendMessageToUserByText(session, "连接成功");}}/*** 连接关闭时处理*/@OnClosepublic void onClose(Session session){LOGGER.info("\n 关闭连接 - {}", session);// 移除用户WebSocketUsers.remove(session.getId());// 获取到信号量则需释放SemaphoreUtils.release(socketSemaphore);}/*** 抛出异常时处理*/@OnErrorpublic void onError(Session session, Throwable exception) throws Exception{if (session.isOpen()){// 关闭连接session.close();}String sessionId = session.getId();LOGGER.info("\n 连接异常 - {}", sessionId);LOGGER.info("\n 异常信息 - {}", exception);// 移出用户WebSocketUsers.remove(sessionId);// 获取到信号量则需释放SemaphoreUtils.release(socketSemaphore);}/*** 服务器接收到客户端消息时调用的方法*/@OnMessagepublic void onMessage(String message, Session session){String msg = message.replace("你", "我").replace("吗", "");WebSocketUsers.sendMessageToUserByText(session, msg);}
}

这段代码是一个WebSocket服务器端的实现，用于处理与客户端的WebSocket通信。让我们逐步解释其关键部分：

1. @Component 注解:

   • 表示将该类作为Spring组件进行管理，可以通过Spring的扫描机制自动识别并注册为bean。

2. @ServerEndpoint("/websocket/message") 注解:

   • 将该类标记为WebSocket的端点，客户端可以通过访问 “/websocket/message” 路径来与服务器建立WebSocket连接，我们如果需要连接WebSocket，那么连接地址就是：ws:localhost:端口号/websocket/message，类似于http请求！

3. @OnOpen 注解:

   • 标注一个方法，在WebSocket连接建立时触发。在这里，它用于处理连接建立成功后的逻辑。
   • 尝试获取信号量，如果成功，表示连接建立成功，会将该连接加入到用户列表中；否则，如果在线人数超过限制，会发送错误信息并关闭连接。

4. @OnClose 注解:

   • 标注一个方法，在WebSocket连接关闭时触发。在这里，它用于处理连接关闭后的清理逻辑，包括移除用户并释放信号量。

5. @OnError 注解:

   • 标注一个方法，在WebSocket发生异常时触发。在这里，它用于处理异常情况，关闭连接并进行相应的清理工作。

6. @OnMessage 注解:

   • 标注一个方法，在服务器接收到客户端消息时触发。在这里，它将接收到的消息进行处理，例如将消息中的 “你” 替换为 “我”，去掉 “吗”，然后通过 WebSocketUsers.sendMessageToUserByText 方法将处理后的消息发送回客户端。

总体而言，这个类实现了WebSocket服务器的基本功能，包括连接的建立、关闭、异常处理以及消息的处理。同时，它通过信号量控制了最大允许同时在线人数，确保连接数不超过设定的限制。

代码也很简单，大家在网上也会找到类似的工具类，只不过若依是多加了一些封装

最后，就到了服务升级了，这也是若依特有的，我们一起看看：

package com.ruoyi.framework.websocket;import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
import javax.websocket.Session;import com.alibaba.fastjson2.JSON;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;/*** websocket 客户端用户集* * @author ruoyi*/
public class WebSocketUsers
{/*** WebSocketUsers 日志控制器*/private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketUsers.class);/*** 用户集*/private static Map<String, Session> USERS = new ConcurrentHashMap<String, Session>();/*** 存储用户** @param key 唯一键* @param session 用户信息*/public static void put(String key, Session session){USERS.put(key, session);}/*** 移除用户** @param session 用户信息** @return 移除结果*/public static boolean remove(Session session){String key = null;boolean flag = USERS.containsValue(session);if (flag){Set<Map.Entry<String, Session>> entries = USERS.entrySet();for (Map.Entry<String, Session> entry : entries){Session value = entry.getValue();if (value.equals(session)){key = entry.getKey();break;}}}else{return true;}return remove(key);}/*** 移出用户** @param key 键*/public static boolean remove(String key){LOGGER.info("\n 正在移出用户 - {}", key);Session remove = USERS.remove(key);if (remove != null){boolean containsValue = USERS.containsValue(remove);LOGGER.info("\n 移出结果 - {}", containsValue ? "失败" : "成功");return containsValue;}else{return true;}}/*** 获取在线用户列表** @return 返回用户集合*/public static Map<String, Session> getUsers(){return USERS;}/*** 群发消息文本消息** @param message 消息内容*/public static void sendMessageToUsersByText(String message){Collection<Session> values = USERS.values();for (Session value : values){sendMessageToUserByText(value, message);}}/*** 发送文本消息** @param userName 自己的用户名* @param message 消息内容*/public static void sendMessageToUserByText(Session session, String message){if (session != null){try{session.getBasicRemote().sendText(message);}catch (IOException e){LOGGER.error("\n[发送消息异常]", e);}}else{LOGGER.info("\n[你已离线]");}}
/************************************************自己增加的内容***************************************************************//*** 群发消息文本消息** @param messages 消息内容*/public static void sendMessageToUsers(List<String> messages, Collection<Session> values) {for (String message : messages) {for (Session value : values) {sendMessageToUserByText(value, message);}}}public static void sendMessageToUsersMap(List<Map<String, Object>> li, Collection<Session> values) {for (Map<String, Object> map : li) {for (Session value : values) {sendMessageToUserByText(value, JSON.toJSONString(map));}}}/*** 获取指定用户** @param userIds 用户id* @return*/public static Collection<Session> getSessionUserMap(List<Long> userIds) {Map<String, Session> sessionUserMap = new HashMap<>();if (CollectionUtils.isNotEmpty(userIds)) {List<String> newUserIds = userIds.stream().map(String::valueOf).collect(Collectors.toList());Map<String, Session> users = USERS;for (Iterator<Map.Entry<String, Session>> it = users.entrySet().iterator(); it.hasNext(); ) {Map.Entry<String, Session> item = it.next();String key = item.getKey();// 发送用户处理if (newUserIds.contains(key.substring(0, key.indexOf("_")))) {sessionUserMap.put(key, item.getValue());}}}return sessionUserMap.values();}/*** 给用户发消息** @param message 消息内容* @param userId  用户id*/public static void sendMessageToUser(String message, Long userId) {List<Long> userList = new ArrayList<>();List<String> msgs = new ArrayList<>();// 添加发送对象userList.add(userId);// 添加发送内容msgs.add(message);Collection<Session> sessionUserMap = WebSocketUsers.getSessionUserMap(userList);WebSocketUsers.sendMessageToUsers(msgs, sessionUserMap);}
}

这段代码定义了一个用于管理WebSocket用户的工具类 WebSocketUsers。

1. put 方法:

   • 用于将用户信息存储到USERS集合中，其中key是用户的唯一标识，session是与用户连接关联的Session对象。

2. remove 方法 (两个重载):

   • 从USERS集合中移除用户。一个版本是通过Session对象移除，另一个版本是通过用户的唯一键移除。
   • 首先检查是否包含给定的Session，然后根据情况找到对应的键，最后移除用户。

3. getUsers 方法:

   • 返回当前在线用户的集合，即USERS集合。

4. sendMessageToUsersByText 方法:

   • 用于向所有在线用户发送文本消息。遍历USERS集合中的每个用户的Session对象，并调用sendMessageToUserByText方法发送消息。

5. sendMessageToUserByText 方法:

   • 向指定用户的Session发送文本消息。
   • 如果session不为null，使用session.getBasicRemote().sendText(message)发送文本消息；否则，记录用户已离线的信息。

6. sendMessageToUsers 方法:

   • 向一组用户发送消息，接收一个包含消息的列表和一个包含Session对象的集合。
   • 遍历消息列表和Session集合，为每个用户的Session对象发送相应的消息。

7. sendMessageToUsersMap 方法:

   • 向一组用户发送包含Map<String, Object>消息的列表。
   • 将每个Map对象转换为JSON格式的字符串，并使用sendMessageToUserByText方法发送给每个用户。

8. getSessionUserMap 方法:

   • 根据给定的用户id列表，返回相应用户的Session对象集合。
   • 使用用户id列表过滤出匹配的Session对象，并将其存储在sessionUserMap中返回。

9. sendMessageToUser 方法:

   • 向指定用户发送消息，接收消息内容和目标用户的id。
   • 创建包含目标用户id和消息内容的列表，然后通过getSessionUserMap方法获取匹配的Session对象集合，并使用sendMessageToUsers方法发送消息。

这些方法共同构成了一个WebSocket用户管理类，用于存储用户信息、发送消息，以及一些特定需求的消息发送，如向指定用户发送消息、向特定用户组发送消息等。

这样，我们就大概完成了整个代码逻辑，也不是非常复杂！

然后，我们就需要进行测试一波----------

ApiPost测试

在线工具测试 https://wstool.js.org/

控制台日志

然后，大家会问，怎么实现http功能加成，其实大家可以看到在WebSocketUsers有需要业务层的处理方法，其实这些就是供http请求调用的

如：

这就是专门提供给http调用的，实现给指定用户发送消息，如消息通知，或者是公告，审批消息提醒等————
完毕，感谢大家阅读！

所谓好运，所谓幸福，显然不是一种客观的程序，而是完全心灵的感受，是强烈的幸福感罢了。
——史铁生