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

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

public static async Task processCalculations(BrokeredMessage message)
    {
        try
        {
            if (message != null)
            {
                if (connection == null || !connection.IsConnected)
                {
                    connection = await ConnectionMultiplexer.ConnectAsync("connection,SyncTimeout=10000,ConnectTimeout=10000");
                    //connection = ConnectionMultiplexer.Connect("connection,SyncTimeout=10000,ConnectTimeout=10000");
                }

                cache = connection.GetDatabase();

                string sandpKey = message.Properties["sandp"].ToString();
                string dateKey = message.Properties["date"].ToString();
                string symbolclassKey = message.Properties["symbolclass"].ToString();
                string stockdataKey = message.Properties["stockdata"].ToString();
                string stockcomparedataKey = message.Properties["stockcomparedata"].ToString();

                var sandpTask = cache.GetAsync<List<StockData>>(sandpKey);
                var dateTask = cache.GetAsync<DateTime>(dateKey);
                var symbolinfoTask = cache.GetAsync<SymbolInfo>(symbolclassKey);
                var stockdataTask = cache.GetAsync<List<StockData>>(stockdataKey);
                var stockcomparedataTask = cache.GetAsync<List<StockMarketCompare>>(stockcomparedataKey);

                await Task.WhenAll(sandpTask, dateTask, symbolinfoTask,
                    stockdataTask, stockcomparedataTask);

                List<StockData> sandp = sandpTask.Result;
                DateTime date = dateTask.Result;
                SymbolInfo symbolinfo = symbolinfoTask.Result;
                List<StockData> stockdata = stockdataTask.Result;
                List<StockMarketCompare> stockcomparedata = stockcomparedataTask.Result;

                StockRating rating = performCalculations(symbolinfo, date, sandp, stockdata, stockcomparedata);

                if (rating != null)
                {
                    saveToTable(rating);
                    if (message.LockedUntilUtc.Minute <= 1)
                    {
                        await message.RenewLockAsync();
                    }
                    await message.CompleteAsync(); // getting exception here
                }
                else
                {
                    Console.WriteLine("Message " + message.MessageId + " Completed!");
                    await message.CompleteAsync();
                }
            }
        }
        catch (TimeoutException time)
        {
            Console.WriteLine(time.Message);
        }
        catch (MessageLockLostException locks)
        {
            Console.WriteLine(locks.Message);
        }
        catch (RedisConnectionException redis)
        {
            Console.WriteLine("Start the redis server service!");
        }
        catch (MessagingCommunicationException communication)
        {
            Console.WriteLine(communication.Message);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
            Console.WriteLine(ex.StackTrace);
        }
    }

UPDATE: я проверяю время до истечения срока действия блокировки, и я вызываю блокировку, если он нужен, но он обновляет блокировку без ошибок, но я все еще получаю это исключение.

timeLeft = message.LockedUntilUtc - DateTime.UtcNow;
  if (timeLeft.TotalMinutes <= 2)
                    {
                        //Console.WriteLine("Renewed lock! " + ((TimeSpan)(message.LockedUntilUtc - DateTime.UtcNow)).TotalMinutes);
                        message.RenewLock();
                    }

catch (MessageLockLostException locks)
        {
            Console.WriteLine("Delivery Count: " + message.DeliveryCount);
            Console.WriteLine("Enqueued Time: " + message.EnqueuedTimeUtc);
            Console.WriteLine("Expires Time: " + message.ExpiresAtUtc);
            Console.WriteLine("Locked Until Time: " + message.LockedUntilUtc);
            Console.WriteLine("Scheduled Enqueue Time: " + message.ScheduledEnqueueTimeUtc);
            Console.WriteLine("Current Time: " + DateTime.UtcNow);
            Console.WriteLine("Time Left: " + timeLeft);
        }

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

Ответ 1

Я потратил часы, пытаясь понять, почему я получаю MessageLockLostException. Причиной для меня было то, что AutoComplete по умолчанию имеет значение true.

Если вы собираетесь вызывать messsage.Complete() (или CompleteAsync()), вам следует создать экземпляр объекта OnMessageOptions, установить для AutoComplete значение false и передать его в вызов OnMessage.

var options = new OnMessageOptions();
options.AutoComplete = false;

client.OnMessage(processCalculations, options);

Ответ 2

У меня была аналогичная проблема. Сообщения обрабатывались успешно, но когда они пошли на завершение, служебная шина больше не имела действительного замка. Оказывается, мой ThemeClient.PrefetchCount был слишком высоким.

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

Ответ 3

Вместо того, чтобы обновлять блокировку вручную, при создании клиентской подписки попробуйте автоматически обновить ее с помощью клиента OnMessageOptions() следующим образом:

        OnMessageOptions options = new OnMessageOptions();

        options.AutoRenewTimeout = TimeSpan.FromMinutes(1);

        try
        {
            client = Subscription.CreateClient();

            client.OnMessageAsync(MessageReceivedComplete, options);
        }
        catch (Exception ex)
        {
            throw new Exception (ex);
        }

Ответ 4

Я получал эту ошибку в своей Azure Function с ServiceBusTrigger. Мой код был, как показано ниже.

[FunctionName("FetchDataFromSubscription")]
public void Run([ServiceBusTrigger(_topicName, _subscriptionName, Connection = "ServiceBusConnectionString")] string mySbMsg, ILogger log) {
    _logger = log;
    PerformValidations(mySbMsg);
}

private void PerformValidations(string mySbMsg) {
    var data = JsonConvert.DeserializeObject < BaseArticleAttributes > (mySbMsg);
    new VerifyModel(_httpClient).ValidateAsync(data);
    _logger.LogInformation($ "Performed Validation");
}

В этой строке var data = JsonConvert.DeserializeObject < BaseArticleAttributes > (mySbMsg); Я получаю сообщение об ошибке: "Предоставленная блокировка недействительна. Либо истек срок действия блокировки, либо сообщение уже удалено из очереди". Все, что мне нужно было сделать, это ввести lock чтобы я мог заблокировать mySbMsg и десериализовать то же самое для объекта.

private void PerformValidations(string mySbMsg) {
    lock(mySbMsg) {
        var data = JsonConvert.DeserializeObject < BaseArticleAttributes > (mySbMsg);
        new VerifyModel(_httpClient).ValidateAsync(data);
    }
    _logger.LogInformation($ "Performed Validation");
}