Skip to content

Y
yd-communication
Commit a9fd76ea by zhangxingmin
Options

push

parent 5217ecc0
Showing with 32 additions and 70 deletions
yd-communication-api/src/main/java/com/yd/communication/api/websocket/CoWebSocketServer.java
...@@ -12,6 +12,8 @@ import com.yd.communication.service.model.CoSession; ...@@ -12,6 +12,8 @@ import com.yd.communication.service.model.CoSession;
import com.yd.communication.service.service.ICoOperationLogService; import com.yd.communication.service.service.ICoOperationLogService;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent; import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.event.EventListener; import org.springframework.context.event.EventListener;
...@@ -124,6 +126,13 @@ public class CoWebSocketServer { ...@@ -124,6 +126,13 @@ public class CoWebSocketServer {
CoWebSocketServer.redisTemplate = redisTemplate; CoWebSocketServer.redisTemplate = redisTemplate;
} }
private static RedissonClient redissonClient;
@Autowired
public void setRedissonClient(RedissonClient redissonClient) {
CoWebSocketServer.redissonClient = redissonClient;
}
// ==================== Redis 键名常量 ==================== // ==================== Redis 键名常量 ====================
/** /**
...@@ -156,73 +165,26 @@ public class CoWebSocketServer { ...@@ -156,73 +165,26 @@ public class CoWebSocketServer {
// ==================== 跨节点订阅（每个节点启动时执行） ==================== // ==================== 跨节点订阅（每个节点启动时执行） ====================
/**
* 监听应用启动完成事件，在容器完全就绪后启动 Redis 订阅
* 替代 @PostConstruct，避免 Bean 未完全初始化时连接 Redis
*/
@EventListener(ApplicationReadyEvent.class) @EventListener(ApplicationReadyEvent.class)
public void startRedisSubscriber() { public void startRedisSubscriber() {
log.info("========================================"); log.info("应用已完全启动，开始初始化 Redisson RTopic 订阅...");
log.info("应用已完全启动，开始初始化 Redis 订阅...");
log.info("当前节点ID: {}", NODE_ID);
log.info("RedisTemplate 是否为空: {}", redisTemplate == null ? "是" : "否");
if (redisTemplate != null) {
log.info("RedisTemplate 类型: {}", redisTemplate.getClass().getName());
}
log.info("========================================");
new Thread(() -> { new Thread(() -> {
doSubscribe(); // 订阅全局频道
}).start(); RTopic topic = redissonClient.getTopic("coordination:channel");
} topic.addListener(String.class, (channel, msg) -> {
log.debug("收到跨节点消息: {}", msg);
handleCrossNodeMessage(msg);
/** });
* 执行 Redis 订阅（支持自动重连） log.info("Redisson RTopic 订阅已启动，频道: coordination:channel");
* 使用 RedisConnection.pSubscribe 进行持久订阅 // 保持线程存活（监听在后台异步进行，线程无需阻塞，但保留以防止退出）
*/ while (true) {
private void doSubscribe() { try {
log.info("=== Redis 订阅线程启动 ==="); Thread.sleep(60000);
log.info("当前节点 ID: {}", NODE_ID); } catch (InterruptedException e) {
break;
while (true) {
RedisConnection connection = null;
try {
// 1. 获取 Redis 连接（非回调方式，保持长连接）
connection = redisTemplate.getConnectionFactory().getConnection();
log.info("获取 Redis 连接成功，开始模式订阅 room:channel:* ...");
// 2. 阻塞订阅（模式匹配）
connection.pSubscribe((message, pattern) -> {
// 收到消息时调用
log.debug("收到跨节点消息，频道: {}, 消息长度: {}",
new String(pattern), message.getBody().length);
handleCrossNodeMessage(new String(message.getBody()));
}, "room:channel:*".getBytes());
// 正常情况 pSubscribe 会一直阻塞，执行到这里说明订阅被取消或连接断开
log.warn("Redis 订阅意外结束，准备重连...");
} catch (Exception e) {
log.error("Redis 订阅异常: {}", e.getMessage(), e);
} finally {
// 关闭连接（如果订阅已结束）
if (connection != null) {
try {
connection.close();
} catch (Exception e) {
// ignore
}
} }
} }
}).start();
// 3. 重连等待
try {
Thread.sleep(5000);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
break;
}
}
} }
/** /**
...@@ -644,17 +606,17 @@ public class CoWebSocketServer { ...@@ -644,17 +606,17 @@ public class CoWebSocketServer {
} }
} }
// 2. 发布到 Redis，通知其他节点广播 // 2. 跨节点广播（发布到 Redisson RTopic）
String channel = String.format(ROOM_BROADCAST_CHANNEL, roomId);
Map<String, String> data = new HashMap<>(); Map<String, String> data = new HashMap<>();
data.put("roomId", roomId); // 房间号 data.put("roomId", roomId);
data.put("message", message); // 消息内容 data.put("message", message);
data.put("sourceNodeId", NODE_ID); // 标记消息来源节点 data.put("sourceNodeId", NODE_ID);
if (exclude != null) { if (exclude != null) {
data.put("excludeSessionId", exclude.getId()); // 需要排除的 sessionId data.put("excludeSessionId", exclude.getId());
} }
// 将数据转为 JSON 并发布到 Redis 频道 String jsonMsg = JSON.toJSONString(data);
redisTemplate.convertAndSend(channel, JSON.toJSONString(data)); RTopic topic = redissonClient.getTopic("coordination:channel");
topic.publish(jsonMsg);
} }
/** /**
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment