Question

У нас странная проблема с реализацией Rabbit MQ.Мы используем контейнеры для масштабирования процессора.В настоящее время мы используем 2 контейнера в качестве пилотного теста.Это базовое консольное приложение, которое вызывает метод для обработки сообщений из очереди.

Полезная нагрузка, которую мы обрабатываем, имеет Guid.Мы замечаем, что один и тот же Guid несколько раз вытаскивается из очереди.Событие после подтверждения на сообщении.Это не должно происходить из нашего понимания кролика MQ.Это может быть связано с нашей текущей реализацией, мы используем C # RabbitMQClient Library.Кроме того, это может быть использование док-контейнеров для наших потребителей.Мы не смогли воспроизвести эту проблему в Prod.

Это происходит только в одном контейнере для Guid.Таким образом, мысль об этом связана с какой-то проблемой в самом процессоре.Если вам нужны более полные журналы, пожалуйста, спросите.

Текущая архитектура

Пара мыслей или идей

Возможно, Ack недостаточно быстро перед следующей попыткой удалить сообщение из очереди.
Что-то не так с нашей реализацией, на которую кто-то может указать.Мы пытаемся создать модель с несколькими очередями для одной очереди, чтобы обрабатывать сообщения быстрее.

Вопросы:

Что вы думаете о реализации сценария выше для опытного кролика?MQ там?
Что может происходить?(Оболочка кода без вызовов приведена ниже вместе с журналами примера)

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);
    }
}
}

Это журналы контейнеров и то, что мы видим.Вы можете увидеть несколько попыток одного и того же руководства по платежам, даже если оно прошло успешно.

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());
    }
}

theMayer · Answer 1 · 28 сентября 2018

Исходя из предоставленного вами кода, выглядит как проблема на стороне производителя.При этом рекомендуется, чтобы обработка сообщений была идемпотентной операцией.На самом деле, дизайн идемпотентности является критическим допущением практически любого внешнего интерфейса (и я бы сказал, что он одинаково важен для внутренних интерфейсов).

Даже если вам удастся найти и решить проблему наИздатель, вы должны помнить о том, что это не гарантирует доставку «ровно один раз».Никакой такой гарантии не может быть сделано.Вместо этого вы можете иметь одну из двух вещей (являющихся взаимоисключающими):

Максимум один раз доставки (0
Минимум один раз доставки (1 <=n) </li>

Из документации RabbitMQ :

Использование подтверждений гарантирует доставку хотя бы раз .Без подтверждений возможна потеря сообщения во время операций публикации и потребления, и гарантируется только доставка не более одного раза .

При публикации и использовании сообщений происходит несколько вещей.Из-за асинхронного характера систем обработки сообщений и, в частности, протокола AMQP, нет способа гарантировать обработку ровно один раз , при этом обеспечивая производительность, необходимую для системы обмена сообщениями (по сути, она заставляет всечерез последовательный процесс в точке дедупликации).Вы выбрали хотя бы один раз обработки, поэтому ваша система должна быть спроектирована таким образом, чтобы дубликат не приводил к нежелательному изменению состояния.

Кролик MQ тянет сообщение больше, чем один раз после подтверждения

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

1 Ответ

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Кролик MQ тянет сообщение больше, чем один раз после подтверждения

Пожалуйста, войдите или зарегистрируйтесь чтобы ответить на этот вопрос.

1 Ответ

Пожалуйста, войдите или зарегистрируйтесь что бы добавить комментарий.

Похожие темы