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

添加数据分库;

Browse Source 
修复taskManager中异步方法没有正常等待的错误;
删除无用的异常捕获;
This commit is contained in:
陈梓阳 2024-01-19 11:17:22 +08:00
parent 45ad15a065
commit 8da3110ecd
12 changed files with 94 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
ConsoleApp2/DataRecord.cs
View File

@ -10,7 +10,7 @@ public class DataRecord
value = string.Empty;
if (record.Headers is null)
throw new InvalidOperationException("Cannot get field when headers of a record have not been set.");
var idx = Array.IndexOf(record.Headers, columnName); //可能可以优化
var idx = Array.IndexOf(record.Headers, columnName);
if (idx == -1)
return false;
value = record.Fields[idx];
@ -23,7 +23,7 @@ public class DataRecord
throw new InvalidOperationException("Headers have not been set.");
var idx = Array.IndexOf(record.Headers, columnName);
if (idx is -1)
throw new IndexOutOfRangeException("Column name not found in this record.");
throw new IndexOutOfRangeException($"Column name {columnName} not found in this record, table name {record.TableName}.");
return record.Fields[idx];
}
@ -34,7 +34,7 @@ public class DataRecord
public string TableName { get; }
public string? Database { get; set; }
public string Database { get; set; }
public int CompanyID { get; set; }

6
ConsoleApp2/HostedServices/InputService.cs
View File

@ -49,11 +49,7 @@ public class InputService : IInputService
count++;
});
if (_context.GetExceptions().Count > 0)
{
_logger.LogInformation("***** Csv input service is canceled *****");
return;
}
_logger.LogInformation("table:'{tableName}' input completed", tableName);
}

12
ConsoleApp2/HostedServices/MainHostedService.cs
View File

@ -34,8 +34,8 @@ public class MainHostedService : BackgroundService
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on inputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
_logger.LogCritical("Exception occurred on inputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
throw;
}
});
@ -47,8 +47,8 @@ public class MainHostedService : BackgroundService
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on transformService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
_logger.LogCritical("Exception occurred on transformService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
throw;
}
});
@ -60,8 +60,8 @@ public class MainHostedService : BackgroundService
}
catch (Exception ex)
{
_context.AddException(ex);
_logger.LogError("Exception occurred on outputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
_logger.LogCritical("Exception occurred on outputService:{Message},{StackTrace}", ex.Message, ex.StackTrace);
throw;
}
});

47
ConsoleApp2/HostedServices/OutputService.cs
View File

