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 ProcessContext _context;
    private readonly IDistributedCache _cache;
    private readonly TaskManager _taskManager;


    public TransformService(ILogger<TransformService> logger, 
        IOptions<DataTransformOptions> options,
        ProcessContext context,
        IDistributedCache cache,
        TaskManager taskManager)
    {
        _logger = logger;
        _options = options;
        _context = context;
        _cache = cache;
        _taskManager = taskManager;
    }

    public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, DataRecordQueue consumerQueue, ProcessContext context, CancellationToken cancellationToken)
    {
        _logger.LogInformation("***** Data transform service started, thread id: {ThreadId} *****", Environment.CurrentManagedThreadId);

        _taskManager.CreateTasks(async () =>
        {
            while ((!context.IsInputCompleted || producerQueue.Count > 0))
            {
                if (_context.GetExceptions().Count > 0)
                {
                    _logger.LogInformation("***** Csv transform service is canceled *****");
                    return;
                }
                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
                await _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();
        },tasksOptions.TransformTaskCount,cancellationToken);
        _logger.LogInformation("***** Data transformation service completed *****");
    }
}