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

update

Browse Source
This commit is contained in:
陈梓阳 2024-02-09 13:41:40 +08:00
parent 41a1dc8a4f
commit 913c725fe1
4 changed files with 15 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
MesETL.App/Program.cs
View File

@ -595,15 +595,17 @@ async Task RunProgram()
host.Services.AddDataSourceFactory(); host.Services.AddDataSourceFactory();
host.Services.AddErrorRecorderFactory(); host.Services.AddErrorRecorderFactory();
host.Services.AddSingleton<ProcessContext>(); host.Services.AddSingleton<ProcessContext>();
host.Services.AddKeyedSingleton<DataRecordQueue>(ConstVar.Producer, new DataRecordQueue(200_000)); var prodLen = host.Configuration.GetRequiredSection("RecordQueue").GetValue<int>("ProducerQueueLength");
host.Services.AddRecordQueuePool(tenantDbOptions.DbGroup.Keys.Select(key => (key:key, queue:new DataRecordQueue(60_000))).ToArray()); var consLen = host.Configuration.GetRequiredSection("RecordQueue").GetValue<int>("ConsumerQueueLength");
host.Services.AddKeyedSingleton<DataRecordQueue>(ConstVar.Producer, new DataRecordQueue(prodLen));
host.Services.AddRecordQueuePool(tenantDbOptions.DbGroup.Keys.Select(key => (key:key, queue:new DataRecordQueue(consLen))).ToArray());
host.Services.AddSingleton<ITaskMonitorLogger, CacheTaskMonitorLogger>(); host.Services.AddSingleton<ITaskMonitorLogger, CacheTaskMonitorLogger>();
host.Services.AddSingleton<ITaskMonitorLogger, LoggerTaskMonitorLogger>(); host.Services.AddSingleton<ITaskMonitorLogger, LoggerTaskMonitorLogger>();
host.Services.AddHostedService<MainHostedService>(); host.Services.AddHostedService<MainHostedService>();
host.Services.AddSingleton<IInputService, FileInputService>(); host.Services.AddSingleton<IInputService, FileInputService>();
host.Services.AddSingleton<ITransformService, TransformService>(); host.Services.AddSingleton<ITransformService, TransformService>();
host.Services.AddSingleton<IOutputService, OutputService>(); host.Services.AddSingleton<IOutputService, VoidOutputService>();
host.Services.AddSingleton<TaskMonitorService>(); host.Services.AddSingleton<TaskMonitorService>();
host.Services.AddRedisCache(redisOptions); host.Services.AddRedisCache(redisOptions);
var app = host.Build(); var app = host.Build();

2
MesETL.App/Services/ETL/CsvReader.cs
View File

@ -49,7 +49,7 @@ public class CsvReader : IDataReader
if (string.IsNullOrWhiteSpace(str)) if (string.IsNullOrWhiteSpace(str))
return false; return false;
var fields = ParseRow(str, QuoteChar, Delimiter[0]); var fields = ParseRowFaster(str, QuoteChar, Delimiter[0]);
Current = new DataRecord(fields, TableName, Headers); Current = new DataRecord(fields, TableName, Headers);
return true; return true;
} }

12
MesETL.App/appsettings.json
View File

@ -11,9 +11,9 @@
}, },
"Transform":{ "Transform":{
"StrictMode": false, // 设为true时如果数据转换发生错误，立刻停止程序 "StrictMode": false, // 设为true时如果数据转换发生错误，立刻停止程序
"EnableFilter": true, // 启用数据过滤 "EnableFilter": false, // 启用数据过滤
"EnableReplacer": true, // 启用数据修改 "EnableReplacer": false, // 启用数据修改
"EnableReBuilder": true, // 启用数据重建 "EnableReBuilder": false, // 启用数据重建
"CleanDate": "202301" // 当数据过滤开启时，删除这个时间之前的数据 "CleanDate": "202301" // 当数据过滤开启时，删除这个时间之前的数据
}, },
"Output":{ "Output":{
@ -23,9 +23,13 @@
"MaxDatabaseOutputTask" : 4, // 每个数据库最大提交任务数 "MaxDatabaseOutputTask" : 4, // 每个数据库最大提交任务数
"TreatJsonAsHex": false // 将json列作为16进制格式输出(0x前缀)，生产库是没有json列的 "TreatJsonAsHex": false // 将json列作为16进制格式输出(0x前缀)，生产库是没有json列的
}, },
"RecordQueue":{
"ProducerQueueLength": 50000, // 输入队列最大长度
"ConsumerQueueLength": 10000, // 每个输出队列最大长度
},
"RedisCache": { "RedisCache": {
"Configuration": "192.168.1.246:6380", "Configuration": "192.168.1.246:6380",
"InstanceName" : "mes-etl:" "InstanceName" : "mes-etl:"
}, },
"TenantDb": // 分库配置 "TenantDb": // 分库配置
{ {

2
MesETL.Test/Test.cs
View File

@ -38,7 +38,7 @@ public class Test
while (!reader.EndOfStream) while (!reader.EndOfStream)
{ {
var str = await reader.ReadLineAsync(); var str = await reader.ReadLineAsync();
// char a; char a;
// foreach (var c in str) // foreach (var c in str)
// { // {
// a = c; // a = c;