Spring Boot整合T-IO实现即时通讯

王柳  论坛元老 | 2025-4-18 11:27:58 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 1890|帖子 1890|积分 5670

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要 登录 才可以下载或查看，没有账号？立即注册

x
在当今的互联网应用中，即时通讯功能已经变得不可或缺。无论是社交应用、在线客服还是实时数据推送，都必要高效的通讯框架来支持。TIO（Try-IO）是一个高性能的Java网络通讯框架，支持TCP/UDP/HTTP/WebSocket等多种协议，非常适合用于即时通讯场景。本文将介绍如何在Spring Boot项目中整合TIO，实现一个简朴的即时通讯功能。
1. TIO简介

TIO是一个基于Java的高性能网络通讯框架，具有以下特点：
支持多种协议：TCP、UDP、HTTP、WebSocket等。
高性能：基于NIO实现，支持高并发。
易用性：提供了丰富的API和机动的配置
2. 情况准备

在开始之前，请确保你已经安装了以下工具：
JDK 1.8及以上版本（本项目利用jdk 17版本）
Maven或Gradle
IntelliJ IDEA或Eclipse
4.添加依赖

在pom.xml文件中添加TIO的相关依赖
  1.         <!-- t-io WebSocket依赖 -->
  2.         <dependency>
  3.             <groupId>org.t-io</groupId>
  4.             <artifactId>tio-websocket-server</artifactId>
  5.             <version>3.8.6.v20240801-RELEASE</version>
  6.         </dependency>
  8.         <!-- t-io 核心依赖 -->
  9.         <dependency>
  10.             <groupId>org.t-io</groupId>
  11.             <artifactId>tio-core</artifactId>
  12.             <version>3.8.6.v20240801-RELEASE</version>
  13.         </dependency>
复制代码
5、创建TIO消息处置惩罚器

