Skip to content

Commit 19f4533

Browse files
committed
fix
1 parent 1b0ff03 commit 19f4533

1 file changed

Lines changed: 97 additions & 8 deletions

File tree

‎Cyaim.WebSocketServer/Cluster/Cyaim.WebSocketServer.Cluster.Hybrid.MessageQueue.RabbitMQ/RabbitMQMessageQueueService.cs‎

Lines changed: 97 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -103,9 +103,13 @@ public async Task ConnectAsync()
103103
try
104104
{
105105
// 如果已连接且正常,直接返回
106-
if (IsConnected())
106+
var isConnected = IsConnected();
107+
_logger.LogDebug("[RabbitMQMessageQueueService] ConnectAsync 调用 - IsConnected: {IsConnected}, ConnectionIsNull: {ConnectionIsNull}, ConnectionIsOpen: {ConnectionIsOpen}, ChannelIsNull: {ChannelIsNull}, ChannelIsOpen: {ChannelIsOpen}, ConsumerCount: {ConsumerCount}",
108+
isConnected, _connection == null, _connection?.IsOpen ?? false, _channel == null, _channel?.IsOpen ?? false, _consumers.Count);
109+
110+
if (isConnected)
107111
{
108-
_logger.LogDebug("[RabbitMQMessageQueueService] 已连接,跳过重连");
112+
_logger.LogDebug("[RabbitMQMessageQueueService] 已连接,跳过重连 - ConsumerCount: {ConsumerCount}", _consumers.Count);
109113
return;
110114
}
111115

@@ -148,7 +152,47 @@ private async Task InternalConnectAsync()
148152
{
149153
try
150154
{
151-
// 清理旧连接
155+
// 双重检查:再次验证连接状态,确保不会误清理
156+
// Double-check: verify connection status again to ensure we don't accidentally cleanup
157+
var isConnectedCheck = IsConnected();
158+
_logger.LogDebug("[RabbitMQMessageQueueService] InternalConnectAsync 内部检查 - IsConnected: {IsConnected}, ConnectionIsNull: {ConnectionIsNull}, ConnectionIsOpen: {ConnectionIsOpen}, ChannelIsNull: {ChannelIsNull}, ChannelIsOpen: {ChannelIsOpen}, ConsumerCount: {ConsumerCount}",
159+
isConnectedCheck, _connection == null, _connection?.IsOpen ?? false, _channel == null, _channel?.IsOpen ?? false, _consumers.Count);
160+
161+
if (isConnectedCheck)
162+
{
163+
_logger.LogDebug("[RabbitMQMessageQueueService] 内部检查:连接和 channel 都正常,跳过清理和重连 - ConsumerCount: {ConsumerCount}", _consumers.Count);
164+
return;
165+
}
166+
167+
// 只有在连接或 channel 真正关闭时才清理
168+
// Only cleanup when connection or channel is truly closed
169+
bool connectionClosed = _connection == null || !_connection.IsOpen;
170+
bool channelClosed = _channel == null || !_channel.IsOpen;
171+
172+
// 如果 connection 和 channel 都正常,但 IsConnected() 返回 false,可能是检查时机问题
173+
// 这种情况下,我们不应该清理,而是直接返回
174+
if (!connectionClosed && !channelClosed)
175+
{
176+
_logger.LogWarning("[RabbitMQMessageQueueService] 连接和 channel 都正常,但 IsConnected() 返回 false,可能是检查时机问题,跳过清理和重连 - ConsumerCount: {ConsumerCount}", _consumers.Count);
177+
return;
178+
}
179+
180+
// 记录需要清理的原因
181+
if (connectionClosed)
182+
{
183+
_logger.LogInformation("[RabbitMQMessageQueueService] Connection 为 null 或已关闭,需要清理 - ConnectionIsNull: {ConnectionIsNull}, ConnectionIsOpen: {ConnectionIsOpen}, ConsumerCount: {ConsumerCount}",
184+
_connection == null, _connection?.IsOpen ?? false, _consumers.Count);
185+
}
186+
else if (channelClosed)
187+
{
188+
_logger.LogInformation("[RabbitMQMessageQueueService] Channel 为 null 或已关闭,需要清理 - ChannelIsNull: {ChannelIsNull}, ChannelIsOpen: {ChannelIsOpen}, ConsumerCount: {ConsumerCount}",
189+
_channel == null, _channel?.IsOpen ?? false, _consumers.Count);
190+
}
191+
192+
// 只有在真正需要时才清理旧连接
193+
// Only cleanup old connection when truly needed
194+
_logger.LogInformation("[RabbitMQMessageQueueService] 开始清理旧连接 - ConnectionClosed: {ConnectionClosed}, ChannelClosed: {ChannelClosed}, ConsumerCount: {ConsumerCount}",
195+
connectionClosed, channelClosed, _consumers.Count);
152196
await CleanupConnectionAsync().ConfigureAwait(false);
153197

154198
_logger.LogInformation("[RabbitMQMessageQueueService] 开始连接 RabbitMQ");
@@ -182,12 +226,36 @@ private async Task CleanupConnectionAsync()
182226
{
183227
try
184228
{
185-
// 取消所有消费者
186-
await CancelAllConsumersAsync().ConfigureAwait(false);
229+
// 记录清理前的状态
230+
var consumerCountBefore = _consumers.Count;
231+
var consumerTagCountBefore = _consumerTags.Count;
232+
var consumerInfoCountBefore = _consumerInfos.Count;
233+
234+
_logger.LogInformation("[RabbitMQMessageQueueService] 开始清理连接 - ConsumerCount: {ConsumerCount}, ConsumerTagCount: {ConsumerTagCount}, ConsumerInfoCount: {ConsumerInfoCount}",
235+
consumerCountBefore, consumerTagCountBefore, consumerInfoCountBefore);
187236

188-
// 清理字典
237+
// 只有在 channel 可用时才取消消费者
238+
// Only cancel consumers if channel is available
239+
if (_channel != null && _channel.IsOpen)
240+
{
241+
_logger.LogInformation("[RabbitMQMessageQueueService] Channel 可用,取消所有消费者");
242+
await CancelAllConsumersAsync().ConfigureAwait(false);
243+
}
244+
else
245+
{
246+
_logger.LogInformation("[RabbitMQMessageQueueService] Channel 不可用,跳过取消消费者 - ChannelIsNull: {ChannelIsNull}, ChannelIsOpen: {ChannelIsOpen}",
247+
_channel == null, _channel?.IsOpen ?? false);
248+
}
249+
250+
// 清理字典(但保留 _consumerInfos 以便重连后恢复)
251+
// Clear dictionaries (but keep _consumerInfos for reconnection recovery)
189252
_consumers.Clear();
190253
_consumerTags.Clear();
254+
// 注意:不清理 _consumerInfos,以便重连后恢复消费者
255+
// Note: Don't clear _consumerInfos, so consumers can be restored after reconnection
256+
257+
_logger.LogInformation("[RabbitMQMessageQueueService] 清理完成 - ConsumerCountBefore: {ConsumerCountBefore}, ConsumerInfoCount: {ConsumerInfoCount}",
258+
consumerCountBefore, _consumerInfos.Count);
191259

192260
// 关闭 channel
193261
if (_channel != null)
@@ -244,8 +312,29 @@ private bool IsConnected()
244312
{
245313
try
246314
{
247-
return _connection != null && _connection.IsOpen &&
248-
_channel != null && _channel.IsOpen;
315+
// 双重检查:不仅检查 IsOpen,还尝试访问属性以确保真正可用
316+
// Double-check: not only check IsOpen, but also try to access properties to ensure truly available
317+
if (_connection == null || _channel == null)
318+
{
319+
return false;
320+
}
321+
322+
if (!_connection.IsOpen || !_channel.IsOpen)
323+
{
324+
return false;
325+
}
326+
327+
// 额外验证:尝试访问 channel 属性,确保 channel 真正可用
328+
// Additional verification: try to access channel property to ensure channel is truly usable
329+
try
330+
{
331+
var _ = _channel.ChannelNumber; // 如果 channel 已关闭,访问属性会抛出异常
332+
return true;
333+
}
334+
catch
335+
{
336+
return false;
337+
}
249338
}
250339
catch
251340
{

0 commit comments

Comments
 (0)