Может ли Azure Service Bus задерживаться до повторной отправки сообщения?

Azure Service Bus поддерживает встроенный механизм повтора, который заставляет забытое сообщение сразу же отображаться для другой попытки чтения. Я пытаюсь использовать этот механизм для обработки некоторых временных ошибок, но сообщение становится доступным сразу же после его отказа.

Я бы хотел сделать сообщение невидимым в течение некоторого периода времени после его отказа, предпочтительно на основе экспоненциально увеличивающейся политики.

Я пытаюсь установить свойство ScheduledEnqueueTimeUtc при отказе от сообщения, но оно не похоже на эффект:

var messagingFactory = MessagingFactory.CreateFromConnectionString(...);

var receiver = messagingFactory.CreateMessageReceiver("test-queue");

receiver.OnMessageAsync(async brokeredMessage =>
{
    await brokeredMessage.AbandonAsync(
        new Dictionary<string, object>
        {
            { "ScheduledEnqueueTimeUtc", DateTime.UtcNow.AddSeconds(30) }
        });
    }
});

Я вообще не думал о том, чтобы отказаться от сообщения и просто разрешить блокировку, но для этого потребуется какой-то способ повлиять на то, как MessageReceiver указывает продолжительность блокировки сообщения, и я ничего не могу найти в API, чтобы я мог изменить это значение. Кроме того, было бы невозможно прочитать количество доставки сообщения (и, следовательно, принять решение о том, как долго ждать следующего повтора) до тех пор, пока уже не потребуется блокировка.

Можно ли каким-либо образом повлиять на политику повтора на шине сообщений или может быть искусственно введена задержка каким-то другим способом?

Ответ 1

Я действительно задал этот же вопрос в прошлом году (реализация в стороне) с тремя подходами, которые я мог бы придумать, глядя на API. @ClemensVasters, который работает в команде SB, ответил, что использование Defer с каким-то повторным получением действительно является единственным способом точно контролировать это.

Вы можете прочитать мой комментарий к его ответу для конкретного подхода к его выполнению, когда я предлагаю использовать дополнительную очередь для хранения сообщений, которые указывают, какие первичные сообщения были отложены, и их необходимо повторно получить из основной очереди. Затем вы можете контролировать, как долго вы будете ждать, установив ScheduledEnqueueTimeUtc на эти вторичные сообщения, чтобы точно контролировать, сколько времени вы ждете, прежде чем повторять попытку.

Ответ 2

Осторожно, потому что я думаю, что вы путаете функцию повтора с автоматическим механизмом Complete/Abandon для обработки сообщений, управляемых событиями OnMessage. Встроенный механизм повторного запуска вступает в игру, когда вызов на служебную шину выходит из строя. Например, если вы вызываете, чтобы установить сообщение как завершенное, а что - неудачно, то механизм повтора запустится. Если вы обрабатываете сообщение, в вашем собственном коде возникает исключение, которое НЕ будет запускать повтор через функцию повтора. Ваш вопрос не получает явного ответа, если ошибка связана с вашим кодом или при попытке связаться с служебной шиной.

Если вы действительно изменили политику повтора, возникающую при возникновении ошибки при попытке связаться с служебной шиной, вы можете изменить RetryPolicy, который установлен в самом MessageReciver. Существует RetryExponitial, который используется по умолчанию, а также абстрактный RetryPolicy, из которого вы можете создать свой собственный.

Я думаю, что вам нужно больше контролировать то, что происходит, когда вы получаете исключение, выполняющее вашу обработку, и вы хотите оттолкнуться от работы над этим сообщением. Есть несколько вариантов:

При создании обработчика сообщений вы можете настроить OnMessageOptions. Одним из свойств является "Автозаполнение". По умолчанию это значение равно true, что означает, что как только обработка для сообщения будет завершена, метод Complete вызывается автоматически. Если возникает исключение, автоматически вызывается отказ, который вы видите. Установив AutoComplete в значение false, вам необходимо вызвать Complete самостоятельно из обработчика сообщений. Несоблюдение этого требования приведет к тому, что блокировка сообщения в конечном итоге закончится, что является одним из способов, которые вы ищете.

Итак, вы можете написать свой обработчик, чтобы при возникновении исключения во время обработки вы просто не вызывали Complete. Затем сообщение останется в очереди до тех пор, пока не завершится блокировка, а затем снова станет доступной. Применяется стандартный механизм мертвой надписи, и после x количество попыток он будет автоматически помещен в очередь с ошибкой.

Осторожность обращения с этим способом заключается в том, что любой тип исключения будет рассматриваться таким образом. Вам действительно нужно подумать о том, какие типы исключений делают это, и если вы действительно хотите оттолкнуть обработку или нет. Например, если вы вызываете стороннюю систему во время обработки, и это дает вам исключение, которое, как вы знаете, преходящее, отлично. Если, однако, это дает вам ошибку, которая, как вы знаете, будет большой проблемой, тогда вы можете решить сделать что-то еще в системе, кроме того, что просто поругаете сообщение.

Вы также можете посмотреть метод "Defer". Этот метод на самом деле не позволит этому сообщению обрабатываться с очереди, если он специально не потянут его порядковым номером. Вы код должен помнить значение порядкового номера и тянуть его. Это не совсем то, что вы описали.

Другой вариант - вы можете отойти от OnMessage, управляемого событиями обработки сообщений. Хотя это очень полезно, вы не получаете большого контроля над вещами. Вместо этого подключите свой собственный цикл обработки и обработайте отказ/завершение самостоятельно. Вам также понадобится обработать некоторые потоковые/параллельные управления вызовами, которые дает шаблон OnMessage. Это может быть больше, но вы обладаете максимальной гибкостью.

Наконец, я считаю, что причина, по которой вызов, который вы сделали для AbandonAsync передачи свойств, которые вы хотели изменить, не работал, заключается в том, что эти свойства ссылаются на Свойства метаданных по методу, а не стандартные свойства в BrokeredMessage.