Merge remote-tracking branch 'origin/multTasks'

# Conflicts:
#	ConsoleApp2/HostedServices/OutputService.cs
#	ConsoleApp2/HostedServices/TransformService.cs
#	ConsoleApp2/Program.cs

ConsoleApp2/ConsoleApp2.csproj

@@ -25,6 +25,7 @@
       <PackageReference Include="Serilog" Version="3.1.2-dev-02097" />
       <PackageReference Include="Serilog.Extensions.Hosting" Version="8.0.0" />
       <PackageReference Include="Serilog.Sinks.Console" Version="5.0.1" />
+      <PackageReference Include="Serilog.Sinks.File" Version="5.0.1-dev-00972" />
       <PackageReference Include="ServiceStack.Text" Version="8.0.0" />
       <PackageReference Include="ZstdSharp.Port" Version="0.7.4" />
     </ItemGroup>

ConsoleApp2/DataRecord.cs

@@ -65,6 +65,8 @@ public class DataRecord
 
     public bool SetField(string columnName,  string value) => SetField(this, columnName,value);
 
+    public string GetCacheKey(string columnName) => GetCacheKey(this, columnName);
+
     public  bool SetField( DataRecord record,string columnName,string value)
     {
         if (record.Headers is null)
@@ -75,4 +77,14 @@ public class DataRecord
         record.Fields[idx] = value;
         return true;
     }
+    public string GetCacheKey(DataRecord record, string columnName)
+    {
+        if (TryGetField(record, columnName, out var  value))
+        {
+            return $"{TableName}_{value}";
+        }else
+            throw new IndexOutOfRangeException($"Column name:{columnName} not found in this record.");
+
+
+    }
 }

ConsoleApp2/HostedServices/Abstractions/IInputService.cs

@@ -1,6 +1,9 @@
-namespace ConsoleApp2.HostedServices.Abstractions;
+using ConsoleApp2.Options;
+using ConsoleApp2.Services;
+
+namespace ConsoleApp2.HostedServices.Abstractions;
 
 public interface IInputService
 {
-    public Task ExecuteAsync(CancellationToken cancellationToken);
+    public Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, ProcessContext context, CancellationToken cancellationToken);
 }

ConsoleApp2/HostedServices/Abstractions/IOutputService.cs

@@ -1,6 +1,9 @@
-namespace ConsoleApp2.HostedServices.Abstractions;
+using ConsoleApp2.Options;
+using ConsoleApp2.Services;
+
+namespace ConsoleApp2.HostedServices.Abstractions;
 
 public interface IOutputService
 {
-    public Task ExecuteAsync(CancellationToken cancellationToken);
+    public Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue consumerQueue, ProcessContext context, CancellationToken cancellationToken);
 }

ConsoleApp2/HostedServices/Abstractions/ITransformService.cs

@@ -1,6 +1,9 @@
-namespace ConsoleApp2.HostedServices.Abstractions;
+using ConsoleApp2.Options;
+using ConsoleApp2.Services;
+
+namespace ConsoleApp2.HostedServices.Abstractions;
 
 public interface ITransformService
 {
-    public Task ExecuteAsync(CancellationToken cancellationToken);
+    public Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, DataRecordQueue consumerQueue, ProcessContext context, CancellationToken cancellationToken);
 }

ConsoleApp2/HostedServices/InputService.cs

@@ -16,24 +16,18 @@ public class InputService : IInputService
 {
     private readonly ILogger _logger;
     private readonly IOptions<DataInputOptions> _dataInputOptions;
-    private readonly IOptions<InputTableOptions> _tableOptions;
-    private readonly DataRecordQueue _producerQueue;
     private readonly ProcessContext _context;
 
     public InputService(ILogger<InputService> logger,
             IOptions<DataInputOptions> dataInputOptions,
-            IOptions<InputTableOptions> tableOptions,
-            [FromKeyedServices(ProcessStep.Producer)] DataRecordQueue producerQueue,
             ProcessContext context)
     {
         _logger = logger;
         _dataInputOptions = dataInputOptions;
-        _tableOptions = tableOptions;
-        _producerQueue = producerQueue;
         _context = context;
     }
 
-    public async Task ExecuteAsync(CancellationToken cancellationToken)
+    public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, ProcessContext context,CancellationToken cancellationToken)
     {
         var inputDir = _dataInputOptions.Value.InputDir;
         _logger.LogInformation("***** Csv input service start, working dir: {InputDir}, thread id: {ThreadId} *****", inputDir, Environment.CurrentManagedThreadId);
@@ -44,14 +38,14 @@ public class InputService : IInputService
             return;
         }
         var count = 0;
-        foreach (var tableName in _tableOptions.Value.TableInfoConfig.Keys)
+        foreach (var tableName in tasksOptions.TableInfoConfig.Keys)
         {
             _logger.LogInformation("Working table: {tableName}", tableName);
             var source = _dataInputOptions.Value.CreateSource?.Invoke(tableName);
             await source.DoEnqueue((record) =>
             {
                 _context.AddInput();
-                _producerQueue.Enqueue(record);
+                producerQueue.Enqueue(record);
                 count++;
 
             });
@@ -63,7 +57,7 @@ public class InputService : IInputService
             _logger.LogInformation("table:'{tableName}' input completed", tableName);
         }
 
-        _context.CompleteInput();
+        context.CompleteInput();
         _logger.LogInformation("***** Csv input service completed *****");
     }
 }

ConsoleApp2/HostedServices/MainHostedService.cs

@@ -1,9 +1,8 @@
 using ConsoleApp2.HostedServices.Abstractions;
+using ConsoleApp2.Options;
 using ConsoleApp2.Services;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
-using System.Threading.Tasks;
-
 namespace ConsoleApp2.HostedServices;
 
 public class MainHostedService : BackgroundService
