fix

Author: Cyaim

Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.FreeRedis/FreeRedisClusterTransport.cs

@@ -262,6 +262,50 @@ public bool IsNodeConnected(string nodeId)
             return _nodes.ContainsKey(nodeId);
         }
 
+        /// <summary>
+        /// Store connection route information (not supported by FreeRedis Pub/Sub transport)
+        /// 存储连接路由信息（FreeRedis Pub/Sub 传输不支持）
+        /// </summary>
+        public Task<bool> StoreConnectionRouteAsync(string connectionId, string nodeId, Dictionary<string, string> metadata = null)
+        {
+            // FreeRedis Pub/Sub transport doesn't support connection route storage
+            // FreeRedis Pub/Sub 传输不支持连接路由存储
+            return Task.FromResult(false);
+        }
+
+        /// <summary>
+        /// Get connection route information (not supported by FreeRedis Pub/Sub transport)
+        /// 获取连接路由信息（FreeRedis Pub/Sub 传输不支持）
+        /// </summary>
+        public Task<string> GetConnectionRouteAsync(string connectionId)
+        {
+            // FreeRedis Pub/Sub transport doesn't support connection route query
+            // FreeRedis Pub/Sub 传输不支持连接路由查询
+            return Task.FromResult<string>(null);
+        }
+
+        /// <summary>
+        /// Remove connection route information (not supported by FreeRedis Pub/Sub transport)
+        /// 删除连接路由信息（FreeRedis Pub/Sub 传输不支持）
+        /// </summary>
+        public Task<bool> RemoveConnectionRouteAsync(string connectionId)
+        {
+            // FreeRedis Pub/Sub transport doesn't support connection route removal
+            // FreeRedis Pub/Sub 传输不支持连接路由删除
+            return Task.FromResult(false);
+        }
+
+        /// <summary>
+        /// Refresh connection route expiration (not supported by FreeRedis Pub/Sub transport)
+        /// 刷新连接路由过期时间（FreeRedis Pub/Sub 传输不支持）
+        /// </summary>
+        public Task<bool> RefreshConnectionRouteAsync(string connectionId, string nodeId)
+        {
+            // FreeRedis Pub/Sub transport doesn't support connection route refresh
+            // FreeRedis Pub/Sub 传输不支持连接路由刷新
+            return Task.FromResult(false);
+        }
+
         /// <summary>
         /// Register a node / 注册节点
         /// </summary>

Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid.Implementations/RabbitMQMessageQueueService.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 using Cyaim.WebSocketServer.Cluster.Hybrid.Abstractions;
 using Microsoft.Extensions.Logging;
@@ -20,6 +21,7 @@ public class RabbitMQMessageQueueService : IMessageQueueService
         private IModel _channel;
         private readonly Dictionary<string, EventingBasicConsumer> _consumers;
         private bool _disposed = false;
+        private readonly SemaphoreSlim _channelLock = new SemaphoreSlim(1, 1);
 
         /// <summary>
         /// Constructor / 构造函数
