c# - RabbitMQ 在 .CreateModel() 上抛出 TimeOut 异常

Question

我创建了一个总线应用程序来封装所有订阅和发布到 RabbitMQ 代理。然后将该总线用作我的其他项目的 SDK。运行此项目后(对于使用我的 Bus sdk 通过 RabbitMQ 消息进行通信的应用程序而言，这是基本的)，我意识到 Bus 在 native 方法 .CreateModel() 上抛出一个 TimeOut 异常，如下所示:

System.TimeoutException: The operation has timed out.
   at RabbitMQ.Util.BlockingCell.GetValue(TimeSpan timeout)
   at RabbitMQ.Client.Impl.SimpleBlockingRpcContinuation.GetReply(TimeSpan timeout)
   at RabbitMQ.Client.Impl.ModelBase.ModelRpc(MethodBase method, ContentHeaderBase header, Byte[] body)
   at RabbitMQ.Client.Framing.Impl.Model._Private_ChannelOpen(String outOfBand)
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateNonRecoveringModel()
   at RabbitMQ.Client.Framing.Impl.AutorecoveringConnection.CreateModel()

以下是有关 my RabbitMQ 代理的一些信息: 版本:3.7.2 二郎:20.1 操作系统:Linux Ubuntu (AWS)

这是我的 RabbitMQConnection 类:

internal class RabbitMQConnection : IDisposable
{
    private object __syncRoot__ { get; } = new object();

    #region Private
    private bool __disposed__ { get; set; }
    private IConnectionFactory __rmqConnFactory__ { get; }
    private IConnection __rmqConnection__ { get; set; }
    private string __connectionIdentifier__ { get; }
    private ILog __logger__ { get; }
    #endregion

    #region Public
    public bool IsConnected
    {
        get
        {
            DisposeCheck();
            return __rmqConnection__ != null && __rmqConnection__.IsOpen && !__disposed__;
        }
    }

    public int ThreadLimit { get; private set; }

    public IConnection GetConnection
    {
        get
        {
            DisposeCheck();
            if (IsConnected == true)
            {
                return __rmqConnection__;
            }
            else
            {
                return null;
            }
        }
    }

    public bool IsDisposed
    {
        get
        {
            return __disposed__;
        }
    }
    #endregion

    public RabbitMQConnection(RabbitMQConfiguration _rmqConfig)
    {
        __logger__ = LogProvider.GetCurrentClassLogger();

        ConnectionFactory rMQconnFactory = new ConnectionFactory()
        {
            HostName = _rmqConfig.Hostname,
            UserName = _rmqConfig.Username,
            Password = _rmqConfig.Password,
            VirtualHost = _rmqConfig.VirtualHost,
            AutomaticRecoveryEnabled = true,
            DispatchConsumersAsync = true,
        };

        __rmqConnFactory__ = rMQconnFactory;

        ThreadLimit = _rmqConfig.ThreadLimit;

        __connectionIdentifier__ = $"{_rmqConfig.CurrentMachineIPAddress} - {_rmqConfig.CurrentMachineHostname}";

        Connect();

        __disposed__ = false;
    }

    private void Connect()
    {
        DisposeCheck();

        if (__rmqConnection__ == null)
        {
            lock (__syncRoot__)
            {
                if (__rmqConnection__ == null)
                {
                    __rmqConnection__ = __rmqConnFactory__.CreateConnection(__connectionIdentifier__);
                }
            }
        }
    }

    public IModel CreateChannel()
    {

        DisposeCheck();

        if (!IsConnected)
        {
            throw new InvalidOperationException("No RabbitMQ connections are available to perform this action");
        }

        return __rmqConnection__.CreateModel(); // Here is where the exception occurs!!!
    }

    #region ShutdownEvents
    private void OnConnectionBlocked(object sender, ConnectionBlockedEventArgs e)
    {
        if (__disposed__) return;

        //Logar algo
        __logger__.Warn("A RabbitMQ connection (OnConnectionBlocked) is shutdown. Trying to re-connect...");
    }

    void OnCallbackException(object sender, CallbackExceptionEventArgs e)
    {
        if (__disposed__) return;

        //Logar algo
        __logger__.Warn(e?.Exception, "A RabbitMQ connection (OnCallbackException) is shutdown. Trying to re-connect...");
    }

    void OnConnectionShutdown(object sender, ShutdownEventArgs reason)
    {
        if (__disposed__) return;

        //Logar algo
        __logger__.Warn("A RabbitMQ connection (OnConnectionShutdown) is on shutdown. Trying to re-connect...");
    }
    #endregion

    public void Dispose()
    {
        if (__disposed__ == true)
        {
            return;
        }

        try
        {
            __rmqConnection__?.Dispose();
        }
        catch (Exception ex)
        {
            //Log here
            __logger__.Fatal(ex, "RabbitMQ Connection: {0}", ex.Message);
        }
        finally
        {
            __disposed__ = true;
        }
    }

    private void DisposeCheck()
    {
        if (__disposed__ == true)
        {
            throw new ObjectDisposedException("RabbitMQConnection");
        }
    }
}

异常发生在行:

return __rmqConnection__.CreateModel();

有人知道为什么会这样吗？我知道有时 RabbitMQ 连接可能会发生波动，但我听说 RabbitMQ 会在发生这种情况时进行 native 重试。

Answer 1

我不知道它现在有多重要。 异常是 {System.TimeoutException} 操作已超时。 如果我们在连接工厂中删除 AutomaticRecoveryEnabled = true，则不会生成异常并创建 channel 。