@@ -25,49 +24,87 @@ public class MainHostedService : BackgroundService
 
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-
+        var taskFun = (TasksOptions taskOp, DataRecordQueue producerQueue, DataRecordQueue consumerQueue, ProcessContext context) =>
-
-        var inputTask = Task.Factory.StartNew(async () =>
-         {
-             try
-             {
-                 await _input.ExecuteAsync(stoppingToken);
-             }
-             catch (Exception ex)
-             {
-                 _context.AddException(ex);
-                 _logger.LogError("Exception occurred on inputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
-             }
-
-         });
-        var transformTask = Task.Factory.StartNew(async () =>
         {
-            try
+            var inputTask = Task.Factory.StartNew(async () =>
             {
-                await _transform.ExecuteAsync(stoppingToken);
+                try
-            }
+                {
-            catch (Exception ex)
+                    await _input.ExecuteAsync(taskOp, producerQueue, context, stoppingToken);
+                }
+                catch (Exception ex)
+                {
+                    _context.AddException(ex);
+                    _logger.LogError("Exception occurred on inputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
+                }
+
+            });
+            var transformTask = Task.Factory.StartNew(async () =>
             {
-                _context.AddException(ex);
+                try
-                _logger.LogError("Exception occurred on transformService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
+                {
-            }
+                    await _transform.ExecuteAsync(taskOp, producerQueue, consumerQueue, context, stoppingToken);
+                }
+                catch (Exception ex)
+                {
+                    _context.AddException(ex);
+                    _logger.LogError("Exception occurred on transformService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
+                }
 
-        });
+            });
+            var outputTask = Task.Factory.StartNew(async () =>
+            {
+                try
+                {
+                    await _output.ExecuteAsync(taskOp, consumerQueue, context,stoppingToken);
+                }
+                catch (Exception ex)
+                {
+                    _context.AddException(ex);
+                    _logger.LogError("Exception occurred on outputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
+                }
 
-        var outputTask = Task.Factory.StartNew(async () =>
+            });
+        };
+
+        var bigTablesDic = new Dictionary<string, TableInfo>
         {
-            try
+            {"order_block_plan",new TableInfo{SimulaRowCount=2725553 }},//CreateTime < 202301的删除 
-            {
+            {"order_block_plan_result",new TableInfo{SimulaRowCount=1174096 }},
-                await _output.ExecuteAsync(stoppingToken);
+            {"order_box_block",new TableInfo{SimulaRowCount=29755672 }},
-            }
+            {"order_item",new TableInfo{SimulaRowCount=1345520079 }},
-            catch (Exception ex)
+            {"simple_plan_order",new TableInfo{SimulaRowCount=351470 }},//CreateTime < 202301的删除
-            {
+        };
-                _context.AddException(ex);
+        taskFun(new TasksOptions { TableInfoConfig = bigTablesDic, OutPutOptions = new OutPutOptions { FlushCount = 10000, OutPutTaskCount = 2 } },
-                _logger.LogError("Exception occurred on outputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
+            new DataRecordQueue(), new DataRecordQueue(),new ProcessContext());
-            }
+        var smallTablesDic = new Dictionary<string, TableInfo>
-
+        {
-        });
+            {"machine",new TableInfo{SimulaRowCount=14655 }},
-
+            {"order",new TableInfo{SimulaRowCount=5019216 }},
-        // await Task.Run(async () => await _output.ExecuteAsync(stoppingToken), stoppingToken);
+            {"order_data_block",new TableInfo{SimulaRowCount=731800334 }},
+            {"order_data_goods",new TableInfo{SimulaRowCount=25803671 }},
+            {"order_data_parts",new TableInfo{SimulaRowCount=468517543 }},
+            {"order_module",new TableInfo{SimulaRowCount=103325385 }},
+            {"order_module_extra",new TableInfo{SimulaRowCount=54361321 }},
+            {"order_module_item",new TableInfo{SimulaRowCount=69173339 }},
+            {"order_package",new TableInfo{SimulaRowCount=16196195 }},
+            {"order_process",new TableInfo{SimulaRowCount=3892685 }},//orderNo < 202301的  
+            {"order_process_step",new TableInfo{SimulaRowCount=8050349 }},//orderNo < 202301的删除   
+            {"order_process_step_item",new TableInfo{SimulaRowCount=14538058 }},//orderNo < 202301的删除 
+            {"order_scrap_board",new TableInfo{SimulaRowCount=123998 }},
+            {"process_group",new TableInfo{SimulaRowCount=1253 }},
+            {"process_info",new TableInfo{SimulaRowCount=7839 }},
+            {"process_item_exp",new TableInfo{SimulaRowCount=28 }},
+            {"process_schdule_capacity",new TableInfo{SimulaRowCount=39736 }},
+            {"process_step_efficiency",new TableInfo{SimulaRowCount=8 }},
+            {"report_template",new TableInfo{SimulaRowCount=7337 }},
+            {"simple_package",new TableInfo{SimulaRowCount=130436 }},//orderNo < 202301的删除
+            {"sys_config",new TableInfo{SimulaRowCount=2296 }},
+            {"work_calendar",new TableInfo{SimulaRowCount=11 }},
+            {"work_shift",new TableInfo{SimulaRowCount=59 }},
+            {"work_time",new TableInfo{SimulaRowCount=62 }},
+        };
+        taskFun(new TasksOptions { TableInfoConfig = smallTablesDic, OutPutOptions = new OutPutOptions { FlushCount = 20000, OutPutTaskCount = 4 } },
+            new DataRecordQueue(), new DataRecordQueue(), new ProcessContext());
     }
 }

ConsoleApp2/HostedServices/OutputService.cs

@@ -5,8 +5,6 @@ using ConsoleApp2.Services;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Options;
-using MySqlConnector;
-using System.Threading;
 
 namespace ConsoleApp2.HostedServices;
 
@@ -16,41 +14,39 @@ namespace ConsoleApp2.HostedServices;
 public class OutputService : IOutputService
 {
     private readonly ILogger _logger;
-    private readonly DataRecordQueue _consumerQueue;
+    private readonly IOptions<DatabaseOutputOptions> _outputOptions;
-    private readonly IOptions<DataTransformOptions> _transOptions;
+    private readonly IOptions<DataTransformOptions> _transformOptions;
-    private readonly IOptions<DatabaseOutputOptions> _options;
     private readonly ProcessContext _context;
     private readonly TaskManager _taskManager;
+    private readonly ErrorRecorder _errorRecorder;
 
     public OutputService(ILogger<OutputService> logger,
-        [FromKeyedServices(ProcessStep.Consumer)] DataRecordQueue consumerQueue,
+        IOptions<DatabaseOutputOptions> outputOptions,
-        IOptions<DatabaseOutputOptions> options,
-        IOptions<DataTransformOptions> transOptions,
         ProcessContext context,
-        TaskManager taskManager)
+        TaskManager taskManager, 
+        IOptions<DataTransformOptions> transformOptions, 
+        ErrorRecorder errorRecorder)
     {
         _logger = logger;
-        _consumerQueue = consumerQueue;
+        _outputOptions = outputOptions;
-        _transOptions = transOptions;
-        _options = options;
         _context = context;
         _taskManager = taskManager;
+        _transformOptions = transformOptions;
+        _errorRecorder = errorRecorder;
     }
 
-    public async Task ExecuteAsync(CancellationToken cancellationToken)
+    public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue consumerQueue, ProcessContext context,CancellationToken cancellationToken)
     {
         _logger.LogInformation("***** Mysql output service started *****");
-        var count = 0;
         _taskManager.CreateTasks(async () =>
         {
             var records = new List<DataRecord>();
-            while (!_context.IsTransformCompleted || _consumerQueue.Count > 0)
+            while (!context.IsTransformCompleted || consumerQueue.Count > 0)
             {
-                if (!_consumerQueue.TryDequeue(out var record)) continue;
+                if (!consumerQueue.TryDequeue(out var record)) continue;
                 records.Add(record);
-                count++;
                 //_logger.LogInformation(@"*****OutputCount: {count} *****",count);
-                if (records.Count >= _options.Value.FlushCount)
+                if (records.Count >= tasksOptions.OutPutOptions.FlushCount)
                 {
                     await FlushAsync(records);
                     records.Clear();
@@ -67,7 +63,7 @@ public class OutputService : IOutputService
                 records.Clear();
                 _logger.LogInformation("***** Mysql output thread completed *****");
             }
-        }, _options.Value.TaskCount);
+        }, tasksOptions.OutPutOptions.OutPutTaskCount);
 
         await _taskManager.WaitAll();
         //_context.CompleteOutput();
