1

尝试从 Azure 队列实现消息的消费者/接收者。我正在使用工作人员服务/后台服务项目.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);
}

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

4

1 回答 1

1

您应该使用StartAsyncandStopAsync的BackgroundService. StartAsync注册消息处理程序。StopAsync关闭queueClient. 否则，您会注册消息处理程序并立即关闭它，从而终止与代理的连接。

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

于 2020-08-25T04:57:23.547 回答