HealthMonitor/HealthMonitor.WebApi/Worker.cs

using HealthMonitor.Common;
using HealthMonitor.Core.Common.Extensions;
using HealthMonitor.Service.Sub;
using TDengineDriver;
using TDengineTMQ;

namespace HealthMonitor.WebApi
{
    public class Worker : BackgroundService
    {
        private readonly ILogger<Worker> _logger;
        private readonly IServiceProvider _services;
        private readonly TDengineDataSubcribe _tdEngineDataSubcribe;
        private readonly PackageProcess _processor;

        private CancellationTokenSource _tokenSource=default!;

        public Worker(ILogger<Worker> logger, IServiceProvider services, PackageProcess processor,TDengineDataSubcribe tdEngineDataSubcribe) 
        {
            _logger = logger;
            _tdEngineDataSubcribe = tdEngineDataSubcribe;
            _services = services;
            _processor = processor;
        }

        public override Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("------StartAsync");
            _tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            // 创建消费者
          //  _tdEngineDataSubcribe.CreateConsumer();
            return base.StartAsync(cancellationToken);
        }

        public override Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("------StopAsync");
            _tokenSource.Cancel();  //停止工作线程
            // 关闭消费者
          //  _tdEngineDataSubcribe.CloseConsumer();
            return base.StopAsync(cancellationToken);
        }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            // var processor = _services.GetService<PackageProcess>();
            TaskFactory factory = new(_tokenSource.Token);
            factory.StartNew(async () =>
            {
                if (_tokenSource.IsCancellationRequested)
                    _logger.LogWarning("Worker exit");

                _logger.LogInformation("------ResolveAsync");
                while (!_tokenSource.IsCancellationRequested)
                {
                  // 
                    await _processor.ResolveAsync().ConfigureAwait(false);
                    //  await  _tdEngineDataSubcribe.ProcessMsg();
                }

            }, TaskCreationOptions.LongRunning);

            factory.StartNew(() =>
            {
                _logger.LogInformation("------_tdEngineDataSubcribe");
                while (!_tokenSource.IsCancellationRequested)
                {
                   
                    _tdEngineDataSubcribe.BeginListen(_tokenSource.Token);
                }
            }, TaskCreationOptions.LongRunning);

            return Task.Delay(1000, _tokenSource.Token);

        }

    }
}