c# - remarks - El bloqueo suministrado no es válido. El bloqueo ha caducado o el mensaje ya se ha eliminado de la cola.

remarks c# documentation

Question

Estoy usando una cola de bus de Microsoft Azure Service para procesar los cálculos y mi programa funciona bien durante unas horas, pero luego comienzo a recibir esta excepción por cada mensaje que proceso a partir de ese momento. No tengo idea de por dónde empezar ya que todo funciona bien durante las primeras horas. Mi código parece ser preciso también. Publicaré el método en el que manejo el mensaje del bus de servicio de 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);
}
}

ACTUALIZACIÓN: verifico el tiempo hasta el vencimiento del bloqueo y llamo a renovar el bloqueo si lo necesita, pero renueva el bloqueo sin errores, pero sigo recibiendo esta excepción.

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

Todo lo que sé hasta ahora es que mi código se ejecuta bien por un tiempo y el bloqueo de renovación se ejecuta y funciona, pero todavía obtengo la excepción de bloqueo y dentro de esa excepción, imprimo el timeleft y sigue aumentando la diferencia de tiempo a medida que se ejecuta el código lo que me hace creer que el tiempo hasta la expiración del bloqueo no se está cambiando de alguna manera?

Answer 1

En lugar de renovar el bloqueo manualmente, al crear la suscripción del cliente, intente renovarla automáticamente utilizando las OnMessageOptions () del cliente, de la siguiente manera:

 OnMessageOptions options = new OnMessageOptions();
options.AutoRenewTimeout = TimeSpan.FromMinutes(1);
try
{
client = Subscription.CreateClient();
client.OnMessageAsync(MessageReceivedComplete, options);
}
catch (Exception ex)
{
throw new Exception (ex);
}

Answer 2

Estaba teniendo un problema similar. Los mensajes se manejaron con éxito, pero cuando se completaron, el Bus de Servicio ya no tenía un bloqueo válido. Resulta que mi TopicClient.PrefetchCount era demasiado alto.

Parece que el bloqueo comienza en todos los mensajes que aparecen previamente en cuanto se recuperan. Si el tiempo de procesamiento de su mensaje acumulativo supera el tiempo de espera de bloqueo, no se completará ningún otro mensaje que haya recibido previamente. Volverá al bus de servicio.

Answer 3

Pasé horas tratando de entender por qué estaba recibiendo una MessageLockLostException . La razón para mí se debió a que AutoComplete predeterminada en verdadero.

Si va a llamar a messsage.Complete() (o CompleteAsync() ), debe crear una instancia de un objeto OnMessageOptions , establecer AutoComplete en falso y pasarlo a su llamada OnMessage .

var options = new OnMessageOptions();
options.AutoComplete = false;
client.OnMessage(processCalculations, options);