@@ -79,8 +75,8 @@ public class OutputService : IOutputService
     {
         var count = 0;
         await using var output = new MySqlDestination(
-            _options.Value.ConnectionString ?? throw new InvalidOperationException("Connection string is required"),
+            _outputOptions.Value.ConnectionString ?? throw new InvalidOperationException("Connection string is required"),
-            _logger, _context,true);
+            _logger, _context, _transformOptions, _errorRecorder);
         //if (records == null || records.Count() == 0) return;
         //var dbName = $"cferp_test_1";
         //if (records != null && records.Count() > 0)
@@ -102,7 +98,7 @@ public class OutputService : IOutputService
             await output.WriteRecordAsync(record);
             count++;
         }
-        await output.FlushAsync(_options.Value.MaxAllowedPacket, _transOptions);
+        await output.FlushAsync(_outputOptions.Value.MaxAllowedPacket);
         _context.AddOutput(count);
     }
 }

ConsoleApp2/HostedServices/TestInputService.cs

@@ -32,7 +32,7 @@ namespace ConsoleApp2.HostedServices
             _producerQueue = producerQueue;
             _context = context;
         }
-        public async Task ExecuteAsync(CancellationToken cancellationToken)
+        public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, ProcessContext context, CancellationToken cancellationToken)
         {
             var tableName = "order_item";
             var headers = new string[] { "ID","OrderNo","ItemNo","ItemType","RoomID","BoxID","DataID","PlanID","PackageID","Num","CompanyID","ShardKey" };

ConsoleApp2/HostedServices/TransformService.cs

@@ -2,6 +2,7 @@
 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;
@@ -16,95 +17,67 @@ 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 IDatabase _db;
+    private readonly IDistributedCache _cache;
+    private readonly TaskManager _taskManager;
 
 
     public TransformService(ILogger<TransformService> logger, 
         IOptions<DataTransformOptions> options,
-        [FromKeyedServices(ProcessStep.Producer)]DataRecordQueue producerQueue,
+        ProcessContext context,
-        [FromKeyedServices(ProcessStep.Consumer)]DataRecordQueue consumerQueue, 
+        IDistributedCache cache,
-        ProcessContext context, IDatabase db)
+        TaskManager taskManager)
     {
         _logger = logger;
         _options = options;
-        _producerQueue = producerQueue;
-        _consumerQueue = consumerQueue;
         _context = context;
-        _db = db;
+        _cache = cache;
+        _taskManager = taskManager;
     }
 
-    public async Task ExecuteAsync(CancellationToken cancellationToken)
+    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);
-        while ((!_context.IsInputCompleted || _producerQueue.Count > 0))
+
+        _taskManager.CreateTasks(async () =>
         {
-            if (_context.GetExceptions().Count > 0)
+            while ((!context.IsInputCompleted || producerQueue.Count > 0))
             {
-                _logger.LogInformation("***** Csv transform service is canceled *****");
+                if (_context.GetExceptions().Count > 0)
-                return;
-            }
-            // var dbOptions = _options.Value.DatabaseFilter(record);
-            if (!_producerQueue.TryDequeue(out var record)) continue;
-          
-            for (var i = 0; i < record.Fields.Length; i++)
-            {
-                var field = record[i];
-
-                if (field == "\\N")
                 {
-                    field = "NULL";
+                    _logger.LogInformation("***** Csv transform service is canceled *****");
-                    goto Escape;
+                    return;
                 }
-                // else if(DumpDataHelper.CheckHexField(field))
+                if (!producerQueue.TryDequeue(out var record)) continue;
-                //     field = $"0x{field}";
 
-                switch (_options.Value.GetColumnType(record.TableName, record.Headers[i]))
+                //过滤不要的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)
                 {
-                    case  ColumnType.Text:
+                    record = replaceRecord;
-
-                        field = string.IsNullOrEmpty(field) ? "''" : _options.Value.TransformBinary?.Invoke(field) ?? field; ;
-                        break;
-                    case ColumnType.Blob:
-                        //field = string.IsNullOrEmpty(field) ? "NULL" : $"0x{field}";
-                        break;
-                    default:
-                        break;
                 }
-
+                consumerQueue.Enqueue(record);
-            Escape:
+                _context.AddTransform();
-                record[i] = field;
+                //数据增加
-            }
+                var addRecords = _options.Value.RecordAdd?.Invoke(record);
-            //过滤不要的record
+                if (addRecords != null && addRecords.Count > 0)
-            if ( await _options.Value.RecordFilter?.Invoke(record,_db) == false) continue;
-            record.Database = _options.Value.DatabaseFilter?.Invoke(record);
-            //修改record
-            _options.Value.RecordModify?.Invoke(record);
-            //缓存record
-            _options.Value.RecordCache?.Invoke(record, _db);
-            //替换record
-            var replaceRecord =await _options.Value.RecordReplace?.Invoke(record, _db);
-            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);
+                    foreach (var rc in addRecords)
-                    _context.AddTransform();
+                    {
+                        consumerQueue.Enqueue(rc);
+                        _context.AddTransform();
+                    }
                 }
             }
-        }
+            context.CompleteTransform();
-
+        },tasksOptions.TransformTaskCount,cancellationToken);
-        _context.CompleteTransform();
         _logger.LogInformation("***** Data transformation service completed *****");
     }
 }

ConsoleApp2/HostedServices/VoidOutputService.cs

@@ -1,5 +1,6 @@
 using ConsoleApp2.Const;
 using ConsoleApp2.HostedServices.Abstractions;
+using ConsoleApp2.Options;
 using ConsoleApp2.Services;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -22,7 +23,7 @@ public class VoidOutputService : IOutputService
         _logger = logger;
     }
 
