Я собрал провайдера сообщений служебной шины Azure для работы в linux CENTOS 7 с ядром 2.0.Я использую пакет "Microsoft.Azure.ServiceBus 3.1.0".Но когда я получаю пятьсот тысяч (500 000) или более сообщений, «RegisterMessageHandler» останавливается даже при чтении многих активных сообщений, больше не получает.У меня есть пять потоков: «MaxConcurrentCall» с сотней, Prefechcount 100 и «MaxAutoRenewDuration» за пять минут.
Этот метод использует «RegisterMessageHandler» для непрерывного чтения подписки.
public async Task RecebeMensagemAsync(CancellationToken cancellationToken, Action<string> tratarObjetoImportado, int prefetchCount)
{
try
{
var receiver = new Microsoft.Azure.ServiceBus.SubscriptionClient(_serviceBusConnString, _topic, _subscription, this._receiveMode, new RetryExponential(TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(10), _retryPolicyMaximumRetryCount));
receiver.PrefetchCount = prefetchCount;
receiver.OperationTimeout = TimeSpan.FromMinutes(_timoutOperacao);
var cicloRecebimentoCompleto = new TaskCompletionSource<bool>();
// fecha receiver e factory quando o CancellationToken é acionado
cancellationToken.Register(
async () =>
{
await receiver.CloseAsync();
cicloRecebimentoCompleto.SetResult(true);
});
// registra o RegisterMessageHandler para tratar mensagem lida assincronamente
receiver.RegisterMessageHandler(
async (message, cancellationTokenResponse) =>
{
if (message.Body == null || message.Body.Length == 0)
{
GerenciadorLog.LogInfo($"Mensagem não possui conteudo, mensageId: {message.MessageId}", LogType.Warning);
await receiver.CompleteAsync(message.SystemProperties.LockToken);
}
else
{
try
{
tratarObjetoImportado(Encoding.UTF8.GetString(message.Body));
if (_receiveMode == ReceiveMode.PeekLock)
await receiver.CompleteAsync(message.SystemProperties.LockToken);
}
catch (MessageSizeExceededException ex) {
GerenciadorLog.LogError($"Tamanho da mensagem:{message.MessageId}", ex);
try
{
if (_receiveMode == ReceiveMode.PeekLock)
await receiver.DeadLetterAsync(message.SystemProperties.LockToken);
}
catch (Exception dlex)
{
GerenciadorLog.LogError($"Erro ao enviar mensagem para 'DeadLetter', mensageId: {message.MessageId}", dlex);
throw new Exception("Erro não identificado", dlex);
}
}
catch (MessageLockLostException ex)
{
GerenciadorLog.LogError($"Expirou tempo de leitura da mensagem, será enviada para DeadLetter:{message.MessageId}", ex);
try
{
await receiver.AbandonAsync(message.SystemProperties.LockToken);
}
catch (Exception ex2)
{
GerenciadorLog.LogError($"Erro ao abandonar mensagem, mensageId:{message.MessageId}", ex2);
await Task.CompletedTask;
}
}
catch (MessagingEntityNotFoundException ex)
{
GerenciadorLog.LogError($"Mensagem não encontrada, mensageId:{message.MessageId}", ex);
try
{
await receiver.AbandonAsync(message.SystemProperties.LockToken);
}
catch (Exception ex2)
{
GerenciadorLog.LogError($"Erro ao abandonar mensagem, mensageId:{message.MessageId}", ex2);
await Task.CompletedTask;
}
}
catch (Exception ex)
{
if (ex != null && !string.IsNullOrEmpty(ex.Message))
GerenciadorLog.LogError($"Erro na importação, mensageId:{message.MessageId}", ex);
try
{
if (_receiveMode == ReceiveMode.PeekLock)
await receiver.DeadLetterAsync(message.SystemProperties.LockToken);
}
catch (Exception dlex)
{
GerenciadorLog.LogError($"Erro ao enviar mensagem para 'DeadLetter', mensageId: {message.MessageId}", dlex);
await Task.CompletedTask;
}
}
}
},
new MessageHandlerOptions((e) => LogMessageHandlerException(e, receiver)) { AutoComplete = false, MaxConcurrentCalls = _maxConcurrentCalls, MaxAutoRenewDuration = TimeSpan.FromSeconds(_lockDuration) });
await cicloRecebimentoCompleto.Task;
}
catch (Exception ex)
{
GerenciadorLog.LogError("Erro ao configurar 'listener' para leitura das mensagens.");
GerenciadorLog.LogError(ex.Message, ex);
throw ex;
}
}
И этот метод продолжает ждать задачи.Может быть несколько потоков, вызывающих этот метод
public async Task Run(int segundosIntnervaloSondaConsultaTopico, int prefetchCount, Action<string> tratarObjetoImportado)
{
IList<CancellationTokenSource> cancellationTokensSource = new List<CancellationTokenSource>();
IList<Task> instanciasDasSondas = new List<Task>();
CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
GerenciadorLog.LogInfo($"start blablalbala --");
var recebeMensagemAsync = RecebeMensagemAsync(cancellationTokenSource.Token, tratarObjetoImportado, prefetchCount);
await Task.WhenAll(Task.Run(() => Console.ReadKey()),
Task.Delay(TimeSpan.FromSeconds(segundosIntnervaloSondaConsultaTopico)).ContinueWith((t) => cancellationTokenSource),
Task.WhenAll(new Task[] { recebeMensagemAsync }));
}
Кто-нибудь знает, как правильно запустить или обновить клиент реестра, или другой подход?