创建一个实现IWsMsgHandler接口的类，用于处置惩罚WebSocket消息：
  1. import com.alibaba.fastjson.JSON;
  2. import com.ruoyi.im.config.TioWebSocketAdapterConfig;
  3. import com.ruoyi.im.domain.ImMessage;
  4. import com.ruoyi.im.model.ChatMessage;
  5. import com.ruoyi.im.service.IImMessageService;
  6. import org.slf4j.Logger;
  7. import org.slf4j.LoggerFactory;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.stereotype.Component;
  10. import org.tio.core.ChannelContext;
  11. import org.tio.http.common.HttpRequest;
  12. import org.tio.http.common.HttpResponse;
  13. import org.tio.websocket.common.WsRequest;
  14. import org.tio.websocket.common.WsResponse;
  15. import org.tio.websocket.server.handler.IWsMsgHandler;
  17. import java.util.Date;
  18. import java.util.UUID;
  20. @Component
  21. public class WebSocketMessageHandler implements IWsMsgHandler {
  22.    
  23.     private static final Logger log = LoggerFactory.getLogger(WebSocketMessageHandler.class);
  24.    
  25.     /**
  26.      * 从ChannelContext中获取参数
  27.      */
  28.     private String getParam(ChannelContext channelContext, String paramName) {
  29.         try {
  30.             // 尝试从channelContext的httpConfig或其他属性中获取参数
  31.             Object requestObj = channelContext.getAttribute("httpRequest");
  32.             if (requestObj != null && requestObj instanceof HttpRequest) {
  33.                 HttpRequest httpRequest = (HttpRequest) requestObj;
  34.                 return httpRequest.getParam(paramName);
  35.             }
  36.             
  37.             // 从user属性中获取用户ID
  38.             if ("userId".equals(paramName)) {
  39.                 return (String) channelContext.getAttribute("userId");
  40.             }
  41.             
  42.             return null;
  43.         } catch (Exception e) {
  44.             log.error("获取参数失败: paramName={}", paramName, e);
  45.             return null;
  46.         }
  47.     }
  48.    
  49.     @Override
  50.     public HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) {
  51.         // 从请求中获取用户ID、token和设备类型
  52.         String userId = httpRequest.getParam("userId");
  53.         String token = httpRequest.getParam("token");
  54.         String deviceTypeStr = httpRequest.getParam("deviceType");
  55.         
  56.         // 验证参数
  57.         if (userId == null || token == null || deviceTypeStr == null) {
  58.             log.error("握手参数不完整: userId={}, token={}, deviceType={}", userId, token, deviceTypeStr);
  59.             return httpResponse;
  60.         }
  61.         
  62.         // 验证token（这里应该调用若依的token验证服务）
  63.         // TODO: 实现token验证逻辑
  64.         
  65.         // 解析设备类型
  66.         UserSessionManager.DeviceType deviceType;
  67.         try {
  68.             deviceType = UserSessionManager.DeviceType.valueOf(deviceTypeStr.toUpperCase());
  69.         } catch (IllegalArgumentException e) {
  70.             log.error("无效的设备类型: {}", deviceTypeStr);
  71.             return httpResponse;
  72.         }
  73.         
  74.         // 保存参数到ChannelContext
  75.         channelContext.setAttribute("httpRequest", httpRequest);
  76.         channelContext.setAttribute("userId", userId);
  77.         
  78.         // 添加用户会话
  79.         userSessionManager.addUserSession(userId, token, deviceType, channelContext);
  80.         
  81.         return httpResponse;
  82.     }
  83.    
  84.     @Override
  85.     public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) {
  86.         String userId = httpRequest.getParam("userId");
  87.         log.info("握手成功，userId: {}, channelId: {}", userId, channelContext.getId());
  88.     }
  89.    
  90.     @Override
  91.     public Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) {
  92.         return null;
  93.     }
  94.    
  95.     @Override
  96.     public Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) {
  97.         // 从channelContext获取用户ID
  98.         String userId = getParam(channelContext, "userId");
  99.         // 移除用户会话
  100.         if (userId != null) {
  101.             userSessionManager.removeUserSession(userId);
  102.         }
  103.         log.info("连接关闭，userId: {}, channelId: {}", userId, channelContext.getId());
  104.         return null;
  105.     }
  106.    
  107.     @Override
  108.     public Object onText(WsRequest wsRequest, String msg, ChannelContext channelContext) {
  109.         // 从channelContext获取用户ID
  110.         String userId = getParam(channelContext, "userId");
  111.         log.info("收到消息: {}, userId: {}, channelId: {}", msg, userId, channelContext.getId());
  112.         
  113.         try {
  114.             // 解析消息
  115.             ChatMessage chatMessage = JSON.parseObject(msg, ChatMessage.class);
  116.             
  117.             // 设置消息ID和发送时间
  118.             if (chatMessage.getMessageId() == null) {
  119.                 chatMessage.setMessageId(UUID.randomUUID().toString());
  120.             }
  121.             if (chatMessage.getSendTime() == null) {
  122.                 chatMessage.setSendTime(new Date());
  123.             }
  124.             
  125.             // 设置发送者ID
  126.             chatMessage.setFromUserId(userId);
  127.             
  128.             // 更新用户最后活动时间
  129.             userSessionManager.updateUserLastActiveTime(userId);
  130.             
  131.             // 创建消息实体并持久化
  132.             ImMessage imMessage = new ImMessage();
  133.             imMessage.setMessageId(chatMessage.getMessageId());
  134.             imMessage.setFromUserId(chatMessage.getFromUserId());
  135.             imMessage.setToUserId(chatMessage.getToUserId());
  136.             imMessage.setGroupId(chatMessage.getGroupId());
  137.             imMessage.setContent(chatMessage.getContent());
  138.             imMessage.setMessageType(chatMessage.getMessageType());
  139.             imMessage.setStatus(0); // 0表示未读
  140.             imMessage.setSendTime(chatMessage.getSendTime());
  141.             imMessage.setCreateTime(chatMessage.getSendTime());
  142.             imMessage.setUpdateTime(chatMessage.getSendTime());
  143.             imMessage.setIsSent(1); // 1表示已发送
  144.             imMessage.setRetryCount(0); // 初始重试次数为0
  145.             
  146.             // 保存消息到数据库
  147.             imMessageService.insertImMessage(imMessage);
  148.             
  149.             // 处理消息
  150.             if (chatMessage.getToUserId() != null) {
  151.                 // 私聊消息
  152.                 userSessionManager.sendMessageToUser(chatMessage.getToUserId(), JSON.toJSONString(chatMessage));
  153.             } else if (chatMessage.getGroupId() != null) {
  154.                 // 群聊消息（需要实现群组管理）
  155.                 // TODO: 实现群聊消息处理
  156.             } else {
  157.                 // 广播消息
  158.                 userSessionManager.broadcastMessage(JSON.toJSONString(chatMessage));
  159.             }
  160.             
  161.             // 发送响应消息
  162.             WsResponse wsResponse = WsResponse.fromText("SendMessageSuccess：" + msg, "UTF-8");
  163.             TioWebSocketAdapterConfig.send(channelContext, wsResponse);
  164.             
  165.         } catch (Exception e) {
  166.             log.error("处理消息失败: {}", msg, e);
  167.             WsResponse wsResponse = WsResponse.fromText("处理消息失败: " + e.getMessage(), "UTF-8");
  168.             TioWebSocketAdapterConfig.send(channelContext, wsResponse);
  169.         }
  170.         
  171.         return null;
  172.     }
  173. }
