IT评测·应用市场-qidao123.com技术社区

标题: Spring Boot整合T-IO实现即时通讯 [打印本页]

---

作者: 王柳    时间: 2025-4-18 11:27
标题: Spring Boot整合T-IO实现即时通讯
在当今的互联网应用中，即时通讯功能已经变得不可或缺。无论是社交应用、在线客服还是实时数据推送，都必要高效的通讯框架来支持。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企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 IT评测·应用市场-qidao123.com技术社区 (https://dis.qidao123.com/)

Powered by Discuz! X3.4