MES-ETL/ConsoleApp2/HostedServices/TransformService.cs

using ConsoleApp2.Const;
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using Microsoft.Extensions.Caching.Distributed;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using StackExchange.Redis;

namespace ConsoleApp2.HostedServices;

/// <summary>
/// 数据处理服务，对导入后的数据进行处理
/// </summary>
public class TransformService : ITransformService
{
    private readonly ILogger _logger;
    private readonly IOptions<DataTransformOptions> _options;
    private readonly DataRecordQueue _producerQueue;
    private readonly DataRecordQueue _consumerQueue;
    private readonly ProcessContext _context;
    private readonly IDistributedCache _cache;


    public TransformService(ILogger<TransformService> logger, 
        IOptions<DataTransformOptions> options,
        [FromKeyedServices(ProcessStep.Producer)]DataRecordQueue producerQueue,
        [FromKeyedServices(ProcessStep.Consumer)]DataRecordQueue consumerQueue, 
        ProcessContext context,
        IDistributedCache cache)
    {
        _logger = logger;
        _options = options;
        _producerQueue = producerQueue;
        _consumerQueue = consumerQueue;
        _context = context;
        _cache = cache;
    }

    public async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        _logger.LogInformation("***** Data transform service started, thread id: {ThreadId} *****", Environment.CurrentManagedThreadId);
        while ((!_context.IsInputCompleted || _producerQueue.Count > 0))
        {
            if (_context.GetExceptions().Count > 0)
            {
                _logger.LogInformation("***** Csv transform service is canceled *****");
                return;
            }
            // var dbOptions = _options.Value.DatabaseFilter(record);
            if (!_producerQueue.TryDequeue(out var record)) continue;
          
            //过滤不要的record
            if ( await _options.Value.RecordFilter?.Invoke(record, _cache) == false) continue;
            record.Database = _options.Value.DatabaseFilter?.Invoke(record);
            //修改record
            _options.Value.RecordModify?.Invoke(record);
            //缓存record
            _options.Value.RecordCache?.Invoke(record, _cache);
            //替换record
            var replaceRecord =await _options.Value.RecordReplace?.Invoke(record, _cache);
            if (replaceRecord != null)
            {
                record = replaceRecord;
            }
            _consumerQueue.Enqueue(record);
             _context.AddTransform();
            //数据增加
            var addRecords=_options.Value.RecordAdd?.Invoke(record);
            if(addRecords != null&& addRecords.Count>0)
            {
                foreach(var rc in addRecords)
                {
                    _consumerQueue.Enqueue(rc);
                    _context.AddTransform();
                }
            }
           
        }

        _context.CompleteTransform();
        _logger.LogInformation("***** Data transformation service completed *****");
    }
}