@ -1,8 +1,7 @@
using ConsoleApp2.Const;
using ConsoleApp2.Helpers;
using ConsoleApp2.HostedServices.Abstractions;
using ConsoleApp2.Options;
using ConsoleApp2.Services;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
@ -35,48 +34,54 @@ public class OutputService : IOutputService
_errorRecorder = errorRecorder;
}
public async Task ExecuteAsync(TasksOptions tasksOptions, DataRecordQueue consumerQueue, ProcessContext context,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>();
//k: database v: records，按照要导出的数据库名分组
var databaseDict = new Dictionary<string, List<DataRecord>>();
while (!context.IsTransformCompleted || consumerQueue.Count > 0)
{
if (!consumerQueue.TryDequeue(out var record)) continue;
records.Add(record);
//_logger.LogInformation(@"*****OutputCount: {count} *****",count);
var dbName = record.Database;
var records = databaseDict.AddOrUpdate(dbName, [record], (_, list) =>
{
list.Add(record);
return list;
});
if (records.Count >= tasksOptions.OutPutOptions.FlushCount)
{
await FlushAsync(records);
await FlushAsync(dbName, records);
records.Clear();
}
if (_context.GetExceptions().Count>0)
}
foreach (var (db, records) in databaseDict)
{
if (records.Count > 0)
{
_logger.LogInformation("***** Csv output thread is canceled *****");
return;
await FlushAsync(db, records);
records.Clear();
}
}
if (records.Count > 0)
{
await FlushAsync(records);
records.Clear();
_logger.LogInformation("***** Mysql output thread completed *****");
}
databaseDict.Clear();
_logger.LogInformation("***** Mysql output thread completed *****");
}, tasksOptions.OutPutOptions.OutPutTaskCount);
await _taskManager.WaitAll();
//_context.CompleteOutput();
_logger.LogInformation(@"***** Mysql output service completed *****");
_logger.LogInformation("***** Mysql output service completed *****");
}
private async Task FlushAsync(IEnumerable<DataRecord> records)
private async Task FlushAsync(string dbName, IEnumerable<DataRecord> records)
{
var count = 0;
await using var output = new MySqlDestination(
_outputOptions.Value.ConnectionString ?? throw new InvalidOperationException("Connection string is required"),
_logger, _context, _transformOptions, _errorRecorder);
var connStr = _outputOptions.Value.ConnectionString ?? throw new InvalidOperationException("ConnectionString is null");
await using var output = new MySqlDestination($"{connStr};Database={dbName};", _logger, _context, _transformOptions, _errorRecorder);
//if (records == null || records.Count() == 0) return;
//var dbName = $"cferp_test_1";
//if (records != null && records.Count() > 0)

1
ConsoleApp2/HostedServices/TaskMonitorService.cs
View File

@ -49,7 +49,6 @@ public class TaskMonitorService : BackgroundService
bool endCheck = false;
while (true)
{
if (_context.GetExceptions().Count>0) return;
EndCheck:
// var running = 0;
// var error = 0;

12
ConsoleApp2/HostedServices/TransformService.cs
View File

@ -43,16 +43,10 @@ public class TransformService : ITransformService
{
while ((!context.IsInputCompleted || producerQueue.Count > 0))
{
if (_context.GetExceptions().Count > 0)
{
_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
@ -63,14 +57,17 @@ public class TransformService : ITransformService
{
record = replaceRecord;
}
//计算需要分流的数据库
record.Database = _options.Value.DatabaseFilter.Invoke(record);
consumerQueue.Enqueue(record);
_context.AddTransform();
//数据增加
var addRecords = _options.Value.RecordAdd?.Invoke(record);
if (addRecords != null && addRecords.Count > 0)
if (addRecords is { Count: > 0 })
{
foreach (var rc in addRecords)
{
rc.Database = _options.Value.DatabaseFilter.Invoke(record);
consumerQueue.Enqueue(rc);
_context.AddTransform();
}
@ -78,6 +75,7 @@ public class TransformService : ITransformService
}
context.CompleteTransform();
},tasksOptions.TransformTaskCount,cancellationToken);
await _taskManager.WaitAll();
_logger.LogInformation("***** Data transformation service completed *****");
}
}

23
ConsoleApp2/Options/TenantDbOptions.cs Normal file
View File

@ -0,0 +1,23 @@
namespace ConsoleApp2.Options;
public class TenantDbOptions
{
public string TenantKey { get; set; }
/// <summary>
/// Key-Value: {DbName}-{TenantKeyLessThan}
/// </summary>
public Dictionary<string, int> DbList { get; set; }
public string GetDbNameByTenantKeyValue(int tenantKeyValue)
{
// var dictionary = new SortedDictionary<int, string>();
// DbList.ForEach(pair => dictionary.Add(pair.Value, pair.Key));
// 注意配置顺序
var dbName = DbList.Cast<KeyValuePair<string, int>?>()
.FirstOrDefault(pair => pair?.Value != null && pair.Value.Value > tenantKeyValue)!.Value.Key;
return dbName ??
throw new ArgumentOutOfRangeException(nameof(tenantKeyValue),
$"已配置的数据库中没有任何符合'{nameof(tenantKeyValue)}'值的对象");
}
}

8
ConsoleApp2/Program.cs
View File

@ -122,7 +122,13 @@ async Task RunProgram()
host.Services.Configure<DataTransformOptions>(options =>
{
options.DatabaseFilter = record => "cferp_test";
var tenantDbOptions = host.Configuration.GetRequiredSection("TenantDb").Get<TenantDbOptions>()
?? throw new ApplicationException("分库配置项不存在");
options.DatabaseFilter = record =>
{
var companyId = int.Parse(record[tenantDbOptions.TenantKey]); // 每个实体都应存在CompanyID，否则异常
return tenantDbOptions.GetDbNameByTenantKeyValue(companyId);
};
options.TransformBinary = field => commandOptions != null && commandOptions.Isutf8mb4 ? $"_utf8mb4 0x{field}" : $"0x{field}";
//数据过滤

6
ConsoleApp2/Services/MySqlDestination.cs
View File

@ -76,8 +76,7 @@ public partial class MySqlDestination : IDisposable, IAsyncDisposable
}
catch (Exception e)
{
_logger.LogCritical(e, "Error when flushing records, sql: {Sql}", cmd.CommandText.Omit(1000));
_context.AddException(e);
_logger.LogError(e, "Error when flushing records, sql: {Sql}", cmd.CommandText.Omit(1000));
var match = MatchTableName().Match(cmd.CommandText);
if (match is { Success: true, Groups.Count: > 1 })
@ -92,8 +91,7 @@ public partial class MySqlDestination : IDisposable, IAsyncDisposable
}
catch (Exception e)
{
_logger.LogCritical(e, "Error when serialize records, record:");
_context.AddException(e);
_logger.LogError(e, "Error when serialize records, record:");
}
finally
{

10
ConsoleApp2/Services/ProcessContext.cs
View File

@ -8,7 +8,6 @@ public class ProcessContext
private int _inputCount;
private int _transformCount;
private int _outputCount;
private IList<Exception> _exceptionList = new List<Exception>();
public bool IsInputCompleted { get; private set; }
public bool IsTransformCompleted { get; private set; }
public bool IsOutputCompleted { get; private set; }
@ -30,14 +29,7 @@ public class ProcessContext
get => _outputCount;
private set => _outputCount = value;
}
public void AddException(Exception ex)
{
_exceptionList.Add(ex);
}
public IList<Exception> GetExceptions()
{
return _exceptionList;
}
public void CompleteInput() => IsInputCompleted = true;
public void CompleteTransform() => IsTransformCompleted = true;

7
ConsoleApp2/Services/TaskManager.cs
View File

@ -21,13 +21,14 @@ public class TaskManager
_logger = logger;
}
public void CreateTask<TResult>(Func<TResult> func, CancellationToken cancellationToken = default)
public void CreateTask(Func<Task> func, CancellationToken cancellationToken = default)
{
var task = Task.Factory.StartNew(func, cancellationToken);
var task = Task.Run(func, cancellationToken);
_tasks.Add(task);
_logger.LogDebug("New task created");
}
public void CreateTasks<TResult>(Func<TResult> func,int taskCount, CancellationToken cancellationToken = default)
public void CreateTasks(Func<Task> func,int taskCount, CancellationToken cancellationToken = default)
{
for (int i = 0; i < taskCount; i++)
{

17
ConsoleApp2/appsettings.json
View File

@ -9,10 +9,25 @@
"OldestTime": "202301"
},
"ConnectionStrings": {
"MySqlMaster": "Server=127.0.0.1;Port=33309;UserId=root;Password=123456;Database=cferp_test;"
"MySqlMaster": "Server=127.0.0.1;Port=33309;UserId=root;Password=123456;" // 要分库，不用加'Database='了
},
"RedisCacheOptions": {
"Configuration": "192.168.1.246:6380",
"InstanceName" : "mes-etl:"
},
"TenantDb": // 分库配置
{
"TenantKey" : "CompanyID",
"DbList": {
/*
* 相当于
* (CompanyId < 1000) -> cferp_test_1
* (CompanyId < 2000) -> cferp_test_2
* (CompanyId < 2147483647) -> cferp_test_3
*/
"cferp_test_1": 1000,
"cferp_test_2": 2000,
"cferp_test_3": 2147483647
}
}
}