@@ -38,22 +40,49 @@ public RabbitMQMessageQueueService(ILogger<RabbitMQMessageQueueService> logger,
         /// </summary>
         public async Task ConnectAsync()
         {
-            if (_connection != null && _connection.IsOpen)
-            {
-                return;
-            }
-
+            // 避免并发重连 / Avoid concurrent reconnects
+            await _channelLock.WaitAsync().ConfigureAwait(false);
             try
             {
-                var factory = new ConnectionFactory { Uri = new Uri(_connectionString) };
-                _connection = factory.CreateConnection();
-                _channel = _connection.CreateModel();
-                _logger.LogInformation("Connected to RabbitMQ");
+                if (_connection != null && _connection.IsOpen && _channel != null && _channel.IsOpen)
+                {
+                    return;
+                }
+
+                // 清理旧连接 / Cleanup old connection
+                try
+                {
+                    _channel?.Close();
+                    _channel?.Dispose();
+                }
+                catch { }
+
+                try
+                {
+                    _connection?.Close();
+                    _connection?.Dispose();
+                }
+                catch { }
+
+                _connection = null;
+                _channel = null;
+
+                try
+                {
+                    var factory = new ConnectionFactory { Uri = new Uri(_connectionString) };
+                    _connection = factory.CreateConnection();
+                    _channel = _connection.CreateModel();
+                    _logger.LogInformation("Connected to RabbitMQ");
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogError(ex, "Failed to connect to RabbitMQ");
+                    throw;
+                }
             }
-            catch (Exception ex)
+            finally
             {
-                _logger.LogError(ex, "Failed to connect to RabbitMQ");
-                throw;
+                _channelLock.Release();
             }
         }
 
@@ -85,10 +114,7 @@ public async Task DisconnectAsync()
         /// </summary>
         public async Task DeclareExchangeAsync(string exchangeName, string exchangeType, bool durable = true)
         {
-            if (_channel == null)
-            {
-                throw new InvalidOperationException("RabbitMQ is not connected");
-            }
+            await EnsureChannelAsync().ConfigureAwait(false);
 
             _channel.ExchangeDeclare(exchangeName, exchangeType, durable: durable, autoDelete: false);
             await Task.CompletedTask;
@@ -99,10 +125,7 @@ public async Task DeclareExchangeAsync(string exchangeName, string exchangeType,
         /// </summary>
         public async Task<string> DeclareQueueAsync(string queueName, bool durable = false, bool exclusive = false, bool autoDelete = true)
         {
-            if (_channel == null)
-            {
-                throw new InvalidOperationException("RabbitMQ is not connected");
-            }
+            await EnsureChannelAsync().ConfigureAwait(false);
 
             var result = _channel.QueueDeclare(
                 queue: queueName,
@@ -119,10 +142,7 @@ public async Task<string> DeclareQueueAsync(string queueName, bool durable = fal
         /// </summary>
         public async Task BindQueueAsync(string queueName, string exchangeName, string routingKey)
         {
-            if (_channel == null)
-            {
-                throw new InvalidOperationException("RabbitMQ is not connected");
-            }
+            await EnsureChannelAsync().ConfigureAwait(false);
 
             _channel.QueueBind(queueName, exchangeName, routingKey);
             await Task.CompletedTask;
@@ -133,9 +153,11 @@ public async Task BindQueueAsync(string queueName, string exchangeName, string r
         /// </summary>
         public async Task PublishAsync(string exchangeName, string routingKey, byte[] message, MessageProperties properties = null)
         {
-            if (_channel == null)
+            await EnsureChannelAsync().ConfigureAwait(false);
+
+            if (_channel == null || !_channel.IsOpen)
             {
-                throw new InvalidOperationException("RabbitMQ is not connected");
+                throw new InvalidOperationException("RabbitMQ channel is not open");
             }
 
             var basicProperties = _channel.CreateBasicProperties();
@@ -171,10 +193,7 @@ public async Task PublishAsync(string exchangeName, string routingKey, byte[] me
         /// </summary>
         public async Task ConsumeAsync(string queueName, Func<byte[], MessageProperties, Task<bool>> handler, bool autoAck = false)
         {
-            if (_channel == null)
-            {
-                throw new InvalidOperationException("RabbitMQ is not connected");
-            }
+            await EnsureChannelAsync().ConfigureAwait(false);
 
             if (_consumers.ContainsKey(queueName))
             {
@@ -244,10 +263,7 @@ public async Task ConsumeAsync(string queueName, Func<byte[], MessageProperties,
         /// </summary>
         public async Task AckAsync(ulong deliveryTag)
         {
-            if (_channel == null)
-            {
-                throw new InvalidOperationException("RabbitMQ is not connected");
-            }
+            await EnsureChannelAsync().ConfigureAwait(false);
 
             _channel.BasicAck(deliveryTag, false);
             await Task.CompletedTask;
@@ -258,15 +274,25 @@ public async Task AckAsync(ulong deliveryTag)
         /// </summary>
         public async Task RejectAsync(ulong deliveryTag, bool requeue = false)
         {
-            if (_channel == null)
-            {
-                throw new InvalidOperationException("RabbitMQ is not connected");
-            }
+            await EnsureChannelAsync().ConfigureAwait(false);
 
             _channel.BasicNack(deliveryTag, false, requeue);
             await Task.CompletedTask;
         }
 
+        /// <summary>
+        /// Ensure connection and channel are open / 确保连接和通道已打开
+        /// </summary>
+        private async Task EnsureChannelAsync()
+        {
+            if (_channel != null && _channel.IsOpen && _connection != null && _connection.IsOpen)
+            {
+                return;
+            }
+
+            await ConnectAsync().ConfigureAwait(false);
+        }
+
         /// <summary>
         /// Dispose / 释放资源
         /// </summary>