Rabbitmq 在确认后不止一次拉消息

Question

我们在 Rabbit MQ 实现中遇到了一个奇怪的问题。我们正在使用容器来扩展处理器。我们目前正在运行 2 个容器作为试点测试。它是一个基本的控制台应用程序，它调用一个方法来处理来自队列的消息。

我们正在处理的负载有一个 Guid。我们注意到同一个 Guid 被多次从队列中拉出。对消息发出 Ack 后的事件。从我们对 rabbit MQ 的理解来看，这不应该发生。这可能与我们当前使用 C# RabbitMQClient 库的实现有关。此外，我们的消费者可能会使用 docker 容器。我们无法在 prod 中重现此问题。

这只发生在 Guid 的一个容器中。因此，这个想法与实际处理器本身的某些问题无关。如果您需要更完整的日志，请询问。

当前架构

一些想法或想法

• 可能在下一次尝试使消息出队之前，确认的速度不够快。
• 这里有人可以指出我们的实现出了点问题。我们正在尝试构建一个队列多消费者模型以更快地处理消息。

问题：

1. 对于您体验过的兔子 MQ 者，对上述场景的实施有何想法？
   
2. 会发生什么？ （下面是没有调用的代码外壳以及示例日志）

 

public class RabbitMQClient : IQueueClient
{
private IConnectionFactory _factory;
private IConnection _connection;
private ILoggerClient _logger;
private IWebApiClient _webApiClient;
private string _queueName;
private string _dlqName;
private string _rqName;
private int _maxRetryCount = 0;
private int _expiration = 0;
private decimal _expirationExponent = 0;

public RabbitMQClient(IConfigurationRoot config, ILoggerClient logger, IWebApiClient webApiClient)
{
    //Setup the ConnectionFactory
    _factory = new ConnectionFactory()
    {
        UserName = config["RabbitMQSettings:Username"],
        Password = config["RabbitMQSettings:Password"],
        VirtualHost = config["RabbitMQSettings:VirtualHost"],
        HostName = config["RabbitMQSettings:HostName"],
        Port = Convert.ToInt32(config["RabbitMQSettings:Port"]),
        AutomaticRecoveryEnabled = true,
        RequestedHeartbeat = 60,
        Ssl = new SslOption()
        {
            ServerName = config["RabbitMQSettings:HostName"],
            Version = SslProtocols.Tls12,
            CertPath = config["RabbitMQSettings:SSLCertPath"],
            CertPassphrase = config["RabbitMQSettings:SSLCertPassphrase"],
            Enabled = true
        }
    };

    _logger = logger;
    _webApiClient = webApiClient;

    _queueName = config["RabbitMQSettings:QueueName"];
    _dlqName = $"{_queueName}.dlq";
    _rqName = $"{_queueName}.rq";
    _maxRetryCount = int.Parse(config["RabbitMQSettings:MessageSettings:MaxRetryCount"]);
    _expiration = int.Parse(config["RabbitMQSettings:MessageSettings:Expiration"]);
    _expirationExponent = decimal.Parse(config["RabbitMQSettings:MessageSettings:ExpirationExponent"]);
}

public void ProcessMessages()
{
    using (_connection = _factory.CreateConnection())
    {
        using (var channel = _connection.CreateModel())
        {
            /*
             * Create the DLQ.
             * This is where messages will go after the retry limit has been hit.
             */
            channel.ExchangeDeclare(_dlqName, "direct");
            channel.QueueDeclare(_dlqName, true, false, false, null);
            channel.QueueBind(_dlqName, _dlqName, _queueName);

            /*
             * Create the main exchange/queue. we need to explicitly declare
             * the exchange so that we can push items back to it from the retry queue
             * once they're expired.
             */
            channel.ExchangeDeclare(_queueName, "direct");
            channel.QueueDeclare(_queueName, true, false, false, new Dictionary<String, Object>
            {
                { "x-dead-letter-exchange", _dlqName }
            });
            channel.QueueBind(_queueName, _queueName, _queueName);

            /*
             * Set the DLX of the retry queue to be the original queue
             * This is needed for the exponential backoff
             */
            channel.ExchangeDeclare(_rqName, "direct");
            channel.QueueDeclare(_rqName, true, false, false, new Dictionary<String, Object>
            {
                { "x-dead-letter-exchange", _queueName }
            });
            channel.QueueBind(_rqName, _rqName, _queueName);                    

            channel.BasicQos(0, 1, false);

            Subscription subscription = new Subscription(channel, _queueName, false);

            foreach (BasicDeliverEventArgs e in subscription)
            {
                Stopwatch stopWatch = new Stopwatch();
                try
                {
                    var payment = (CreditCardPaymentModel)e.Body.DeSerialize(typeof(CreditCardPaymentModel));

                    _logger.EventLog("Payment Dequeued", $"PaymentGuid:{payment.PaymentGuid}");

                    stopWatch.Start();

                    var response = //The Call to the Web API Happens here we will either get a 200 or a 400 from the WebService

                    stopWatch.Stop();

                    var elapsedTime = stopWatch.Elapsed.Seconds.ToString();

                    if (response.ResponseStatus == HttpStatusCode.BadRequest)
                    {
                        var errorMessage = $"PaymentGuid: {payment.PaymentGuid} | Elapsed Call Time: {elapsedTime} | ResponseStatus: {((int)response.ResponseStatus).ToString()}"
                                           + $"/n ErrorMessage: {response.ResponseErrorMessage}";
                        _logger.EventLog("Payment Not Processed", errorMessage);
                        Retry(e, subscription, errorMessage, payment.PaymentGuid);
                    }
                    else
                    {

                        //All the Responses are making it here. But even after the ACK they are being picked up and processoed again.
                        subscription.Ack(e);
                        _logger.EventLog("Payment Processed", $"--- Payment Processed - PaymentGuid : {payment.PaymentGuid} | Elapsed Call Time: {elapsedTime} | SourceStore : {payment.SourceStore} | Request Response: {(int)response.ResponseStatus}");
                    }
                }
                catch (Exception ex)
                {
                    Retry(e, subscription, ex.Message);
                    _logger.ErrorLog("Payment Not Processed", ex.ToString(), ErrorLogLevel.ERROR);
                }
            }
        }
    }
}

