问使用来自队列的Azure服务总线消息的Windows服务

EN

 public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .UseWindowsService()
                .ConfigureAppConfiguration((context, config) =>
                {

                    // configure the app here.
                })
                .ConfigureServices((hostContext, services) =>
                {                        
                    services.AddHostedService<QueueWorker>();
                }).UseSerilog();
    }

public class QueueWorker : BackgroundService, IDisposable
    {
        protected ILogger<QueueWorker> _logger;
        protected IQueueMessageReceiver _queueProcessor;

        public QueueWorker()
        {

        }

        public QueueWorker(ILogger<QueueWorker> logger, IQueueMessageReceiver queueMessageReceiver)
        {
            _logger = logger;
            _queueProcessor = queueMessageReceiver;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            await Task.CompletedTask.ConfigureAwait(false);
        }

        public override Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Service Starting");
            var task = _queueProcessor.StartProcessor(cancellationToken);
            task.Wait();
            if (task.IsFaulted)
            {
                throw new Exception("Unable to start Processor");
            }
            return base.StartAsync(cancellationToken);
        }

        public override async Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("Stopping Service");
            await _queueProcessor.StopProcessor().ConfigureAwait(false);
            await base.StopAsync(cancellationToken).ConfigureAwait(false);
        }

        public override void Dispose()
        {
            _logger.LogInformation("Disposing Service");
            var loopCount = 0;

            while (_queueProcessor != null && !_queueProcessor.IsClosedOrClosing() && loopCount < 5)
            {
                var task = Task.Delay(600);
                task.Wait();
                loopCount++;
            }
            base.Dispose();
            GC.SuppressFinalize(this);
        }

public class QueueMessageReceiver : IQueueMessageReceiver
    {
        private readonly ServiceBusClient _queueClient;
        private ServiceBusProcessor _processor;
        private readonly ReceiverConfiguration _configuration;
        private readonly ILogger _logger;
        private readonly ILoggerFactory _loggerFactory;
        private Dictionary<string, string> _executionMatrix;
        private readonly IServiceProvider _provider;
        private CancellationToken _cancellationToken;

        public QueueMessageReceiver(ReceiverConfiguration configuration, ILogger<QueueMessageReceiver> logger, IExecutionMatrix executionMatrix, ILoggerFactory loggerFactory, IServiceProvider serviceProvider)
        {
            if (configuration == null) throw new ArgumentException($"Configuration is missing from the expected ");
            _configuration = configuration;
            _logger = logger;
            _loggerFactory = loggerFactory;
            _executionMatrix = executionMatrix.GetExecutionMatrix();
            _provider = serviceProvider;
            _queueClient = new ServiceBusClient(_configuration.ConnectionString);
            if (string.IsNullOrWhiteSpace(configuration.ConnectionString)) throw new ArgumentException($"ServiceBusConnectionString Object missing from the expected configuration under ConnectionStrings ");

            if (configuration.QueueName == null) throw new ArgumentException($"Queue Name value missing from the expected configuration");
        }

        public async Task StartProcessor(CancellationToken cancellationToken)
        {
            if (!IsClosedOrClosing())
            {
                throw new FatalSystemException("ServiceBusProcessor is already running. ");
            }
            _cancellationToken = cancellationToken;
            var options = new ServiceBusProcessorOptions
            {
                AutoCompleteMessages = _configuration.AutoComplete,
                MaxConcurrentCalls = _configuration.MaxConcurrentCalls,
                MaxAutoLockRenewalDuration = _configuration.MaxAutoRenewDuration
            };
            _processor = _queueClient.CreateProcessor(_configuration.QueueName, options);
            _processor.ProcessMessageAsync += ProcessMessagesAsync;
            _processor.ProcessErrorAsync += ProcessErrorAsync;
            await _processor.StartProcessingAsync().ConfigureAwait(false);
        }
        public async Task StopProcessor()
        {
            await _processor.StopProcessingAsync();
            await _processor.CloseAsync();
        }
    
        private Task ProcessErrorAsync(ProcessErrorEventArgs args)
        {
            _logger.LogError(args.Exception, "Uncaught handled exception", args.ErrorSource, args.FullyQualifiedNamespace, args.EntityPath);
            return Task.CompletedTask;
        }
        private async Task ProcessMessagesAsync(ProcessMessageEventArgs args)
        {
            var message = args.Message;
            // Process the message.
            var sbMessage = $"Received message: SequenceNumber:{message.SequenceNumber} Body:{Encoding.UTF8.GetString(message.Body)}";
            _logger.LogInformation(sbMessage);

          //Handle your message
        }

        public bool IsClosedOrClosing()
        {
            return ((_processor == null) || _processor.IsClosed || !_processor.IsProcessing);
        }
    }