RabbitMQ集群节点故障时CreateModel调用耗时过长问题排查

我搭建了一个RabbitMQ集群，应用连接集群后运行正常，但发现当某一节点故障时，发布消息耗时远超正常水平（约5-10秒）。以下是应用中全局单例的RabbitMQ连接及创建通道的代码：

public class RabbitMQConnection : IDisposable
{
    private readonly ConnectionFactory _connectionFactory;
    private readonly object _lock;
    private readonly IList<AmqpTcpEndpoint> _endpoints;
    private bool _shouldClose = false;

    private IConnection Connection { get; set; }

    public RabbitMQConnection(IEnumerable<string> hostList, string username, string password)
    {
        _endpoints = hostList.Select(s =>
        {
            var split = s.Split(':');
            var hostname = split[0];
            int port = -1;
            if (split.Length > 1 && int.TryParse(split[1], out int parsedPort))
                port = parsedPort;
            return new AmqpTcpEndpoint(hostname, port);
        }).ToList();
        _connectionFactory = new ConnectionFactory
        {
            UserName = username,
            Password = password,
            AutomaticRecoveryEnabled = true,
            NetworkRecoveryInterval = TimeSpan.FromSeconds(5),
        };

        _lock = new object();
    }

    ~RabbitMQConnection() => Dispose();

    private void StartConnection()
    {
        if (_shouldClose)
            return;

        lock (_lock)
        {
            if (Connection?.IsOpen == true)
                return;

            if (Connection != null)
                Connection.Dispose();

            Connection = _connectionFactory.CreateConnection(_endpoints);
            Connection.ConnectionShutdown += Connection_ConnectionShutdown;
        }
    }

    private void Connection_ConnectionShutdown(object sender, ShutdownEventArgs e) => StartConnection();

    public void Dispose()
    {
        _shouldClose = true;
        lock (_lock)
        {
            if (Connection != null)
            {
                if (Connection.IsOpen)
                    Connection.Close();
                Connection.Dispose();
                Connection = null;
            }
        }
    }

    public IModel CreateChannel()
    {
        StartConnection();
        return this.Connection.CreateModel();
    }
}

问题在于，当某节点故障时，this.Connection.CreateModel()这一行需要约5秒才会返回，请问我忽略了什么配置或逻辑？

分析与解决方法 #

• 未配置连接/请求超时：当前ConnectionFactory未设置超时参数，客户端尝试连接故障节点时会使用默认超时（通常5秒左右）。需显式设置较短的超时时间，减少等待：

  _connectionFactory = new ConnectionFactory
  {
      UserName = username,
      Password = password,
      AutomaticRecoveryEnabled = true,
      NetworkRecoveryInterval = TimeSpan.FromSeconds(5),
      RequestTimeout = TimeSpan.FromSeconds(1), // 高版本客户端用此参数
      // 低版本客户端可替换为 ConnectionTimeout = TimeSpan.FromSeconds(1)
  };
  

• 故障节点未被动态排除：每次创建连接时仍会遍历包含故障节点的完整列表，重复触发超时等待。可以维护健康端点列表：

  1. 监听Connection.ConnectionFailed事件，记录故障节点
  2. 创建连接时优先使用健康节点，定期对故障节点进行重试检测，恢复后重新加入列表

• 自动恢复与同步阻塞冲突：StartConnection是同步执行的，当连接触发自动恢复时，CreateModel()会等待恢复完成才返回。可以将连接恢复逻辑改为异步，避免阻塞通道创建请求：

  private async Task StartConnectionAsync()
  {
      if (_shouldClose) return;
      lock (_lock)
      {
          if (Connection?.IsOpen == true) return;
          Connection?.Dispose();
      }
      // 异步创建连接，不阻塞主线程
      var newConn = await _connectionFactory.CreateConnectionAsync(_endpoints);
      lock (_lock)
      {
          if (_shouldClose) { newConn.Dispose(); return; }
          Connection = newConn;
          Connection.ConnectionShutdown += async (s, e) => await StartConnectionAsync();
      }
  }
  
  public async Task<IModel> CreateChannelAsync()
  {
      await StartConnectionAsync();
      return this.Connection.CreateModel();
  }
  

• 通道创建时的隐式验证：CreateModel()会同步验证连接状态，若连接处于恢复中则等待。改用异步方法CreateModelAsync()可以避免阻塞，让调用方自行处理异步逻辑。

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