push

yd-communication-api/src/main/java/com/yd/communication/api/websocket/CoWebSocketServer.java

@@ -27,6 +27,7 @@ import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
+import org.springframework.data.redis.connection.RedisConnection;
 
 
 /**
 /**
  * 协同 WebSocket 服务端（支持多节点部署）
  * 协同 WebSocket 服务端（支持多节点部署）
@@ -174,62 +175,55 @@ public class CoWebSocketServer {
         }).start();
         }).start();
     }
     }
 
 
+
     /**
     /**
      * 执行 Redis 订阅（支持自动重连）
      * 执行 Redis 订阅（支持自动重连）
-     * 增加详细日志，便于排查 Redis 连接问题
+     * 使用 RedisConnection.pSubscribe 进行持久订阅
      */
      */
     private void doSubscribe() {
     private void doSubscribe() {
         log.info("=== Redis 订阅线程启动 ===");
         log.info("=== Redis 订阅线程启动 ===");
         log.info("当前节点 ID: {}", NODE_ID);
         log.info("当前节点 ID: {}", NODE_ID);
 
 
         while (true) {
         while (true) {
+            RedisConnection connection = null;
             try {
             try {
-                log.info("尝试连接 Redis，地址: {}:{}",
-                        redisTemplate.getConnectionFactory().getConnection().getNativeConnection()  // 这行可能报错，改用手动获取配置
-                );
-                // 改为从配置中读取 host/port（更好的方式，但这里简化为打印配置）
-                // 由于无法直接获取，可打印 RedisTemplate 的哈希码，确认不为 null
-                log.info("RedisTemplate 实例: {}", System.identityHashCode(redisTemplate));
+                // 1. 获取 Redis 连接（非回调方式，保持长连接）
+                connection = redisTemplate.getConnectionFactory().getConnection();
+                log.info("获取 Redis 连接成功，开始模式订阅 room:channel:* ...");
 
 
-                // 先发送 PING 测试连接
-                try {
-                    String pingResult = redisTemplate.execute((connection) -> {
-                        return connection.ping();
-                    }, true);
-                    log.info("Redis PING 响应: {}", pingResult);
-                } catch (Exception pingEx) {
-                    log.error("Redis PING 失败: {}", pingEx.getMessage());
-                    // 继续尝试订阅，因为可能订阅能成功但 PING 失败（少见）
-                }
-
-                log.info("开始订阅 Redis 频道: room:channel:*");
-                redisTemplate.execute((connection) -> {
-                    connection.subscribe(
-                            (message, pattern) -> {
+                // 2. 阻塞订阅（模式匹配）
+                connection.pSubscribe((message, pattern) -> {
+                    // 收到消息时调用
                     log.debug("收到跨节点消息，频道: {}, 消息长度: {}",
                     log.debug("收到跨节点消息，频道: {}, 消息长度: {}",
                             new String(pattern), message.getBody().length);
                             new String(pattern), message.getBody().length);
                     handleCrossNodeMessage(new String(message.getBody()));
                     handleCrossNodeMessage(new String(message.getBody()));
-                            },
-                            "room:channel:*".getBytes()
-                    );
-                    log.warn("Redis 订阅意外退出，将重新尝试...");
-                    return null;
-                }, true);
+                }, "room:channel:*".getBytes());
+
+                // 正常情况 pSubscribe 会一直阻塞，执行到这里说明订阅被取消或连接断开
+                log.warn("Redis 订阅意外结束，准备重连...");
 
 
             } catch (Exception e) {
             } catch (Exception e) {
-                log.error("Redis 订阅异常，错误类型: {}, 消息: {}",
-                        e.getClass().getSimpleName(), e.getMessage(), e);
-                log.error("完整堆栈: ", e);
+                log.error("Redis 订阅异常: {}", e.getMessage(), e);
+            } finally {
+                // 关闭连接（如果订阅已结束）
+                if (connection != null) {
+                    try {
+                        connection.close();
+                    } catch (Exception e) {
+                        // ignore
+                    }
+                }
+            }
+
+            // 3. 重连等待
             try {
             try {
                 Thread.sleep(5000);
                 Thread.sleep(5000);
             } catch (InterruptedException ex) {
             } catch (InterruptedException ex) {
                 Thread.currentThread().interrupt();
                 Thread.currentThread().interrupt();
-                    log.warn("Redis 订阅线程被中断，退出循环");
                 break;
                 break;
             }
             }
         }
         }
     }
     }
-    }
 
 
     /**
     /**
      * Bean 初始化完成后执行，启动 Redis 订阅线程
      * Bean 初始化完成后执行，启动 Redis 订阅线程