如何让RabbitMQ自动恢复功能在C#库中生效？

我使用RabbitMQ C#库构建了一个简单的连接，当前可正常工作并传输数据，但遇到以下问题：当RabbitMQ Broker临时宕机后，无论发布者还是消费者都无法在Broker恢复可达性后重新连接。查看ConnectionFactory的源码可知自动恢复功能默认处于启用状态，但即使过了数分钟，RabbitMQ管理界面中仍未显示任何连接。

当前代码 #

RabbitMQ.Client.ConnectionFactory _factory;
RabbitMQ.Client.IConnection _connection;
RabbitMQ.Client.IModel _channel;
String _queueName;
EventingBasicConsumer _consumer;
ConcurrentQueue<byte[]> _queue;

try
{
   _factory = new ConnectionFactory { 
      HostName = "some.url.com",
      Port = 5671, 
      UserName = "guest", 
      Password = "guest", 
      VirtualHost = "/",
      AutomaticRecoveryEnabled = true,
      NetworkRecoveryInterval = TimeSpan.FromSeconds(10),
      ContinuationTimeout = TimeSpan.FromSeconds(10),
      RequestedHeartbeat = TimeSpan.FromSeconds(10)
   };

   _connection = _factory.CreateConnection();   
}
catch(Exception e)
{
    Debug.WriteLine("Error while connection to RabbitMQ Broker: " + e.Message);
    throw new Exception("Could not connect to MQTT Broker");
}
   
_channel = _connection.CreateModel();
_channel.ExchangeDeclare(exchange: "exchangekey", 
                         type: ExchangeType.Topic);

_queueName = _channel.QueueDeclare().QueueName; 

_channel.QueueBind(queue: _queueName,
                   exchange: "exchangekey",
                   routingKey: "routingkey"); 

_consumer = new EventingBasicConsumer(_channel);
_consumer.Received += (source, args) =>
{
    _queue.Enqueue(args.Body.ToArray());
};
_channel.BasicConsume(queue: _queueName,
                      autoAck: true,
                      consumer: _consumer);

额外添加的配置项 #

AutomaticRecoveryEnabled = true,
NetworkRecoveryInterval = TimeSpan.FromSeconds(10),
ContinuationTimeout = TimeSpan.FromSeconds(10),
RequestedHeartbeat = TimeSpan.FromSeconds(10)

解决方案 #

• 监控连接状态与异常处理：当前仅在初始化阶段捕获连接异常，自动恢复过程中的异常会被忽略。建议订阅连接的ConnectionShutdown、CallbackException、ConnectionBlocked事件，及时排查恢复失败的原因：

  _connection.ConnectionShutdown += (sender, args) => Debug.WriteLine("连接已断开: " + args.ReplyText);
  _connection.CallbackException += (sender, args) => Debug.WriteLine("回调异常: " + args.Exception.Message);
  _connection.ConnectionBlocked += (sender, args) => Debug.WriteLine("连接被阻塞: " + args.Reason);
  

• 手动恢复通道与消费者：RabbitMQ自动恢复仅负责重建连接，不会自动恢复通道上的交换器、队列声明及消费者。需在连接恢复成功后重新执行这些操作，可通过ConnectionRecoverySucceeded事件实现：

  _connection.ConnectionRecoverySucceeded += (sender, args) =>
  {
      // 重新创建通道
      _channel = _connection.CreateModel();
      // 重新声明交换器、队列、绑定
      _channel.ExchangeDeclare(exchange: "exchangekey", type: ExchangeType.Topic);
      _queueName = _channel.QueueDeclare().QueueName;
      _channel.QueueBind(queue: _queueName, exchange: "exchangekey", routingKey: "routingkey");
      
      // 重新注册消费者
      _consumer = new EventingBasicConsumer(_channel);
      _consumer.Received += (source, args) =>
      {
          _queue.Enqueue(args.Body.ToArray());
      };
      _channel.BasicConsume(queue: _queueName, autoAck: true, consumer: _consumer);
  };
  

• 完善SSL配置：你使用的是5671端口（AMQP over TLS），需确保启用SSL选项，否则恢复时可能因SSL握手失败导致重连失败：

  _factory = new ConnectionFactory { 
      // 原有配置...
      Ssl = new SslOption { Enabled = true }
  };
  

• 调整超时参数：ContinuationTimeout设置为10秒过短，可能因网络延迟导致恢复超时，建议调整为30秒或更长：

  ContinuationTimeout = TimeSpan.FromSeconds(30)
  

• 避免变量引用问题：确保_connection、_channel等变量在恢复过程中被正确重新赋值，避免引用失效的旧实例。

内容的提问来源于stack exchange，提问作者abones