无法从 Azure 服务总线队列接收工作服务中的消息

Question

正在尝试实施 consumer/receiver 来自 Azure 队列的消息。我正在使用 worker service/backgroundservice 项目 .Net core 3.1.

以下是Program.cs

 public static IHostBuilder CreateHostBuilder(string[] args) =>
    Host.CreateDefaultBuilder(args)
        .ConfigureServices((hostContext, services) =>
        {
            services.AddSingleton<IQueueClient>(x => new QueueClient(hostContext.Configuration.GetConnectionString("ServiceBusConnectionString") ?? hostContext.Configuration["ServiceBusConnectionString"], hostContext.Configuration.GetValue<string>("QueueName")));
            services.AddHostedService<WorkerService>();
        });

以下是工作人员service.cs

public class WorkerService : BackgroundService
{
    private readonly IQueueClient queueClient;
    private readonly ILogger<PolCommunicationRequestQueueServiceBusSubscriber> logger;

    public WorkerService(IQueueClient queueClient, ILogger<WorkerService> logger)
    {
        this.queueClient = queueClient;
        this.logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            queueClient.RegisterMessageHandler(ProcessMessagesAsync
            , new MessageHandlerOptions(ExceptionReceivedHandler)
            {
                MaxConcurrentCalls = 1,
                AutoComplete = false
            });
        }
        catch (Exception ex)
        {
            logger.LogError(ex.Message);
        }
        finally
        {
            await queueClient.CloseAsync();
        }
    }

    public async Task ProcessMessagesAsync(Message message, CancellationToken token)
    {
        // Process the message
        logger.LogInformation($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");
        await queueClient.CompleteAsync(message.SystemProperties.LockToken);
    }

    public Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
    {
        logger.LogError($"Message handler encountered an exception {exceptionReceivedEventArgs.Exception}.");
        return Task.CompletedTask;
    }
  }
}

然而，该方法的一个变体有效，但它抛出错误“消息处理程序已被注册。”

while (!stoppingToken.IsCancellationRequested)
{
    queueClient.RegisterMessageHandler((Message message, CancellationToken cancellationToken) =>
    {
        logger.LogInformation($"Received message: SequenceNumber:{message.SystemProperties.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}");

                    //Implement DB/other logic here
        return queueClient.CompleteAsync(message.SystemProperties.LockToken);
    }, new MessageHandlerOptions(ExceptionReceivedHandler)
        {
            MaxConcurrentCalls = 1,
            AutoComplete = false
        });

    logger.LogInformation("Worker running at: {time}", DateTimeOffset.Now);

    await Task.Delay(1000, stoppingToken);
}

当我在控制台项目中实现它时，它起作用了。不确定这里缺少什么。

Answer 1

你应该使用StartAsync和StopAsync的BackgroundService。 StartAsync 注册消息处理程序。 StopAsync 关闭 queueClient。否则，发生的情况是您注册消息处理程序并立即关闭它，从而终止与代理的连接。

对于“有效”的变体，您有一个紧密的循环，因此客户端不会关闭。但是你 re-register 多次使用相同的消息处理程序，这是错误的。

无法从 Azure 服务总线队列接收工作服务中的消息

Unable to receive message in worker service from azure service bus queue

background-service

azureservicebus

azure-servicebus-queues

.net-core-3.1

ihostedservice