.NET集成RabbitMQ首次高并发CreateConnection()失败

问题现象 #

在.NET应用中采用「单例维护IConnection连接 + 对象池管理IModel通道」的方案使用RabbitMQ时，出现如下异常：

• 服务启动后首次有大量并发请求调用消息发送方法，执行factory.CreateConnection()时抛出异常：

RabbitMQ.Client.Exceptions.BrokerUnreachableException
Message=None of the specified endpoints were reachable
IOException: connection.start was never received, likely due to a network timeout

• 如果首次调用发送方法时只有单个请求，连接创建成功后，后续即使有大量并发请求也可正常运行。

原有实现代码如下：

IPooledObjectPolicy<IModel>实现 #

private readonly RabbitMqConfigurations rabbitMqConfigurations;
private readonly IConnection _connection;

public RabbitModelPooledObjectPolicy(IOptions<RabbitMqConfigurations> RabbitMqConfigurations)
{
    rabbitMqConfigurations = RabbitMqConfigurations.Value;
    _connection = GetConnection();
}

private IConnection GetConnection()
{
    var factory = new ConnectionFactory
    {
        HostName = rabbitMqConfigurations.HostName
    };

    return factory.CreateConnection(); // 异常抛出点
}

public IModel Create() => return _connection.CreateModel();

public bool Return(IModel obj)
{
    if (obj.IsOpen)
        return true;
    else
    {
        obj?.Dispose();
        return false;
    }
}

消息发送实现 #

private readonly DefaultObjectPool<IModel> objectPool;

public RabbitMQProducer(IPooledObjectPolicy<IModel> objectPolicy)
{
    objectPool = new DefaultObjectPool<IModel>(objectPolicy, Environment.ProcessorCount * 2);
}

public void SendMessage<T>(T message, string queue) where T : class
{
    var channel = objectPool.Get();

    try
    {
        channel.QueueDeclare(queue: queue, durable: true, exclusive: false, autoDelete: false, arguments: null);

        var properties = channel.CreateBasicProperties();
        properties.Persistent = true;

        var json = JsonConvert.SerializeObject(message);
        var body = Encoding.UTF8.GetBytes(json);

        channel.BasicPublish(exchange: "", routingKey: queue, basicProperties: properties, body: body);
    }
    catch
    {
        throw;
    }
    finally
    {
        objectPool.Return(channel);
    }
}

服务注册代码 #

services.AddSingleton<IPooledObjectPolicy<IModel>, RabbitModelPooledObjectPolicy>();
services.AddSingleton<IMessageProducer, RabbitMQProducer>();

问题根因 #

异常提示的网络超时只是表面现象，核心是连接和通道的初始化逻辑存在并发缺陷：

1. 初始化时机错误：当前代码没有在应用启动阶段主动初始化RabbitMQ连接，连接创建是被首次业务请求触发的。CreateConnection()需要完成TCP三次握手、AMQP协议协商，耗时较长。启动瞬间大量并发请求涌入时，极端场景下会出现单例初始化重入，多个线程并发调用CreateConnection()，短时间发起大量握手请求触发RabbitMQ服务端的同IP连接限流，服务端直接丢弃握手包，客户端等待connection.start帧超时。
2. 对象池无预热：DefaultObjectPool初始状态为空，即使连接只创建一次，多个并发请求会同时调用Create()方法，在刚建立、还未完全就绪的连接上并发创建通道，并发发送channel.open帧会打乱AMQP协议的帧交互顺序，导致服务端主动断开连接，抛出和网络超时完全一致的异常。
3. 连接创建完成、内部状态稳定后，RabbitMQ的IConnection本身是线程安全的，后续并发创建通道、发送消息不会触发这类问题，这也是首次单请求成功后高并发运行正常的原因。

修复方案 #

1. 优化连接创建逻辑，增加并发保护 #

修改RabbitModelPooledObjectPolicy，给连接创建加锁，增加连接状态检查、自动恢复配置，修复原有语法错误：

public class RabbitModelPooledObjectPolicy : IPooledObjectPolicy<IModel>
{
    private readonly RabbitMqConfigurations _rabbitMqConfigurations;
    private IConnection _connection;
    private static readonly object _connLock = new object();

    public RabbitModelPooledObjectPolicy(IOptions<RabbitMqConfigurations> config)
    {
        _rabbitMqConfigurations = config.Value;
        // 构造函数中触发首次连接创建
        EnsureConnection();
    }

    private void EnsureConnection()
    {
        // 连接正常时直接返回
        if (_connection is { IsOpen: true }) return;

        lock (_connLock)
        {
            // 双重检查避免重复创建
            if (_connection is { IsOpen: true }) return;
            
            _connection?.Dispose();
            var factory = new ConnectionFactory
            {
                HostName = _rabbitMqConfigurations.HostName,
                // 适当调大连接超时，避免启动时网络波动触发超时
                RequestedConnectionTimeout = TimeSpan.FromSeconds(15),
                // 开启自动连接恢复
                AutomaticRecoveryEnabled = true,
                NetworkRecoveryInterval = TimeSpan.FromSeconds(10)
            };
            _connection = factory.CreateConnection();
        }
    }

    public IModel Create()
    {
        EnsureConnection();
        return _connection.CreateModel();
    }

    public bool Return(IModel obj)
    {
        if (obj.IsOpen) return true;
        
        obj?.Dispose();
        // 通道异常时检查连接状态，必要时触发重连
        EnsureConnection();
        return false;
    }
}

2. 预热对象池，避免首次并发创建通道 #

修改RabbitMQProducer构造函数，初始化对象池后提前创建好所有需要的通道实例，避免首次请求并发创建通道打乱连接状态：

public class RabbitMQProducer : IMessageProducer
{
    private readonly DefaultObjectPool<IModel> _objectPool;

    public RabbitMQProducer(IPooledObjectPolicy<IModel> objectPolicy)
    {
        _objectPool = new DefaultObjectPool<IModel>(objectPolicy, Environment.ProcessorCount * 2);
        // 预热对象池：提前创建满容量的通道
        for (int i = 0; i < Environment.ProcessorCount * 2; i++)
        {
            var channel = _objectPool.Get();
            _objectPool.Return(channel);
        }
    }

    // SendMessage方法保持原有逻辑即可
}

3. 应用启动阶段主动预热服务 #

在Program.cs中，应用启动后主动解析RabbitMQ相关服务，保证应用开始接收请求前连接、对象池已经完全初始化完成：

var app = builder.Build();

// 其他中间件配置...

// 主动解析服务，触发连接和对象池预热
app.Services.GetRequiredService<IMessageProducer>();

app.Run();

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