-    public Task ExecuteAsync(CancellationToken stoppingToken)
+    public Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue consumerQueue, ProcessContext context, CancellationToken cancellationToken)
     {
         _logger.LogInformation("***** Void output service started, thread id: {ThreadId} *****", Environment.CurrentManagedThreadId);
         while (!_context.IsTransformCompleted || _consumerQueue.Count > 0)

ConsoleApp2/Options/DataTransformOptions.cs

@@ -1,4 +1,5 @@
-using StackExchange.Redis;
+using Microsoft.Extensions.Caching.Distributed;
+using StackExchange.Redis;
 
 namespace ConsoleApp2.Options;
 
@@ -6,6 +7,7 @@ public enum ColumnType
 {
     Blob,
     Text,
+    Json,
     UnDefine,
 }
 
@@ -15,11 +17,11 @@ public class DataTransformOptions
 
     public Func<string, string>? TransformBinary { get; set; }//Binary转字符串方法
 
-    public Func<DataRecord, IDatabase, Task<bool>>? RecordFilter { get; set; }//数据过滤方法
+    public Func<DataRecord, IDistributedCache, Task<bool>>? RecordFilter { get; set; }//数据过滤方法
     public Action<DataRecord>? RecordModify { get; set; }//数据修改
-    public Func<DataRecord, IDatabase, Task<DataRecord?>>? RecordReplace { get; set; }//数据替换
+    public Func<DataRecord, IDistributedCache, Task<DataRecord?>>? RecordReplace { get; set; }//数据替换
     public Func<DataRecord, IList<DataRecord>?>? RecordAdd { get; set; }//数据替换
-    public Action<DataRecord, IDatabase>? RecordCache { get; set; }//数据缓存
+    public Func<DataRecord, IDistributedCache, Task>? RecordCache { get; set; }//数据缓存
 
     /// <summary>
     /// 配置导入数据的特殊列

ConsoleApp2/Options/DatabaseOutputOptions.cs

@@ -6,14 +6,5 @@ public class DatabaseOutputOptions
     /// 数据库连接字符串
     /// </summary>
     public string? ConnectionString { get; set; }
-    /// <summary>
-    /// 输出服务的任务(Task)数
-    /// </summary>
-    public int TaskCount { get; set; }
-    /// <summary>
-    /// 每个任务每次提交到数据库的记录数量（每N条构建一次SQL语句）
-    /// </summary>
-    public int FlushCount { get; set; }
-
     public int MaxAllowedPacket { get; set; } = 64*1024*1024;
 }

ConsoleApp2/Options/TasksOptions.cs

@@ -10,8 +10,15 @@ namespace ConsoleApp2.Options
     {
         public long SimulaRowCount { get; set; }//模拟的记录条数
     }
-    public class InputTableOptions
+    public class TasksOptions
     {
         public Dictionary<string, TableInfo> TableInfoConfig { get; set; } = new();
+        public int TransformTaskCount { get; set; } = 1;
+        public OutPutOptions OutPutOptions { get; set; }=new();
+    }
+    public class OutPutOptions
+    {
+        public int FlushCount { get; set; } = 10000;
+        public int OutPutTaskCount { get; set; } = 2;
     }
 }

ConsoleApp2/Program.cs

@@ -1,11 +1,9 @@
 using ConsoleApp2;
 using ConsoleApp2.Const;
-using ConsoleApp2.Helpers;
 using ConsoleApp2.HostedServices;
 using ConsoleApp2.HostedServices.Abstractions;
 using ConsoleApp2.Options;
 using ConsoleApp2.Services;
-using ConsoleApp2.SimulationService;
 using Microsoft.Extensions.Caching.StackExchangeRedis;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection;
@@ -13,13 +11,10 @@ using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 using MySqlConnector;
 using Serilog;
-using Serilog.Core;
+using Microsoft.Extensions.Caching.Distributed;
-using StackExchange.Redis;
+using Serilog.Events;
-using System.Reflection.PortableExecutable;
 
 
-// 运行之前把Mysql max_allowed_packets 调大
-// 运行之前把process_step表的外键删掉
 await RunProgram();
 return;
 
@@ -66,51 +61,49 @@ async Task RunProgram()
     var host = Host.CreateApplicationBuilder(args);
     var commandOptions = host.Configuration.GetSection("CmdOptions").Get<CommandOptions>() ?? new CommandOptions();
     Console.WriteLine($"InputDir:{commandOptions?.InputDir}");
-    Console.WriteLine($"OutPutFlushCount:{commandOptions?.FlushCount}");
-    Console.WriteLine($"OutPutTaskCount:{commandOptions?.TaskCount}");
 
     var oldestTime = DateTime.ParseExact(commandOptions.OldestTime, "yyyyMM", System.Globalization.DateTimeFormatInfo.InvariantInfo);
-    host.Services.Configure<InputTableOptions>(option =>
+    //host.Services.Configure<InputTableOptions>(option =>
-    {
+    //{
-        option.TableInfoConfig = new Dictionary<string, TableInfo>
+    //    option.TableInfoConfig = new Dictionary<string, TableInfo>
-        {
+    //    {
 
-            //order_block_plan_item从order_item表查询，然后程序插入                
+    //        //order_block_plan_item从order_item表查询，然后程序插入                
-            //order_package_item从order_item表查询，然后程序插入
+    //        //order_package_item从order_item表查询，然后程序插入
-            //order_patch_detail生产没有这个表，不处理
+    //        //order_patch_detail生产没有这个表，不处理
 
 
-            {"machine",new TableInfo{SimulaRowCount=14655 }},
+    //        {"machine",new TableInfo{SimulaRowCount=14655 }},
-            {"order",new TableInfo{SimulaRowCount=5019216 }},
+    //        {"order",new TableInfo{SimulaRowCount=5019216 }},
-            {"order_block_plan",new TableInfo{SimulaRowCount=2725553 }},//CreateTime < 202301的删除
+    //        {"order_block_plan",new TableInfo{SimulaRowCount=2725553 }},//CreateTime < 202301的删除
-            {"order_block_plan_result",new TableInfo{SimulaRowCount=1174096 }},
+    //        {"order_block_plan_result",new TableInfo{SimulaRowCount=1174096 }},
-            {"order_box_block",new TableInfo{SimulaRowCount=29755672 }},
+    //        {"order_box_block",new TableInfo{SimulaRowCount=29755672 }},
-            {"order_data_block",new TableInfo{SimulaRowCount=731800334 }},
+    //        {"order_data_block",new TableInfo{SimulaRowCount=731800334 }},
-            {"order_data_goods",new TableInfo{SimulaRowCount=25803671 }},
+    //        {"order_data_goods",new TableInfo{SimulaRowCount=25803671 }},
-            {"order_data_parts",new TableInfo{SimulaRowCount=468517543 }},
+    //        {"order_data_parts",new TableInfo{SimulaRowCount=468517543 }},
-            {"order_item",new TableInfo{SimulaRowCount=1345520079 }},
+    //        {"order_item",new TableInfo{SimulaRowCount=1345520079 }},
-            {"order_module",new TableInfo{SimulaRowCount=103325385 }},
+    //        {"order_module",new TableInfo{SimulaRowCount=103325385 }},
-            {"order_module_extra",new TableInfo{SimulaRowCount=54361321 }},
+    //        {"order_module_extra",new TableInfo{SimulaRowCount=54361321 }},
-            {"order_module_item",new TableInfo{SimulaRowCount=69173339 }},
+    //        {"order_module_item",new TableInfo{SimulaRowCount=69173339 }},
-            {"order_package",new TableInfo{SimulaRowCount=16196195 }},
+    //        {"order_package",new TableInfo{SimulaRowCount=16196195 }},
-            {"order_process",new TableInfo{SimulaRowCount=3892685 }},//orderNo < 202301的
+    //        {"order_process",new TableInfo{SimulaRowCount=3892685 }},//orderNo < 202301的
-            {"order_process_step",new TableInfo{SimulaRowCount=8050349 }},//orderNo < 202301的删除
+    //        {"order_process_step",new TableInfo{SimulaRowCount=8050349 }},//orderNo < 202301的删除
-            {"order_process_step_item",new TableInfo{SimulaRowCount=14538058 }},//orderNo < 202301的删除
+    //        {"order_process_step_item",new TableInfo{SimulaRowCount=14538058 }},//orderNo < 202301的删除
-            {"order_scrap_board",new TableInfo{SimulaRowCount=123998 }},
+    //        {"order_scrap_board",new TableInfo{SimulaRowCount=123998 }},
-            {"process_group",new TableInfo{SimulaRowCount=1253 }},
+    //        {"process_group",new TableInfo{SimulaRowCount=1253 }},
-            {"process_info",new TableInfo{SimulaRowCount=7839 }},
+    //        {"process_info",new TableInfo{SimulaRowCount=7839 }},
-            {"process_item_exp",new TableInfo{SimulaRowCount=28 }},
+    //        {"process_item_exp",new TableInfo{SimulaRowCount=28 }},
-            {"process_schdule_capacity",new TableInfo{SimulaRowCount=39736 }},
+    //        {"process_schdule_capacity",new TableInfo{SimulaRowCount=39736 }},
-            {"process_step_efficiency",new TableInfo{SimulaRowCount=8 }},
+    //        {"process_step_efficiency",new TableInfo{SimulaRowCount=8 }},
-            {"report_template",new TableInfo{SimulaRowCount=7337 }},
+    //        {"report_template",new TableInfo{SimulaRowCount=7337 }},
-            {"simple_package",new TableInfo{SimulaRowCount=130436 }},//orderNo < 202301的删除
+    //        {"simple_package",new TableInfo{SimulaRowCount=130436 }},//orderNo < 202301的删除
-            {"simple_plan_order",new TableInfo{SimulaRowCount=351470 }},//CreateTime < 202301的删除
+    //        {"simple_plan_order",new TableInfo{SimulaRowCount=351470 }},//CreateTime < 202301的删除
-            {"sys_config",new TableInfo{SimulaRowCount=2296 }},
+    //        {"sys_config",new TableInfo{SimulaRowCount=2296 }},
-            {"work_calendar",new TableInfo{SimulaRowCount=11 }},
+    //        {"work_calendar",new TableInfo{SimulaRowCount=11 }},
-            {"work_shift",new TableInfo{SimulaRowCount=59 }},
+    //        {"work_shift",new TableInfo{SimulaRowCount=59 }},
-            {"work_time",new TableInfo{SimulaRowCount=62 }},
+    //        {"work_time",new TableInfo{SimulaRowCount=62 }},
-        };
+    //    };
-    });
+    //});
     host.Services.Configure<CsvOptions>(option =>
     {
         option.Delimiter = ",";
@@ -133,7 +126,7 @@ async Task RunProgram()
 
         options.TransformBinary = field => commandOptions != null && commandOptions.Isutf8mb4 ? $"_utf8mb4 0x{field}" : $"0x{field}";
         //数据过滤
-        options.RecordFilter = async (record, db) =>
+        options.RecordFilter = async (record, cache) =>
         {
             //var index = Array.IndexOf(record.Headers, "ShardKey");
             if (record.TryGetField("ShardKey", out var skStr))
@@ -183,20 +176,20 @@ async Task RunProgram()
             if (record.TableName == "order_process_step" || record.TableName == "order_process_step_item")
             {
                 //如果缓存中不存在OrderProcessID,则丢弃
-                if(record.TryGetField("OrderProcessID",out var orderProcessID))
+                //if(record.TryGetField("OrderProcessID",out var orderProcessID))
-                {
+
-                    var value = await db.StringGetAsync(orderProcessID);
+ 
-                    if (string.IsNullOrEmpty(value.ToString()))return false;
+                    var value = await cache.GetStringAsync(record.GetCacheKey("OrderProcessID"));
-                }
+                    if (string.IsNullOrEmpty(value))return false;
+                
             }
             if (record.TableName == "order_block_plan_result" )
             {
                 //如果缓存中不存在ID,则丢弃(ID 对应order_block_plan中的ID)
-                if (record.TryGetField("ID", out var id))
+
-                {
+                    var value = await cache.GetStringAsync(record.GetCacheKey("ID"));
-                    var value = await db.StringGetAsync(id);
+                    if (string.IsNullOrEmpty(value)) return false;
-                    if (string.IsNullOrEmpty(value.ToString())) return false;
+                
-                }
             }
             return true;
 
@@ -208,7 +201,7 @@ async Task RunProgram()
             {
                 if (record.TryGetField("OrderNos", out var nos))
                 {
-                    if (nos.Length <= 2) record.SetField("OrderNos", "\"[]\"");
+                    if (nos.Length <= 2) record.SetField("OrderNos", "");
                 }
 
             }
@@ -218,7 +211,7 @@ async Task RunProgram()
                 if (record.TryGetField("NextStepID", out var idStr))
                 {
 
-                    if (idStr == "NULL")
+                    if (idStr == "\\N")
                     {
                         record.SetField("NextStepID", "0");
                     }
@@ -227,7 +220,7 @@ async Task RunProgram()
 
         };
         //数据缓存
-        options.RecordCache = async (record, db) =>
+        options.RecordCache = async (record, cache) =>
         {
             if (record.TableName == "order")
             {
@@ -235,7 +228,7 @@ async Task RunProgram()
                 {
                     if (record.TryGetField("CompanyID", out var companyid))
                     {
-                        await db.StringSetAsync(orderNo, companyid);
+                        await cache.SetStringAsync(record.GetCacheKey("OrderNo"), companyid);
                     }
                     
                 }
@@ -249,14 +242,7 @@ async Task RunProgram()
                    
                     if( record.TryGetField("ID", out var id))
                     {
-                        try
+                        await cache.SetStringAsync(record.GetCacheKey("ID"), sk);
-                        {
-                            await db.StringSetAsync(id, sk);
-                        }
-                        catch (Exception ex)
-                        {
-
-                        }
                     }
           
                 }
@@ -266,12 +252,12 @@ async Task RunProgram()
                 if (record.TryGetField("CompanyID", out var companyid))
                 {
                     record.TryGetField("ID", out var id);
-                    await db.StringSetAsync(id, companyid);
+                    await cache.SetStringAsync(record.GetCacheKey("ID"), companyid);
                 }
             }
         };
         //数据替换
-        options.RecordReplace = async (record, db) =>
+        options.RecordReplace = async (record, cache) =>
         {
             //删除数据源里simple_plan_order.ProcessState 字段和值
 
@@ -306,8 +292,9 @@ async Task RunProgram()
                     var headers = new List<string>(record.Headers);
                     var fields =new List<string>(record.Fields);
                     headers.Add("CompanyID");
-                    var companyidResult =await db.StringGetAsync(id);
+                    var companyidResult =await cache.GetStringAsync(record.GetCacheKey("ID"));
-                    _ = int.TryParse(companyidResult.ToString(), out var companyid);
+
+                    _ = int.TryParse(companyidResult, out var companyid);
                     fields.Add(companyid.ToString());
                     return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), companyid);
                 }
@@ -321,8 +308,8 @@ async Task RunProgram()
                         var headers = new List<string>(record.Headers);
                         var fields = new List<string>(record.Fields);
                         headers.Add("CompanyID");
-                        var companyidResult = await db.StringGetAsync(orderNo);
+                        var companyidResult = await cache.GetStringAsync(record.GetCacheKey("OrderNo"));
-                        _ = int.TryParse(companyidResult.ToString(), out var cpid);
+                        _ = int.TryParse(companyidResult, out var cpid);
                         fields.Add(cpid.ToString());
                         return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), cpid);
                     }
@@ -363,20 +350,12 @@ async Task RunProgram()
                 {
                     if (record.TryGetField("OrderProcessID",out var processID))
                     {
-                        try
+                        var shardKey =await cache.GetStringAsync(record.GetCacheKey("OrderProcessID"));
-                        {
-
-                        var shardKey =await db.StringGetAsync(processID);
                         var headers = new List<string>(record.Headers);
                         var fields = new List<string>(record.Fields);
                         headers.Add("ShardKey");
-                        fields.Add(shardKey.ToString());
+                        fields.Add(shardKey??"0");
                         return new DataRecord(fields.ToArray(), record.TableName, headers.ToArray(), record.CompanyID);
-                        }
-                        catch (Exception ex)
-                        {
-
-                        }
                     }
                 }
             }
@@ -455,7 +434,7 @@ async Task RunProgram()
             { "process_item_exp.ItemJson", ColumnType.Text },
             { "report_template.Template", ColumnType.Text },
             { "report_template.SourceConfig", ColumnType.Text },
-            { "order_block_plan.OrderNos", ColumnType.Text },
+            { "order_block_plan.OrderNos", ColumnType.Json },
             { "order_block_plan.BlockInfo", ColumnType.Text },
         };
     });
@@ -478,19 +457,24 @@ async Task RunProgram()
             TreatTinyAsBoolean = false,
             MaximumPoolSize = 50
         }.ConnectionString;
-        options.TaskCount = commandOptions.TaskCount;
-        options.FlushCount = commandOptions.FlushCount;
     });
+    
     host.Services.AddLogging(builder =>
     {
         builder.ClearProviders();
-        builder.AddSerilog(new LoggerConfiguration().WriteTo.Console().CreateLogger());
+        builder.AddSerilog(new LoggerConfiguration()
+            .WriteTo.Console()
+            .WriteTo.File("./log/error.log", restrictedToMinimumLevel:LogEventLevel.Error)
+            // .WriteTo.File("./log/info.log", restrictedToMinimumLevel:LogEventLevel.Information) //性能考虑暂不使用
+            .CreateLogger()
+        );
     });
 
-    host.Services.AddSingleton<ProcessContext>();
+    host.Services.AddScoped<ProcessContext>();
     host.Services.AddKeyedSingleton<DataRecordQueue>(ProcessStep.Producer);
     host.Services.AddKeyedSingleton<DataRecordQueue>(ProcessStep.Consumer);
     host.Services.AddTransient<TaskManager>();
+    host.Services.AddSingleton<ErrorRecorder>();
 
     host.Services.AddHostedService<MainHostedService>();
     host.Services.AddHostedService<TaskMonitorService>();
@@ -498,8 +482,11 @@ async Task RunProgram()
     host.Services.AddSingleton<ITransformService, TransformService>();
     host.Services.AddSingleton<IOutputService, OutputService>();
     var redisOptions = host.Configuration.GetSection("RedisCacheOptions").Get<RedisCacheOptions>() ?? new RedisCacheOptions();
-    var redis = ConnectionMultiplexer.Connect(redisOptions.Configuration);
+    host.Services.AddStackExchangeRedisCache(options =>
-    host.Services.AddSingleton(redis.GetDatabase());
+    {
+        options.Configuration = redisOptions.Configuration;
+        options.InstanceName = redisOptions.InstanceName;
+    });
     var app = host.Build();
     await app.RunAsync();
 }

ConsoleApp2/Services/ErrorRecorder.cs

@@ -0,0 +1,104 @@
+using System.Text;
+using Microsoft.Extensions.Logging;
+
+namespace ConsoleApp2.Services;
+
+public class ErrorRecorder
+{
+    private readonly string _outputDir = "./ErrorRecords";
+    private readonly ILogger _logger;
+    private readonly Dictionary<string, int> _logIndex = new();
+
+    /// <summary>
+    /// 当次执行标识
+    /// </summary>
+    private static readonly string UID = DateTime.Now.ToString("yyyy-MM-dd HH-mm-ss");
+
+    public ErrorRecorder(ILogger<ErrorRecorder> logger)
+    {
+        _logger = logger;
+        var dir = Path.Combine(_outputDir, UID);
+        if (!Directory.Exists(dir))
+        {
+            Directory.CreateDirectory(dir);
+        }
+    }
+
+    /// <summary>
+    /// 记录已知表名发生错误的SQL
+    /// </summary>
+    /// <param name="commandText"></param>
+    /// <param name="tableName"></param>
+    /// <param name="exception"></param>
+    public async Task LogErrorSqlAsync(string commandText, string tableName, Exception exception)
+    {
+        if (!_logIndex.TryGetValue(tableName, out var idx))
+        {
+            idx = 0;
+            _logIndex.Add(tableName, idx);
+        }
+        var filePath = Path.Combine(_outputDir, UID, $"{tableName}-{idx}.errlog");
+            
+        if (File.Exists(filePath) && new FileInfo(filePath).Length > 10 * 1024 * 1024)
+        {
+            ++idx;
+            _logIndex[tableName] = idx;
+            filePath = Path.Combine(_outputDir, UID, $"{tableName}-{idx}.errlog");
+        }
+        var content = $"""
+                      /* [{DateTime.Now:yyyy-MM-dd HH:mm:ss}]
+                       * Error occurred when export table '{tableName}':
+                       * {exception.Message}
+                       */
+                      
+                      {commandText}
+
+
+                      """;
+        await File.AppendAllTextAsync(filePath, content, Encoding.UTF8);
+    }
+
+    /// <summary>
+    /// 记录发生错误的SQL
+    /// </summary>
+    /// <param name="commandText"></param>
+    /// <param name="exception"></param>
+    public async Task LogErrorSqlAsync(string commandText, Exception exception)
+    {
+        var filePath = Path.Combine(_outputDir, UID, "UnknownTables.errlog");
+        var content = $"""
+                       /* [{DateTime.Now:yyyy-MM-dd HH:mm:ss}]
+                        * Error occurred when export table with unknown table name:
+                        * {exception.Message}
+                        */
+                       {commandText}
+                       
+                       
+                       """;
+        await File.AppendAllTextAsync(filePath, content, Encoding.UTF8);
+    }
+
+    public async Task LogErrorRecordsAsync(IDictionary<string, DataRecord> records, Exception exception)
+    {
+        var pathDict = new Dictionary<string, string>();
+        foreach (var pair in records)
+        {
+            if(!pathDict.TryGetValue(pair.Key, out var path))
+            {
+                path = Path.Combine(_outputDir, UID, "ErrorRecords", $"{pair.Key}.errlog");
+                pathDict.Add(pair.Key, path);
+            }
+            //
+            await File.AppendAllTextAsync(path, string.Join(',', pair.Value.Fields));
+        }
+    }
+    
+    public void ClearErrorRecords()
+    {
+        _logger.LogInformation("***** Clear error records *****");
+        foreach (var file in Directory.GetFiles(_outputDir, "*.errlog", SearchOption.AllDirectories))
+        {
+            File.Delete(file);
+        }
+    }
+}

