MES-ETL/ConsoleApp2/HostedServices/MainHostedService.cs

using System.Diagnostics;
using System.Text;
using ConsoleApp2.Helpers;
using ConsoleApp2.Helpers.Database;
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using ConsoleApp2.Services.ErrorRecorder;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;

namespace ConsoleApp2.HostedServices;

public class MainHostedService : BackgroundService
{
    private Stopwatch? _stopwatch;
    private readonly IInputService _input;
    private readonly ITransformService _transform;
    private readonly IOutputService _output;
    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)
    {
        _input = input;
        _transform = transform;
        _output = output;
        _logger = logger;
        _tenantDbOptions = tenantDbOptions;
        _databaseOptions = databaseOptions;
        _config = config;
        _context = context;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        _stopwatch = Stopwatch.StartNew();
        await SetVariableAsync(); // 开启延迟写入，禁用重做日志 >>> 重做日志处于禁用状态时绝对不要关闭数据库服务！
        
        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"));
            if (_config["RestoreIndex"] is not null)
                await RestoreIndexAsync();

            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;
                """);
        }
    }
    
    /// <summary>
    /// 还原所有数据库的索引...
    /// </summary>
    /// <returns></returns>
    /// <exception cref="ApplicationException"></exception>
    private async Task RestoreIndexAsync()
    {
        var databases = _tenantDbOptions.Value.DbList?.Keys 
                        ?? throw new ApplicationException("无法还原索引，因为分库配置中没有配置数据库");
        var connStr = _databaseOptions.Value.ConnectionString 
                      ?? throw new ApplicationException("无法还原索引，因为没有配置数据库连接字符串");
        var list = new List<Task>();
        foreach(var db in databases)
        {
            var task = DatabaseHelper.NonQueryAsync(connStr + $";Database={db};",
                await File.ReadAllTextAsync(Path.Combine(AppDomain.CurrentDomain.BaseDirectory, "RestoreIndex.sql"))
                );
            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());
    }
}