CZY/MES-ETL
1
0
Fork 0
Code Pull Requests Releases Activity

支持按多个表开立线程

Browse Source
This commit is contained in:
lindj
2024-01-18 14:36:36 +08:00
parent 629a4d2fb5
commit 97e359468f
16 changed files with 232 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
ConsoleApp2/HostedServices/Abstractions/IInputService.cs
View File

@@ -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);
}

7
ConsoleApp2/HostedServices/Abstractions/IOutputService.cs
View File

@@ -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);
}

7
ConsoleApp2/HostedServices/Abstractions/ITransformService.cs
View File

@@ -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);
}

14
ConsoleApp2/HostedServices/InputService.cs
View File

@@ -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 *****");
}
}

117
ConsoleApp2/HostedServices/MainHostedService.cs
View File

@@ -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 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 () =>
var taskFun = (TasksOptions taskOp, DataRecordQueue producerQueue, DataRecordQueue consumerQueue, ProcessContext context) =>
{
try
var inputTask = Task.Factory.StartNew(async () =>
{
await _transform.ExecuteAsync(stoppingToken);
}
catch (Exception ex)
try
{
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);
_logger.LogError("Exception occurred on transformService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
}
try
{
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
{
await _output.ExecuteAsync(stoppingToken);
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on outputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
}
});
// await Task.Run(async () => await _output.ExecuteAsync(stoppingToken), stoppingToken);
{"order_block_plan",new TableInfo{SimulaRowCount=2725553 }},//CreateTime < 202301的删除
{"order_block_plan_result",new TableInfo{SimulaRowCount=1174096 }},
{"order_box_block",new TableInfo{SimulaRowCount=29755672 }},
{"order_item",new TableInfo{SimulaRowCount=1345520079 }},
{"simple_plan_order",new TableInfo{SimulaRowCount=351470 }},//CreateTime < 202301的删除
};
taskFun(new TasksOptions { TableInfoConfig = bigTablesDic, OutPutOptions = new OutPutOptions { FlushCount = 10000, OutPutTaskCount = 2 } },
new DataRecordQueue(), new DataRecordQueue(),new ProcessContext());
var smallTablesDic = new Dictionary<string, TableInfo>
{
{"machine",new TableInfo{SimulaRowCount=14655 }},
{"order",new TableInfo{SimulaRowCount=5019216 }},
{"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());
}
}

15
ConsoleApp2/HostedServices/OutputService.cs
View File

@@ -16,39 +16,36 @@ namespace ConsoleApp2.HostedServices;
public class OutputService : IOutputService
{
private readonly ILogger _logger;
private readonly DataRecordQueue _consumerQueue;
private readonly IOptions<DatabaseOutputOptions> _outputOptions;
private readonly IOptions<DataTransformOptions> _transformOptions;
private readonly ProcessContext _context;
private readonly TaskManager _taskManager;
private readonly IOptions<DatabaseOutputOptions> _outputOptions;
public OutputService(ILogger<OutputService> logger,
[FromKeyedServices(ProcessStep.Consumer)] DataRecordQueue consumerQueue,
IOptions<DatabaseOutputOptions> outputOptions,
ProcessContext context,
TaskManager taskManager,
IOptions<DataTransformOptions> transformOptions)
{
_logger = logger;
_consumerQueue = consumerQueue;
_outputOptions = outputOptions;
_context = context;
_taskManager = taskManager;
_transformOptions = transformOptions;
}
public async Task ExecuteAsync(CancellationToken cancellationToken)
public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue consumerQueue, ProcessContext context,CancellationToken cancellationToken)
{
_logger.LogInformation("***** Mysql output service started *****");
_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);
//_logger.LogInformation(@"*****OutputCount: {count} *****",count);
if (records.Count >= _outputOptions.Value.FlushCount)
if (records.Count >= tasksOptions.OutPutOptions.FlushCount)
{
await FlushAsync(records);
records.Clear();
@@ -65,7 +62,7 @@ public class OutputService : IOutputService
records.Clear();
_logger.LogInformation("***** Mysql output thread completed *****");
}
}, _outputOptions.Value.TaskCount);
}, tasksOptions.OutPutOptions.OutPutTaskCount);
await _taskManager.WaitAll();
//_context.CompleteOutput();

2
ConsoleApp2/HostedServices/TestInputService.cs
View File

@@ -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" };

82
ConsoleApp2/HostedServices/TransformService.cs
View File

@@ -17,69 +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 IDistributedCache _cache;
private readonly TaskManager _taskManager;
public TransformService(ILogger<TransformService> logger,
IOptions<DataTransformOptions> options,
[FromKeyedServices(ProcessStep.Producer)]DataRecordQueue producerQueue,
[FromKeyedServices(ProcessStep.Consumer)]DataRecordQueue consumerQueue,
ProcessContext context,
IDistributedCache cache)
IDistributedCache cache,
TaskManager taskManager)
{
_logger = logger;
_options = options;
_producerQueue = producerQueue;
_consumerQueue = consumerQueue;
_context = context;
_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 *****");
return;
}
// var dbOptions = _options.Value.DatabaseFilter(record);
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
_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)
if (_context.GetExceptions().Count > 0)
{
_consumerQueue.Enqueue(rc);
_context.AddTransform();
_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
_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();
context.CompleteTransform();
},tasksOptions.TransformTaskCount,cancellationToken);
_logger.LogInformation("***** Data transformation service completed *****");
}
}

3
ConsoleApp2/HostedServices/VoidOutputService.cs
View File

@@ -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)