Я хочу разработать пакетный процесс в .NET Core для синхронизации файла CSV, содержащего 3 миллиона строк. Процесс считывает данные из файла, обновляет существующие элементы в базе данных и создает новые элементы, если они не существуют. Однако в настоящее время эта операция занимает значительное количество времени; например, обработка 100 000 строк занимает 30 минут. Я использовал параллельные задачи для повышения производительности, но все еще сталкиваюсь с проблемами задержки. Есть ли какое-нибудь решение сократить время обработки?
Ниже моего кода:
public async Task ImportDataFromCsvFile()
{
try
{
// Read CSV file asynchronously
List data = await _azureBlobService.ReadCsvFileAsync();
List Invoices = new List();
// Parallel processing of data to create InvoiceAutoDb list
Parallel.ForEach(data, Invoice =>
{
var entity = new InvoiceDb
{
Id_Invoice = Invoice.IdInvoice,
DateCheck = Invoice.DateCheck
};
lock (Invoices) // Ensure thread-safety when adding to the list
{
Invoices.Add(entity);
}
});
// Get distinct Invoices
var distinctInvoice = Invoices
.GroupBy(c => c.Id_Invoice)
.Where(x => x.Count() == 1)
.Select(g => g.First())
.ToList();
var nonDistinctInvoice = Invoices
.Except(distinctInvoice)
.ToList();
if (nonDistinctInvoice.Any())
{
Log.Warning("Invoice Dupliqué(s) {duplicata}", nonDistinctInvoice.Select(c => c.Id_Invoice).Distinct());
}
// Use Parallel.ForEachAsync for parallel asynchronous processing
//this is the part where it consume time
await Parallel.ForEachAsync(distinctInvoice, async (Invoice, cancellationToken) =>
{
using (var scope = _serviceProvider.CreateScope())
{
var service = scope.ServiceProvider.GetRequiredService();
var existingInvoice = await service.GetByNumeroInvoiceAsync(Invoice.Id_Invoice);
if (existingInvoice == null)
{
await service.ImportNewInvoiceAsync(Invoice);
}
else
{
existingInvoice.DateCheck = Invoice.DateCheck;
await service.UpdateImportedInvoiceAsync(existingInvoice);
}
await service.SaveChangesAsync();
}
});
Log.Information("{result} line imported successfully", distinctInvoice.Count);
// Publish to Kafka
await PublishInvoice(distinctInvoice);
return true;
}
catch (Exception ex)
{
Log.Error("Import Data failed {errorMessage}", ex.Message);
throw new Exception(ex.Message);
}
}
`
Подробнее здесь: https://stackoverflow.com/questions/787 ... otnet-core
