MES-ETL/MesETL.App/HostedServices/MainHostedService.cs

using System.Diagnostics;
using System.Text;
using MesETL.App.Helpers;
using MesETL.App.HostedServices.Abstractions;
using MesETL.App.Options;
using MesETL.App.Services;
using MesETL.App.Services.ErrorRecorder;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using MySqlConnector;

namespace MesETL.App.HostedServices;

public class MainHostedService : BackgroundService
{
    private Stopwatch? _stopwatch;
    private readonly IInputService _input;
    private readonly ITransformService _transform;
    private readonly IOutputService _output;
    private readonly TaskMonitorService _taskMonitor;
    private readonly ILogger _logger;
    private readonly ProcessContext _context;
    
    private readonly IOptions<DatabaseOutputOptions> _databaseOptions;
    private readonly IOptions<TenantDbOptions> _tenantDbOptions;
    private readonly IConfiguration _config;

    public MainHostedService(IInputService input, 
        ITransformService transform,
        IOutputService output, 
        ILogger<MainHostedService> logger, 
        IOptions<TenantDbOptions> tenantDbOptions, 
        IOptions<DatabaseOutputOptions> databaseOptions, 
        IConfiguration config, 
        ProcessContext context, 
        TaskMonitorService taskMonitor)
    {
        _input = input;
        _transform = transform;
        _output = output;
        _logger = logger;
        _tenantDbOptions = tenantDbOptions;
        _databaseOptions = databaseOptions;
        _config = config;
        _context = context;
        _taskMonitor = taskMonitor;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _logger.LogInformation("Command argument detected, execute for each database");
        var command = _config["Command"];
        if (!string.IsNullOrEmpty(command))
        {
            _logger.LogInformation("***** Running Sql Command *****");
            await ExecuteEachDatabase(command, stoppingToken);
            Environment.Exit(0);
        }
        
        _stopwatch = Stopwatch.StartNew();
        await SetVariableAsync(); // 开启延迟写入，禁用重做日志 >>> 重做日志处于禁用状态时不要关闭数据库服务！

        var monitorTask = Task.Run(async () => await _taskMonitor.Monitor(stoppingToken), stoppingToken);
        var inputTask = ExecuteAndCatch(
            async () => await _input.ExecuteAsync(stoppingToken), "文件输入程序出现异常", stoppingToken);
        var transformTask = ExecuteAndCatch(
            async () => await _transform.ExecuteAsync(stoppingToken), "转换程序出现异常", stoppingToken);
        var outputTask = ExecuteAndCatch(
            async () => await _output.ExecuteAsync(stoppingToken), "输出程序出现异常", stoppingToken);

        await Task.WhenAll(inputTask, transformTask, outputTask);
        _stopwatch.Stop();
        _logger.LogInformation("***** All tasks completed *****");
        _logger.LogInformation("***** ElapseTime: {Time}", (_stopwatch.ElapsedMilliseconds / 1000f).ToString("F3"));
        await Task.Delay(5000, stoppingToken);

        await SetVariableAsync(false); // 关闭延迟写入，开启重做日志
        if (!stoppingToken.IsCancellationRequested)
        {
            await ExportResultAsync();
            _logger.LogInformation("The execution result export to {Path}",
                Path.Combine(AppDomain.CurrentDomain.BaseDirectory, $"Result-{ErrorRecorder.UID}.md"));

            Environment.Exit(0);
        }
        else Environment.Exit(1);
    }
    
    private Task ExecuteAndCatch(Func<Task> func, string message, CancellationToken ct)
    {
        return Task.Run(async () =>
        {
            try
            {
                await func();
            }
            catch (Exception e)
            {
                _logger.LogCritical(e, "{Msg}\t{ErrMsg}", message, e.Message);
                _context.AddException(e);
                Environment.Exit(1);
            }
        }, ct);
    }

    private async Task SetVariableAsync(bool enable = true)
    {
        var connStr = _databaseOptions.Value.ConnectionString
                      ?? throw new ApplicationException("分库配置中没有配置数据库");
        if (enable)
        {
            await DatabaseHelper.NonQueryAsync(connStr,
                """
                SET GLOBAL innodb_flush_log_at_trx_commit = 0;
                ALTER INSTANCE DISABLE INNODB REDO_LOG;
                """);
        }
        else
        {
            await DatabaseHelper.NonQueryAsync(connStr,
                """
                SET GLOBAL innodb_flush_log_at_trx_commit = 1;
                ALTER INSTANCE ENABLE INNODB REDO_LOG;
                """);
        }
    }

    private async Task ExecuteEachDatabase(string command, CancellationToken cancellationToken = default)
    {
        var databases = _tenantDbOptions.Value.DbGroup?.Keys 
                        ?? throw new ApplicationException("分库配置中没有配置数据库");
        var list = new List<Task>();
        foreach (var db in databases)
        {
            var connStr = new MySqlConnectionStringBuilder(_databaseOptions.Value.ConnectionString
                                                           ?? throw new ApplicationException("没有配置数据库连接字符串"))
            {
                ConnectionTimeout = 60,
                DefaultCommandTimeout = 0,
                Database = db
            }.ConnectionString;
            var task = Task.Run(async () => await DatabaseHelper.NonQueryAsync(connStr, command),
                cancellationToken);
            list.Add(task);
        }

        await Task.WhenAll(list);
    }

    private async Task ExportResultAsync()
    {
        var sb = new StringBuilder();
        if (_context.HasException)
            sb.AppendLine("# Program Completed With Error");
        else sb.AppendLine("# Program Completed Successfully");
        sb.AppendLine("## Process Count");
        var processCount = new[]
        {
            new { State = "Input", Count = _context.InputCount },
            new { State = "Transform", Count = _context.TransformCount },
            new { State = "Output", Count = _context.OutputCount }
        };
        sb.AppendLine(processCount.ToMarkdownTable());
        sb.AppendLine("\n---\n");
        sb.AppendLine("## Table Output Progress");
        var tableOutputProgress = _context.TableProgress.Select(pair =>
            new { Table = pair.Key, Count = pair.Value });
        sb.AppendLine(tableOutputProgress.ToMarkdownTable());
        sb.AppendLine("\n---\n");
        sb.AppendLine("## Result");
        var elapsedTime = (_stopwatch!.ElapsedMilliseconds / 1000f);
        var result = new[]
        {
            new { Field = "ElapsedTime", Value = elapsedTime.ToString("F2") },
            new
            {
                Field = "Average Output Speed",
                Value = (_context.OutputCount / elapsedTime).ToString("F2") + "records/s"
            }
        };
        sb.AppendLine(result.ToMarkdownTable());
        await File.WriteAllTextAsync(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, $"Result-{ErrorRecorder.UID}.md"),
            sb.ToString());
    }
}