您是否有任何指示如何确定何时发生订阅问题以便我可以重新连接？

我的服务使用 RabbitMQ.Client.MessagePatterns.Subscription 进行订阅。一段时间后，我的客户默默地停止接收消息。我怀疑是网络问题，因为我们的 VPN 连接不是最可靠的。

我已经通读了一段时间的文档，寻找一个 key 来查明此订阅何时可能因网络问题而中断，但运气不佳。我试过检查连接和 channel 是否仍然打开，但它似乎总是报告它仍然打开。

它处理的消息工作得很好，并被确认回队列，所以我认为这不是“确认”的问题。

我确定我一定只是遗漏了一些简单的东西，但我还没有找到它。

public void Run(string brokerUri, Action<byte[]> handler)
{
    log.Debug("Connecting to broker: {0}".Fill(brokerUri));
    ConnectionFactory factory = new ConnectionFactory { Uri = brokerUri };

    using (IConnection connection = factory.CreateConnection())
    {
        using (IModel channel = connection.CreateModel())
        {
            channel.QueueDeclare(queueName, true, false, false, null);

            using (Subscription subscription = new Subscription(channel, queueName, false))
            {
                while (!Cancelled)
                {
                    BasicDeliverEventArgs args;

                    if (!channel.IsOpen)
                    {
                        log.Error("The channel is no longer open, but we are still trying to process messages.");
                        throw new InvalidOperationException("Channel is closed.");
                    }
                    else if (!connection.IsOpen)
                    {
                        log.Error("The connection is no longer open, but we are still trying to process message.");
                        throw new InvalidOperationException("Connection is closed.");
                    }

                    bool gotMessage = subscription.Next(250, out args);

                    if (gotMessage)
                    {
                        log.Debug("Received message");
                        try
                        {
                            handler(args.Body);
                        }
                        catch (Exception e)
                        {
                            log.Debug("Exception caught while processing message. Will be bubbled up.", e);
                            throw;
                        }

                        log.Debug("Acknowledging message completion");
                        subscription.Ack(args);
                    }
                }
            }
        }
    }
}

更新:

我通过在虚拟机中运行服务器来模拟网络故障，当我断开连接足够长，所以也许这不是网络问题。现在我不知道它会是什么，但它在运行几个小时后就失败了。

最佳答案

编辑:由于我仍然对此表示赞同，我应该指出 .NET RabbitMQ 客户端现在内置了此功能:https://www.rabbitmq.com/dotnet-api-guide.html#connection-recovery

理想情况下，您应该能够使用它并避免手动实现重新连接逻辑。

我最近不得不实现几乎相同的事情。据我所知，关于 RabbitMQ 的大部分可用信息都假定您的网络非常可靠，或者您在与任何客户端发送或接收消息的同一台机器上运行 RabbitMQ 代理，从而允许 Rabbit 处理任何连接问题。

设置 Rabbit 客户端以防止连接丢失确实不难，但是您需要处理一些特殊情况。

您需要做的第一件事是打开心跳:

ConnectionFactory factory = new ConnectionFactory() 
{
  Uri = brokerUri,
  RequestedHeartbeat = 30,
}; 

将“RequestedHeartbeat”设置为 30 将使客户端每 30 秒检查一次连接是否仍然有效。如果不启用此功能，消息订阅者将愉快地坐在那里等待另一条消息进入，而不知道其连接已变坏。

打开心跳也会让服务器检查连接是否仍然有效，这可能非常重要。如果在订阅者接收消息后但在确认消息之前连接变坏，则服务器会假设客户端花费了很长时间，并且消息会“卡在”死连接上，直到它关闭。启用心跳后，服务器将识别连接何时变坏并关闭它，将消息放回队列中，以便另一个订阅者可以处理它。没有心跳，我不得不手动进入并关闭 Rabbit 管理 UI 中的连接，以便将卡住的消息传递给订阅者。

其次，您需要处理 OperationInterruptedException。正如您所注意到的，这通常是 Rabbit 客户端在注意到连接已中断时抛出的异常。如果在连接中断时调用 IModel.QueueDeclare()，您将得到此异常。通过处理您的订阅、 channel 和连接并创建新的来处理此异常。

最后，您必须处理消费者在尝试使用来自已关闭连接的消息时所做的操作。不幸的是，在 Rabbit 客户端中从队列中消费消息的每种不同方式似乎都有不同的 react 。如果您在关闭的连接上调用 QueueingBasicConsumer.Queue.Dequeue，QueueingBasicConsumer 将抛出 EndOfStreamException。 EventingBasicConsumer 什么都不做，因为它只是在等待消息。根据我的尝试，您正在使用的 Subscription 类似乎从对 Subscription.Next 的调用中返回 true，但是 args 的值 为空。再一次，通过处理您的连接、 channel 和订阅并重新创建它们来处理此问题。

connection.IsOpen 的值将在连接失败且心跳打开时更新为 False，因此您可以根据需要进行检查。但是，由于心跳在单独的线程上运行，您仍然需要处理连接在检查时打开但在调用 subscription.Next() 之前关闭的情况。

要注意的最后一件事是 IConnection.Dispose()。如果您在连接关闭后调用 dispose，此调用将抛出 EndOfStreamException。这对我来说似乎是一个错误，我不喜欢不在 IDisposable 对象上调用 dispose，所以我调用它并吞下异常。

将所有内容放在一个快速而肮脏的示例中:

public bool Cancelled { get; set; }

IConnection _connection = null;
IModel _channel = null;
Subscription _subscription = null;

public void Run(string brokerUri, string queueName, Action<byte[]> handler)
{
    ConnectionFactory factory = new ConnectionFactory() 
    {
        Uri = brokerUri,
        RequestedHeartbeat = 30,
    };

    while (!Cancelled)
    {               
        try
        {
            if(_subscription == null)
            {
                try
                {
                    _connection = factory.CreateConnection();
                }
                catch(BrokerUnreachableException)
                {
                    //You probably want to log the error and cancel after N tries, 
                    //otherwise start the loop over to try to connect again after a second or so.
                    continue;
                }

                _channel = _connection.CreateModel();
                _channel.QueueDeclare(queueName, true, false, false, null);
                _subscription = new Subscription(_channel, queueName, false);
            }

            BasicDeliverEventArgs args;
            bool gotMessage = _subscription.Next(250, out args);
            if (gotMessage)
            {
                if(args == null)
                {
                    //This means the connection is closed.
                    DisposeAllConnectionObjects();
                    continue;
                }

                handler(args.Body);
                _subscription.Ack(args);
            }
        }
        catch(OperationInterruptedException ex)
        {
            DisposeAllConnectionObjects();
        }
    }
    DisposeAllConnectionObjects();
}

private void DisposeAllConnectionObjects()
{
    if(_subscription != null)
    {
        //IDisposable is implemented explicitly for some reason.
        ((IDisposable)_subscription).Dispose();
        _subscription = null;
    }

    if(_channel != null)
    {
        _channel.Dispose();
        _channel = null;
    }

    if(_connection != null)
    {
        try
        {
            _connection.Dispose();
        }
        catch(EndOfStreamException) 
        {
        }
        _connection = null;
    }
}

关于c# - RabbitMQ C# 驱动程序停止接收消息，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/12499174/