MES-ETL/ConsoleApp2/MySqlDestination.cs

using System.Text;
using ConsoleApp2.Entities;
using ConsoleApp2.Helpers;
using Microsoft.Extensions.Logging;
using MySqlConnector;

namespace ConsoleApp2;

public class MySqlDestination : IDisposable, IAsyncDisposable
{
    private readonly Dictionary<string, List<DataRecord>> _recordCache;
    private readonly MySqlConnection _conn;
    private readonly ILogger _logger;

    public static int AddCount;

    public MySqlDestination(string connStr, ILogger logger)
    {
        _conn = new MySqlConnection(connStr);
        _conn.Open();
        _recordCache = new Dictionary<string, List<DataRecord>>();
        _logger = logger;
    }

    public Task WriteRecordAsync(DataRecord record)
    {
        _recordCache.AddOrUpdate(record.TableName, [record], (key, value) =>
        {
            value.Add(record);
            Interlocked.Increment(ref AddCount);
            return value;
        });
        return Task.CompletedTask;
    }

    public async Task WriteRecordsAsync(IEnumerable<DataRecord> records)
    {
        foreach (var record in records)
        {
            await WriteRecordAsync(record);
        }
    }

    public async Task FlushAsync()
    {
        if (_recordCache.Count == 0)
            return;
        try
        {
            var cmd = _conn.CreateCommand();
            var sb = new StringBuilder();

            var count = 0;
            foreach (var (tableName, records) in _recordCache)
            {
                if (records.Count == 0)
                    continue;
                sb.Append($"INSERT INTO `{tableName}`(");
                for (var i = 0; i < records[0].Headers.Length; i++)
                {
                    var header = records[0].Headers[i];
                    sb.Append($"`{header}`");
                    if (i != records[0].Headers.Length - 1)
                        sb.Append(',');
                }

                sb.AppendLine(") VALUES");

                for (var i = 0; i < records.Count; i++)
                {
                    count++;
                    var record = records[i];
                    sb.Append('(');
                    for (var j = 0; j < record.Fields.Length; j++)
                    {
                        var field = record.Fields[j];

                        #region HandleFields

                        if (field == "\\N")
                            sb.Append("NULL");
                        else if (DumpDataHelper.CheckHexField(field)) // TODO: 性能消耗
                        {
                            if (StringExtensions.CheckJsonHex(field))
                                sb.Append($"UNHEX(\"{field}\")");
                            else
                                sb.Append($"\"{field}\"");
                        }
                        else
                            sb.Append($"\"{field}\"");

                        #endregion

                        if (j != record.Fields.Length - 1)
                            sb.Append(',');
                    }

                    sb.Append(')');

                    if (i != records.Count - 1) // not last field
                        sb.AppendLine(",");
                }

                sb.AppendLine(";\n");
            }

            cmd.CommandText = sb.ToString();

            await cmd.ExecuteNonQueryAsync();
            _recordCache.Clear();
        }
        catch (Exception e)
        {
            _logger.LogCritical(e, "Error when flushing records");
            throw;
        }
    }

    public void Dispose()
    {
        _conn.Dispose();
    }

    public async ValueTask DisposeAsync()
    {
        await _conn.DisposeAsync();
    }
}