    public void Retry(BasicDeliverEventArgs payload, Subscription subscription, string errorMessage, Guid paymentGuid = new Guid())
    {

        if(paymentGuid != Guid.Empty)
        {
            _logger.EventLog("Retry Called", $"Retry on Payment Guid {paymentGuid}");
        }
        else
        {
            _logger.EventLog("Retry Called", errorMessage);
        }

        //Get or set the retryCount of the message
        IDictionary<String, object> headersDict = payload.BasicProperties.Headers ?? new Dictionary<String, object>();
        var retryCount = Convert.ToInt32(headersDict.GetValueOrDefault("x-retry-count"));

        //Check if the retryCount is still less than the max and republish the message
        if (retryCount < _maxRetryCount)
        {
            var originalExpiration = Convert.ToInt32(headersDict.GetValueOrDefault("x-expiration"));
            var newExpiration = Convert.ToInt32(originalExpiration == 0 ? _expiration : originalExpiration * _expirationExponent);

            payload.BasicProperties.Expiration = newExpiration.ToString();
            headersDict["x-expiration"] = newExpiration;
            headersDict["x-retry-count"] = ++retryCount;

            payload.BasicProperties.Headers = headersDict;

            subscription.Model.BasicPublish(_rqName, _queueName, payload.BasicProperties, payload.Body);
            subscription.Ack(payload);
        }
        else //Reject the message, which will send it to the DLX / DLQ
        {
            headersDict.Add("x-error-msg", errorMessage);
            payload.BasicProperties.Headers = headersDict;

            subscription.Nack(payload, false, false);
            _logger.ErrorLog("Error", errorMessage, ErrorLogLevel.ERROR);
        }
    }
}

public static class DictionaryExtensions
{
    public static TValue GetValueOrDefault<TKey, TValue>(this IDictionary<TKey, TValue> dic, TKey key)
    {
        return (dic != null && dic.TryGetValue(key, out TValue result)) ? result : default(TValue);
    }
}
}

这些是容器日志和我们所看到的。即使成功，您也可以看到多次拉取同一个付款 guid。

Container 1