复制代码
  1. import com.ruoyi.im.config.TioWebSocketAdapterConfig;
  2. import org.slf4j.Logger;
  3. import org.slf4j.LoggerFactory;
  4. import org.springframework.stereotype.Component;
  5. import org.tio.core.ChannelContext;
  6. import org.tio.websocket.common.WsResponse;
  8. import java.util.Map;
  9. import java.util.Set;
  10. import java.util.concurrent.ConcurrentHashMap;
  12. @Component
  13. public class UserSessionManager {
  14.    
  15.     private static final Logger log = LoggerFactory.getLogger(UserSessionManager.class);
  16.    
  17.     // 用户ID -> 用户会话信息
  18.     private final Map<String, UserSession> userSessions = new ConcurrentHashMap<>();
  19.    
  20.     // 设备类型枚举
  21.     public enum DeviceType {
  22.         WEB,    // Web端
  23.         MINI,   // 小程序
  24.         APP     // APP
  25.     }
  26.    
  27.     // 用户会话信息类
  28.     public static class UserSession {
  29.         private String userId;
  30.         private String token;
  31.         private DeviceType deviceType;
  32.         private ChannelContext channelContext;
  33.         private long lastActiveTime;
  34.         
  35.         public UserSession(String userId, String token, DeviceType deviceType, ChannelContext channelContext) {
  36.             this.userId = userId;
  37.             this.token = token;
  38.             this.deviceType = deviceType;
  39.             this.channelContext = channelContext;
  40.             this.lastActiveTime = System.currentTimeMillis();
  41.         }
  42.         
  43.         // Getters and setters
  44.         public String getUserId() { return userId; }
  45.         public void setUserId(String userId) { this.userId = userId; }
  46.         public String getToken() { return token; }
  47.         public void setToken(String token) { this.token = token; }
  48.         public DeviceType getDeviceType() { return deviceType; }
  49.         public void setDeviceType(DeviceType deviceType) { this.deviceType = deviceType; }
  50.         public ChannelContext getChannelContext() { return channelContext; }
  51.         public void setChannelContext(ChannelContext channelContext) { this.channelContext = channelContext; }
  52.         public long getLastActiveTime() { return lastActiveTime; }
  53.         public void setLastActiveTime(long lastActiveTime) { this.lastActiveTime = lastActiveTime; }
  54.     }
  55.    
  56.     // 添加用户会话
  57.     public void addUserSession(String userId, String token, DeviceType deviceType, ChannelContext channelContext) {
  58.         UserSession session = new UserSession(userId, token, deviceType, channelContext);
  59.         userSessions.put(userId, session);
  60.         
  61.         // 绑定用户ID到ChannelContext，使用适配器
  62.         TioWebSocketAdapterConfig.bindUser(channelContext, userId);
  63.         
  64.         log.info("用户会话添加成功: userId={}, deviceType={}", userId, deviceType);
  65.     }
  66.    
  67.     // 移除用户会话
  68.     public void removeUserSession(String userId) {
  69.         UserSession session = userSessions.remove(userId);
  70.         if (session != null) {
  71.             // 解绑用户ID，使用适配器
  72.             TioWebSocketAdapterConfig.unbindUser(session.getChannelContext(), userId);
  73.             log.info("用户会话移除成功: userId={}", userId);
  74.         }
  75.     }
  76.    
  77.     // 获取用户会话
  78.     public UserSession getUserSession(String userId) {
  79.         return userSessions.get(userId);
  80.     }
  81.    
  82.     // 更新用户最后活动时间
  83.     public void updateUserLastActiveTime(String userId) {
  84.         UserSession session = userSessions.get(userId);
  85.         if (session != null) {
  86.             session.setLastActiveTime(System.currentTimeMillis());
  87.         }
  88.     }
  89.    
  90.     // 发送消息给指定用户
  91.     public void sendMessageToUser(String userId, String message) {
  92.         UserSession session = userSessions.get(userId);
  93.         if (session != null && session.getChannelContext() != null) {
  94.             WsResponse wsResponse = WsResponse.fromText(message, "UTF-8");
  95.             // 使用适配器发送消息
  96.             TioWebSocketAdapterConfig.send(session.getChannelContext(), wsResponse);
  97.         } else {
  98.             log.warn("用户不在线，无法发送消息: userId={}", userId);
  99.         }
  100.     }
  101.    
  102.     // 广播消息给所有用户
  103.     public void broadcastMessage(String message) {
  104.         if (userSessions.isEmpty()) {
  105.             return;
  106.         }
  107.         
  108.         WsResponse wsResponse = WsResponse.fromText(message, "UTF-8");
  109.         for (UserSession session : userSessions.values()) {
  110.             if (session.getChannelContext() != null) {
  111.                 // 使用适配器发送消息
  112.                 TioWebSocketAdapterConfig.send(session.getChannelContext(), wsResponse);
  113.             }
  114.         }
  115.     }
  116.    
  117.     // 获取在线用户数量
  118.     public int getOnlineUserCount() {
  119.         return userSessions.size();
  120.     }
  121.    
  122.     // 获取所有在线用户ID
  123.     public Set<String> getOnlineUserIds() {
  124.         return userSessions.keySet();
  125.     }
  126. }
