backend/java-mall
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

im调整组的sesion,代码优化

Browse Source
This commit is contained in:
liyj 2025-11-20 10:35:14 +08:00
parent 0b6c87aa31
commit 47331a5b2a
4 changed files with 25 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
mall-im/src/main/java/com/suisung/mall/im/common/websocket/service/DistributedMessageService.java
View File

@ -36,10 +36,9 @@ public class DistributedMessageService {
* 发送消息给用户
* @param targetUserId 目标用户就是account的id
* @param message
* @param userSessions
* @return
*/
public boolean sendToUser(String targetUserId, SendVO message, ConcurrentHashMap<String, List<WebSocketSession>> userSessions) {
public boolean sendToUser(String targetUserId, SendVO message) {
log.info("targetUserId{},message:{}", targetUserId, message);
String targetServer = sessionService.getUserServer(targetUserId);
@ -50,7 +49,7 @@ public class DistributedMessageService {
if (isCurrentServer(targetServer)) {
// 用户在当前服务器，直接发送
return localSessionManager.sendToUser(targetUserId, message,userSessions);
return localSessionManager.sendToUser(targetUserId, message);
} else {
// 用户在其他服务器，通过RabbitMQ转发
forwardToUser(targetServer, targetUserId, message);

19
mall-im/src/main/java/com/suisung/mall/im/common/websocket/service/LocalSessionManager.java
View File

@ -27,10 +27,11 @@ public class LocalSessionManager {
private ConcurrentHashMap<String, List<WebSocketSession>> userSessions;
// 本地存储的群组会话 (groupId -> sessions)
private ConcurrentHashMap<String, List<WebSocketSession>> groupSessions = new ConcurrentHashMap<>();
private ConcurrentHashMap<String, List<WebSocketSession>> groupSessions;
public LocalSessionManager() {
this.userSessions = MallsuiteImSocketHandler.userSessions;
this.groupSessions = MallsuiteImSocketHandler.groupSessions;
}
/**
@ -51,18 +52,14 @@ public class LocalSessionManager {
* 发送消息给用户
* @param userId account的id
* @param message
* @param userSession
* @return
*/
public boolean sendToUser(String userId, SendVO message,ConcurrentHashMap<String, List<WebSocketSession>> userSession) {
public boolean sendToUser(String userId, SendVO message) {
if (userId == null || message == null) {
log.info("发送消息失败: 参数为空");
return false;
}
if(null!=userSession){
this.userSessions=userSession;
}
log.info("userSessions={}", userSession);
log.info("userSessions={}", userSessions);
List<WebSocketSession> sessions = userSessions.get(userId);
if (sessions == null || sessions.isEmpty()) {
log.info("用户没有活跃会话: {}", userId);
@ -111,14 +108,6 @@ public class LocalSessionManager {
return false;
}
/**
* 添加群组会话
*/
public void addGroupSession(String groupId, WebSocketSession session) {
groupSessions.computeIfAbsent(groupId, k -> new CopyOnWriteArrayList<>())
.add(session);
}
/**
* 移除群组会话
*/

5
mall-im/src/main/java/com/suisung/mall/im/common/websocket/service/MessageConsumerService.java
View File

@ -26,8 +26,7 @@ public class MessageConsumerService {
public void consumeForwardMessage(Map<String, Object> message) {
try {
String type = (String) message.get("type");
System.out.println("收到RabbitMQ消息, 类型: " + type + ", 服务器: " + rabbitMQManager.getServerId());
logger.info("收到RabbitMQ消息, 类型: {}, 服务器: {}", type, rabbitMQManager.getServerId());
switch (type) {
case "SEND_TO_USER":
handleSendToUser(message);
@ -65,7 +64,7 @@ public class MessageConsumerService {
}
if (userMsg != null) {
boolean success = localSessionManager.sendToUser(userId, userMsg,null);
boolean success = localSessionManager.sendToUser(userId, userMsg);
logger.info("发送消息给用户 {}: {}", userId, success ? "成功" : "失败");
} else {
logger.info("消息格式错误，无法发送给用户: {}", userId);

22
mall-im/src/main/java/com/suisung/mall/im/common/websocket/service/onchat/MallsuiteImSocketHandler.java
View File

@ -36,6 +36,8 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
public static ConcurrentHashMap<String, List<WebSocketSession>> userSessions;
public static ConcurrentHashMap<String, List<WebSocketSession>> groupSessions;
@Autowired
private DistributedSessionService distributedSessionService;
@ -50,6 +52,7 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
static {
userSessions=new ConcurrentHashMap<>();
groupSessions=new ConcurrentHashMap<>();
}
//用户上线后触发
@ -134,7 +137,7 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
chat.setMsg(msg);
chat.setType("friend");
chat.setCreate_date(new Date());
boolean isSuccess = distributedMessageService.sendToUser(String.valueOf(receiveDTO.getTo().getFriend_id()), sendVO,localSessionManager.getUserSessions());
boolean isSuccess = distributedMessageService.sendToUser(String.valueOf(receiveDTO.getTo().getFriend_id()), sendVO);
if (isSuccess) {
chat.setStatus("1");
} else {
@ -144,7 +147,7 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
sendVO.setMsg_type("text");
sendVO.setMessage_length("0");
sendVO.setId(to.getId());
distributedMessageService.sendToUser(String.valueOf(receiveDTO.getTo().getFriend_id()), sendVO,localSessionManager.getUserSessions());
distributedMessageService.sendToUser(String.valueOf(receiveDTO.getTo().getFriend_id()), sendVO);
chat.setStatus("0");
}
chatHistoryService.saveNew(chat);
@ -160,7 +163,7 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
session.getAttributes().put(Constants.WEBSOCKET_GROUP_KEY, zid);
// 添加到本地群组
localSessionManager.addGroupSession(zid, session);
this.addGroupSession(zid, session);
// 添加到分布式群组
String userId = (String) session.getAttributes().get(Constants.WEBSOCKET_LOGINNAME);
@ -233,7 +236,7 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
* 给某个用户发送消息 (兼容原有接口)
*/
public boolean sendMessageToUser(String loginName, SendVO message) {
return distributedMessageService.sendToUser(loginName, message,localSessionManager.getUserSessions());
return distributedMessageService.sendToUser(loginName, message);
}
/**
@ -333,7 +336,7 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
for (Integer userId : onlineUserIds) {
String userIdStr = String.valueOf(userId);
distributedMessageService.sendToUser(userIdStr, message,localSessionManager.getUserSessions());
distributedMessageService.sendToUser(userIdStr, message);
}
logger.debug("向 {} 个在线用户发送消息", onlineUserIds.size());
@ -360,4 +363,13 @@ public class MallsuiteImSocketHandler implements WebSocketHandler {
logger.info("添加用户会话成功: userId={}, sessionId={}", userId, session.getId());
}
/**
* 添加群组会话
*/
public void addGroupSession(String groupId, WebSocketSession session) {
groupSessions.computeIfAbsent(groupId, k -> new CopyOnWriteArrayList<>())
.add(session);
}
}