Main
AutomaticPaymentQueue
1
EventName: Payment Dequeued | EventMessage: PaymentGuid:32d065a9-57e8-4359-afac-b7339b4904cc
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : 32d065a9-57e8-4359-afac-b7339b4904cc | Elapsed Call Time: 9 | SourceStore : C0222 | Request Response: 200
EventName: Payment Dequeued | EventMessage: PaymentGuid:65ad87a8-4cfe-47e8-863c-88e0c83fcd6f
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : 65ad87a8-4cfe-47e8-863c-88e0c83fcd6f | Elapsed Call Time: 2 | SourceStore : C0222 | Request Response: 200
EventName: Payment Dequeued | EventMessage: PaymentGuid:5dc2d38f-cbc9-492b-bd41-37531974c66d
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : 5dc2d38f-cbc9-492b-bd41-37531974c66d | Elapsed Call Time: 2 | SourceStore : C0222 | Request Response: 200
EventName: Payment Dequeued | EventMessage: PaymentGuid:5dc2d38f-cbc9-492b-bd41-37531974c66d
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : 5dc2d38f-cbc9-492b-bd41-37531974c66d | Elapsed Call Time: 1 | SourceStore : C0222 | Request Response: 200
EventName: Payment Dequeued | EventMessage: PaymentGuid:dad2616c-924d-4255-ad91-a262e3bcd245
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : dad2616c-924d-4255-ad91-a262e3bcd245 | Elapsed Call Time: 1 | SourceStore : C0222 | Request Response: 200

Container 2

Main
AutomaticPaymentQueue
1
EventName: Payment Dequeued | EventMessage: PaymentGuid:cb4fcb7a-48a7-422f-86d4-69c881366f05
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : cb4fcb7a-48a7-422f-86d4-69c881366f05 | Elapsed Call Time: 4 | SourceStore : C0222 | Request Response: 200
EventName: Payment Dequeued | EventMessage: PaymentGuid:5dc2d38f-cbc9-492b-bd41-37531974c66d
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : 5dc2d38f-cbc9-492b-bd41-37531974c66d | Elapsed Call Time: 2 | SourceStore : C0222 | Request Response: 200
EventName: Payment Dequeued | EventMessage: PaymentGuid:dad2616c-924d-4255-ad91-a262e3bcd245
EventName: Payment Processed | EventMessage: --- Payment Processed - PaymentGuid : dad2616c-924d-4255-ad91-a262e3bcd245 | Elapsed Call Time: 2 | SourceStore : C0222 | Request Response: 200

发布消息的类

public class RabbitMQClient : IQueueClient
{
    private static ConnectionFactory _factory;
    private static IConnection _connection;
    private static IModel _model;
    private const string QueueName = "AutomaticPaymentQueue";



    private void CreateConnection()
    {
        _factory = new ConnectionFactory();

        //Basic Login Infomration
        _factory.UserName = ConfigurationManager.AppSettings["RabbitMQUserName"]; ;
        _factory.Password = ConfigurationManager.AppSettings["RabbitMQPassword"];
        _factory.VirtualHost = ConfigurationManager.AppSettings["RabbitMQVirtualHost"];
        _factory.Port = Int32.Parse(ConfigurationManager.AppSettings["RabbitMQPort"]);


        //TLS Settings
        _factory.HostName = ConfigurationManager.AppSettings["RabbitMQHostName"];
        _factory.Ssl.ServerName = ConfigurationManager.AppSettings["RabbitMQHostName"];

        //SSL
        _factory.Ssl.Version = SslProtocols.Tls12;
        _factory.Ssl.CertPath = ConfigurationManager.AppSettings["RabbitMQSSLCertPath"];
        _factory.Ssl.CertPassphrase = ConfigurationManager.AppSettings["RabbitMQSSLCertPassphrase"];
        _factory.Ssl.Enabled = true;

        _connection = _factory.CreateConnection();
        _model = _connection.CreateModel();

    }

    public void SendMessage(Payload payload)
    {
         CreateConnection();
        _model.BasicPublish("", "AutomaticPaymentQueue", null, payload.Serialize());
    }
}

Answer 1

根据您提供的代码，看起来问题出在生产端。话虽如此，最佳实践是让消息处理成为idempotent 操作。事实上，幂等性设计是几乎所有外部接口的关键假设（我认为它在内部接口中同样重要）。

即使您确实设法找到并解决了发布商的问题，您也应该注意这样一个事实，即这并不能保证“仅一次”交付。不能作出这样的保证。相反，您可以拥有以下两种情况之一（相互排斥）：

• 最多一次交付 (0
• 至少一次交付 (1

来自RabbitMQ documentation：

使用确认保证至少一次交付。如果没有确认，则在发布和使用操作期间可能会丢失消息，并且只能保证最多一次传递。

在发布和使用消息时会发生一些事情。由于消息处理系统的异步特性，特别是 AMQP 协议，没有办法保证只处理一次同时仍然产生消息系统所需的性能（本质上，它强制一切都通过重复数据删除点的串行过程进行）。您已选择至少进行一次处理，因此您的系统需要设计为不会导致重复的状态更改。