ConsoleApp2/Services/MySqlDestination.cs

@@ -1,5 +1,6 @@
-using System.Reflection.Metadata;
+using System.Data.Common;
 using System.Text;
+using System.Text.RegularExpressions;
 using ConsoleApp2.Helpers;
 using ConsoleApp2.Options;
 using Microsoft.Extensions.Logging;
@@ -12,27 +13,31 @@ namespace ConsoleApp2.Services;
 /// <summary>
 /// Mysql导出
 /// </summary>
-public class MySqlDestination : IDisposable, IAsyncDisposable
+public partial class MySqlDestination : IDisposable, IAsyncDisposable
 {
     private readonly Dictionary<string, IList<DataRecord>> _recordCache;
     private readonly MySqlConnection _conn;
     private readonly ILogger _logger;
-    private readonly bool _prettyOutput;
-    private readonly int _maxAllowPacket;
     private readonly ProcessContext _context;
+    private readonly IOptions<DataTransformOptions> _transformOptions;
+    private readonly ErrorRecorder _errorRecorder;
 
-    public MySqlDestination(string connStr, ILogger logger, ProcessContext context,bool prettyOutput = false)
+    public MySqlDestination(
+        string connStr, 
+        ILogger logger, 
+        ProcessContext context, 
+        IOptions<DataTransformOptions> transformOptions,
+        ErrorRecorder errorRecorder)
     {
         _conn = new MySqlConnection(connStr);
         _conn.Open();
         _recordCache = new Dictionary<string, IList<DataRecord>>();
         _logger = logger;
         _context = context;
-        _prettyOutput = prettyOutput;
+        _transformOptions = transformOptions;
-
+        _errorRecorder = errorRecorder;
-
-
     }
+    
     public Task WriteRecordAsync(DataRecord record)
     {
         _recordCache.AddOrUpdate(record.TableName, [record], (key, value) =>
@@ -51,28 +56,44 @@ public class MySqlDestination : IDisposable, IAsyncDisposable
         }
     }
 
-    public async Task FlushAsync(int maxAllowPacket, IOptions<DataTransformOptions> transOptions)
+    public async Task FlushAsync(int maxAllowPacket)
     {
         if (_recordCache.Count == 0)
             return;
 
         var cmd = _conn.CreateCommand();
         cmd.CommandTimeout = 3 * 60;
-        var excuseList = GetExcuseList(_recordCache, maxAllowPacket, transOptions, _prettyOutput);
+       
         try
         {
+            var excuseList = GetExcuseList(_recordCache, maxAllowPacket).ToList();
             foreach (var insertSql in excuseList)
             {
                 cmd.CommandText = insertSql;
-                await cmd.ExecuteNonQueryAsync();
+                try
+                {
+                    await cmd.ExecuteNonQueryAsync();
+                }
+                catch (Exception e)
+                {
+                    _logger.LogCritical(e, "Error when flushing records, sql: {Sql}", cmd.CommandText.Omit(1000));
+                    _context.AddException(e);
+                    
+                    var match = MatchTableName().Match(cmd.CommandText);
+                    if (match is { Success: true, Groups.Count: > 1 })
+                    {
+                        var tableName = match.Groups[1].Value;
+                        await _errorRecorder.LogErrorSqlAsync(cmd.CommandText, tableName, e);
+                    }
+                    else await _errorRecorder.LogErrorSqlAsync(cmd.CommandText, e);
+                }
             }
             _recordCache.Clear();
         }
         catch (Exception e)
         {
-            _logger.LogCritical(e, "Error when flushing records, sql: {Sql}", cmd.CommandText.Omit(1000));
+            _logger.LogCritical(e, "Error when serialize records, record:");
             _context.AddException(e);
-            throw;
         }
         finally
         {
@@ -80,89 +101,102 @@ public class MySqlDestination : IDisposable, IAsyncDisposable
         }
     }
     
-    public static IList<string> GetExcuseList(IDictionary<string, IList<DataRecord>> tableRecords,int maxAllowPacket, IOptions<DataTransformOptions> transOptions,
+    [GeneratedRegex("INSERT INTO `([^`]+)`")]
-           bool prettyOutput = false)
+    private static partial Regex MatchTableName();
+
+    public IEnumerable<string> GetExcuseList(IDictionary<string, IList<DataRecord>> tableRecords,int maxAllowPacket)
     {
-        var resultList = new List<string>();
+        var sb = new StringBuilder();
-        var headerSb = new StringBuilder();
-        var recordSb = new StringBuilder();
         foreach (var (tableName, records) in tableRecords)
         {
             if (records.Count == 0)
                 continue;
-            headerSb.Append($"INSERT INTO `{tableName}`(");
+            
+            var recordIdx = 0;
+            StartBuild:
+            var noCommas = true;
+            
+            // INSERT INTO ... VALUES >>>
+            sb.Append($"INSERT INTO `{tableName}`(");
             for (var i = 0; i < records[0].Headers.Length; i++)
             {
                 var header = records[0].Headers[i];
-                headerSb.Append($"`{header}`");
+                sb.Append($"`{header}`");
                 if (i != records[0].Headers.Length - 1)
-                    headerSb.Append(',');
+                    sb.Append(',');
             }
 
-            headerSb.Append(") VALUES ");
+            sb.Append(") VALUES ");
-            if (prettyOutput)
-                headerSb.AppendLine();
             
-            var sbList = new List<string>();
+            // ([FIELDS]), >>>
-            var currentLength = headerSb.Length;
+            for (;recordIdx < records.Count; recordIdx++)
-            for (var i = 0; i < records.Count; i++)
             {
-                var record = records[i];
+                var record = records[recordIdx];
+                var recordSb = new StringBuilder();
                 recordSb.Append('(');
-                for (var j = 0; j < record.Fields.Length; j++)
+                for (var fieldIdx = 0; fieldIdx < record.Fields.Length; fieldIdx++)
                 {
-                    var field = record.Fields[j];
+                    var field = record.Fields[fieldIdx];
-                    var header = record.Headers[j];
+
-                    if (transOptions.Value.GetColumnType(record.TableName, header) ==ColumnType.Blob)
+                    // 在这里处理特殊列
+                    #region HandleFields
+                    if (field == "\\N")
                     {
-                        if (string.IsNullOrEmpty(field))
+                        recordSb.Append("NULL");
-                        {
+                        goto Escape;
-                            recordSb.Append("NULL");
-                        }
-                        else
-                        recordSb.Append("0x"+field);
                     }
-                    else
+                    
-                    recordSb.Append(field);
+                    switch (_transformOptions.Value.GetColumnType(record.TableName, record.Headers[fieldIdx]))
-                    if (j != record.Fields.Length - 1)
+                    {
+                        case ColumnType.Text:
+                            recordSb.Append(string.IsNullOrEmpty(field)
+                                ? "''"
+                                : _transformOptions.Value.TransformBinary?.Invoke(field) ?? field);
+                            break;
+                        case ColumnType.Blob:
+                            if (string.IsNullOrEmpty(field))
+                                recordSb.Append("''");
+                            else recordSb.Append($"0x{field}");
+                            break;
+                        case ColumnType.Json:
+                            recordSb.Append(string.IsNullOrEmpty(field)
+                                ? "\"[]\""
+                                : _transformOptions.Value.TransformBinary?.Invoke(field) ?? field);
+                            break;
+                        case ColumnType.UnDefine:
+                        default:
+                            recordSb.Append(field);
+                            break;
+                    }
+                    
+                    Escape:
+
+                    #endregion
+                    if (fieldIdx != record.Fields.Length - 1)
                         recordSb.Append(',');
                 }
 
                 recordSb.Append(')');
 
-                //if (i != records.Count - 1) // not last field
+                // 若字符数量即将大于限制，则返回SQL，清空StringBuilder，保留当前记录的索引值，然后转到StartBuild标签重新开始一轮INSERT
-                //    recordSb.Append(',');
+                if (sb.Length + recordSb.Length + 1 > maxAllowPacket)
-                if (prettyOutput) recordSb.AppendLine();
+                {
+                    sb.Append(';');
+                    yield return sb.ToString();
+                    sb.Clear();
+                    goto StartBuild;
+                }
 
-                if (currentLength + recordSb.Length >= maxAllowPacket)
+                if (!noCommas)
-                {
+                    sb.Append(',').AppendLine();
-                    var insertSb = new StringBuilder(headerSb.ToString());
+                noCommas = false;
-                    insertSb.Append(string.Join(",", sbList));
+                sb.Append(recordSb); // StringBuilder.Append(StringBuilder)不会分配多余的内存
-                    insertSb.Append(";");
-                    resultList.Add(insertSb.ToString());
-                    insertSb.Clear();
-                    sbList.Clear();
-                    sbList.Add(recordSb.ToString());
-                    currentLength = headerSb.Length + 1;//逗号长度加1
-                }
-                else
-                {
-                    sbList.Add(recordSb.ToString());
-                }
-                currentLength += recordSb.Length;
-                recordSb.Clear();
             }
-            if (sbList.Count > 0)
+
-            {
+            sb.Append(';');
-                var insertSb = new StringBuilder(headerSb.ToString());
+            yield return sb.ToString();
-                insertSb.Append(string.Join(",", sbList));
+            sb.Clear();
-                insertSb.Append(";");
-                resultList.Add(insertSb.ToString());
-                insertSb.Clear();
-            }
-            headerSb.Clear();
         }
-        return resultList;
     }
 
 

ConsoleApp2/SimulationService/SimulationInputService.cs

@@ -16,23 +16,20 @@ namespace ConsoleApp2.SimulationService
     {
         private readonly ILogger _logger;
         private readonly IOptions<DataInputOptions> _dataInputOptions;
-        private readonly IOptions<InputTableOptions> _tableOptions;
         private readonly DataRecordQueue _producerQueue;
         private readonly ProcessContext _context;
 
         public SimulationInputService(ILogger<InputService> logger,
             IOptions<DataInputOptions> dataInputOptions,
-            IOptions<InputTableOptions> tableOptions,
             [FromKeyedServices(ProcessStep.Producer)] DataRecordQueue producerQueue,
             ProcessContext context)
         {
             _logger = logger;
             _dataInputOptions = dataInputOptions;
-            _tableOptions = tableOptions;
             _producerQueue = producerQueue;
             _context = context;
         }
-        public async Task ExecuteAsync(CancellationToken cancellationToken)
+        public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue producerQueue, ProcessContext context, CancellationToken cancellationToken)
         {
             var inputDir = _dataInputOptions.Value.InputDir;
             _logger.LogInformation("***** simulation input service start, working dir: {InputDir}, thread id: {ThreadId} *****", inputDir, Environment.CurrentManagedThreadId);
@@ -42,9 +39,9 @@ namespace ConsoleApp2.SimulationService
                 _logger.LogInformation("No source files found in {InputDir}", inputDir);
                 return;
             }
-            foreach (var tableName in _tableOptions.Value.TableInfoConfig.Keys)
+            foreach (var tableName in tasksOptions.TableInfoConfig.Keys)
             {
-                    var dataCount = _tableOptions.Value.TableInfoConfig[tableName].SimulaRowCount;//当前表要生成的总数据量
+                    var dataCount = tasksOptions.TableInfoConfig[tableName].SimulaRowCount;//当前表要生成的总数据量
                     var companyTotallCount = 1000;//当前表每个公司生成的总数据量
                     var tempRecords = new List<DataRecord>();
                     var sk = DataHelper.shareKeys.First();

ConsoleApp2/appsettings.json

@@ -2,7 +2,7 @@
   "CmdOptions": {
     "InputFileType": "CSV",
     "InputDir": "D:/MyDumper-ZST",
-    "TaskCount": 4,
+    "TaskCount": 6,
     "FlushCount": 10000,
     "Isutf8mb4": true,
     "OldestShardKey": 23000,
@@ -12,6 +12,7 @@
     "MySqlMaster": "Server=127.0.0.1;Port=33309;UserId=root;Password=123456;Database=cferp_test;"
   },
   "RedisCacheOptions": {
-    "Configuration": "localhost:6379"
+    "Configuration": "192.168.1.246:6380",
+    "InstanceName" : "mes-etl:"
   }
 }