c# - Azure Servicebus 主题连续接收器消息停止

问题描述

我构建了一个 Azure 服务总线消息提供程序以在具有核心 2.0 的 linux CENTOS 7 中运行。我正在使用包“Microsoft.Azure.ServiceBus 3.1.0”。但是，当我收到五十万（500 000）条或更多消息时，即使有许多活动消息要阅读，“RegisterMessageHandler”也会停止，不再收到。我有五个线程，“MaxConcurrentCall”有 100 个线程，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 }));
    }

有人知道继续运行或更新注册客户端的正确方法，还是其他方法？

标签： c#azureasync-await.net-core-2.0azure-servicebus-topics

c# - Azure Servicebus 主题连续接收器消息停止

问题描述

解决方案

推荐阅读