复制代码
6、创建TIO配置类和 TIO WebSocket适配器配置

  1. import com.ruoyi.im.websocket.WebSocketMessageHandler;
  2. import org.springframework.beans.factory.annotation.Autowired;
  3. import org.springframework.context.annotation.Bean;
  4. import org.springframework.context.annotation.Configuration;
  5. import org.tio.websocket.server.WsServerConfig;
  6. import org.tio.websocket.server.WsServerStarter;
  8. import java.io.IOException;
  10. @Configuration
  11. public class TioWebSocketConfig {
  12.    
  13.     @Autowired
  14.     private WebSocketMessageHandler webSocketMessageHandler;
  15.    
  16.     @Bean
  17.     public WsServerStarter wsServerStarter() throws IOException {
  18.         // 配置t-io websocket服务器，指定端口
  19.         WsServerConfig wsServerConfig = new WsServerConfig(9321);
  20.         
  21.         // 创建WebSocket服务器
  22.         WsServerStarter wsServerStarter = new WsServerStarter(wsServerConfig, webSocketMessageHandler);
  23.         
  24.         // 这里不再获取和配置ServerTioConfig
  25.         
  26.         return wsServerStarter;
  27.     }
  28. }
复制代码
  1. import org.slf4j.Logger;
  2. import org.slf4j.LoggerFactory;
  3. import org.springframework.context.annotation.Configuration;
  4. import org.tio.core.ChannelContext;
  5. import org.tio.core.Tio;
  6. import org.tio.core.intf.Packet;
  8. /**
  9. * t-io WebSocket适配器配置
  10. * 帮助处理不同版本的t-io兼容性问题
  11. */
  12. @Configuration
  13. public class TioWebSocketAdapterConfig {
  14.    
  15.     private static final Logger log = LoggerFactory.getLogger(TioWebSocketAdapterConfig.class);
  17.     /**
  18.      * 发送消息
  19.      *
  20.      * @param channelContext 通道上下文
  21.      * @param packet 数据包
  22.      */
  23.     public static void send(ChannelContext channelContext, Packet packet) {
  24.         if (channelContext == null || packet == null) {
  25.             log.warn("发送消息失败：通道上下文或数据包为空");
  26.             return;
  27.         }
  29.         try {
  30.             // 直接调用 Tio 的 send 方法
  31.             Tio.send(channelContext, packet);
  32.         } catch (Exception e) {
  33.             log.error("发送消息失败", e);
  34.         }
  35.     }
  38.     /**
  39.      * 绑定用户
  40.      *
  41.      * @param channelContext 通道上下文
  42.      * @param userId 用户ID
  43.      */
  44.     public static void bindUser(ChannelContext channelContext, String userId) {
  45.         if (channelContext == null || userId == null) {
  46.             log.warn("绑定用户失败：通道上下文或用户ID为空");
  47.             return;
  48.         }
  50.         try {
  51.             // 直接调用 Tio 的 bindUser 方法
  52.             Tio.bindUser(channelContext, userId);
  53.         } catch (Exception e) {
  54.             log.error("绑定用户失败", e);
  55.         }
  56.     }
  58.     /**
  59.      * 解绑用户
  60.      *
  61.      * @param channelContext 通道上下文
  62.      * @param userId 用户ID
  63.      */
  64.     public static void unbindUser(ChannelContext channelContext, String userId) {
  65.         if (channelContext == null) {
  66.             log.warn("解绑用户失败：通道上下文为空");
  67.             return;
  68.         }
  70.         try {
  71.             // 直接调用 Tio 的 unbindUser 方法
  72.             Tio.unbindUser(channelContext);
  73.         } catch (Exception e) {
  74.             log.error("解绑用户失败", e);
  75.         }
  76.     }
  77. }
复制代码
7、配置TIO启动类

  1. import org.slf4j.Logger;
  2. import org.slf4j.LoggerFactory;
  3. import org.springframework.beans.factory.annotation.Autowired;
  4. import org.springframework.stereotype.Component;
  5. import org.tio.websocket.server.WsServerStarter;
  7. import javax.annotation.PostConstruct;
  8. import javax.annotation.PreDestroy;
  9. import java.io.IOException;
  10. import java.lang.reflect.Method;
  12. @Component
  13. public class WebSocketServer {
  14.    
  15.     private static final Logger log = LoggerFactory.getLogger(WebSocketServer.class);
  16.    
  17.     @PostConstruct
  18.     public void start() {
  19.         try {
  20.             wsServerStarter.start();
  21.             log.info("WebSocket服务器启动成功，监听端口：9321");
  22.         } catch (IOException e) {
  23.             log.error("WebSocket服务器启动失败", e);
  24.         }
  25.     }
  26.    
复制代码
启动springboot主类，即可启动tio服务
  1.   public static void main(String[] args) {
  2.         SpringApplication.run(RuoYiApplication.class, args);
  3.     }
复制代码
8、测试即时通讯功能

可以利用websocket工具进行测试
  1. ##链接地址：
  2. ws://127.0.0.1:9321?userId=1&token=1&deviceType=MINI
  4. ##链接参数说明：
  5. userId:当前用户id
  6. token:token值
  7. deviceType:设备类型  WEB Web端、MINI 小程序 、APP
  9. ##发送消息示例：
  10. {
  11.     "fromUserId": "1",发送者用户ID
  12.     "toUserId": "2",//接收者用户ID
  13.     "content": "测试消息内容",//消息内容
  14.     "messageType": "1"//消息类型1=文本,2=图片,3=语音,4=视频
  15. }
  17. ##发送后响应示例：
  18. SendMessageSuccess：{ "fromUserId": "1", "toUserId": "2", "content": "测试消息内容", "messageType": "1" }
复制代码
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

王柳

论坛元老
这个人很懒什么都没写!

楼主热帖

标签云

AI 运维 CIO 存储 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表