快捷導(dǎo)航

C#實(shí)現(xiàn)千萬數(shù)據(jù)秒級(jí)導(dǎo)入的代碼

更新時(shí)間：2025年09月12日 09:12:46 作者：李袁明

在實(shí)際開發(fā)中excel導(dǎo)入很常見,現(xiàn)代社會(huì)中很容易遇到大數(shù)據(jù)處理業(yè)務(wù),所以本文我就給大家分享一下千萬數(shù)據(jù)秒級(jí)導(dǎo)入怎么實(shí)現(xiàn),文中有詳細(xì)的代碼示例供大家參考,需要的朋友可以參考下

前言

在實(shí)際開發(fā)中excel導(dǎo)入很常見，現(xiàn)代社會(huì)中很容易遇到大數(shù)據(jù)處理業(yè)務(wù)，這里我就給大家分享一下千萬數(shù)據(jù)秒級(jí)導(dǎo)入怎么實(shí)現(xiàn)

一、數(shù)據(jù)存儲(chǔ)

這里使用到的數(shù)據(jù)庫是clickhouse，因?yàn)橄駇ysql、sqlserver這類關(guān)系型數(shù)據(jù)庫對于大數(shù)據(jù)的支持不怎么好所以這里我使用的是clickhouse,它是由俄羅斯的yindex公司開發(fā)的列式存儲(chǔ)數(shù)據(jù)庫。查詢語句這些和mysql差不多上手容易學(xué)起來簡單。需要注意的是clickhouse對數(shù)據(jù)的刪除比較嚴(yán)格一般都不會(huì)去刪除它里面的數(shù)據(jù)庫，如果要?jiǎng)h除數(shù)據(jù)，我們一般都是采用用一個(gè)字段標(biāo)記數(shù)據(jù)是否被刪除，然后過濾掉這些被刪除的數(shù)據(jù)，clickhouse的四種引擎也可以去了解一下

二、處理邏輯

優(yōu)化前導(dǎo)入耗時(shí)兩分鐘以上優(yōu)化后導(dǎo)入耗時(shí)30秒以內(nèi)

優(yōu)化前代碼處理邏輯

優(yōu)化后導(dǎo)入一千萬數(shù)據(jù)大約需要兩分鐘完成處理。處理邏輯是將數(shù)據(jù)循環(huán)裝入List中，當(dāng)數(shù)據(jù)量達(dá)到預(yù)設(shè)的chunkSize閾值（150萬）時(shí)開始批量插入。值得注意的是，1000萬數(shù)據(jù)需要循環(huán)處理7次。這里存在一個(gè)常見誤區(qū)：認(rèn)為提高150萬的閾值能加快處理速度。實(shí)際上，處理速度慢的原因并非循環(huán)次數(shù)，而是由于單線程運(yùn)行無法充分發(fā)揮ClickHouse的高并發(fā)優(yōu)勢。

private async Task InsertContentsAsync2(string filePath, BlackListCreateDto dto, ulong maxId, long batchId)
 {
     Stopwatch stopwatch = new();
     var regex = RegexConst.Phone();
     var userId = CurrentUser.Id.ToString();
     var isTxt = filePath.Split(".").Last() == "txt";
     var chunkSize = 1500000;
     List<string> blackLists = new();
     int offset = 0;
     int total = 0;
     stopwatch.Start();
     if (isTxt)
     {
         using StreamReader sr = new(filePath);
         string line;
         while ((line = await sr.ReadLineAsync()) != null)
         {
             if (regex.IsMatch(line))
             {
                 blackLists.Add(line);
                 offset++;
                 total++;
             }
             if (offset >= chunkSize)
             {
                 await InsertDataAsync(blackLists);
                 offset = 0;
                 blackLists.Clear();
             }
         }

         if (offset >= 0)
         {
             await InsertDataAsync(blackLists);
             offset = 0;
             blackLists.Clear();
         }
     }
     else
     {

         var rows = await MiniExcel.QueryAsync(filePath);
         foreach (IEnumerable<dynamic> row in rows)
         {
             var pair = (IDictionary<string, object>)row;
             var cells = pair.Values;
             if (cells != null && cells.Any())
             {
                 foreach (var cell in cells)
                 {
                     if (cell is not null && cell is string str)
                     {
                         blackLists.Add(str);
                         offset++;
                         total++;
                     }

                 }
             }
             if (offset >= chunkSize)
             {
                 await InsertDataAsync(blackLists);
                 offset = 0;
                 blackLists.Clear();
             }
         }

         if (offset >= 0)
         {
             await InsertDataAsync(blackLists);
             offset = 0;
             blackLists.Clear();
         }
     }

     async Task InsertDataAsync(List<string> blackLists)
     {
         List<BlackList> list = blackLists.Select(it => CreateBlackList(dto, batchId, ++maxId, userId, it.ToString())).ToList();
         var result = await _clickHouseClientRepository.BulkInsertAsync(list, _clickHouseTables.Value.SMS_BLACK_LIST_TABLE);
         blackLists.Clear();
     }
     stopwatch.Stop();
     Logger.LogInformation("文件路徑:{filePath},{total}條數(shù)據(jù), 耗時(shí):{Elapsed} ", filePath, total, stopwatch.Elapsed);
 }

優(yōu)化后的代碼

優(yōu)化后的代碼將單線程執(zhí)行模式改為多消費(fèi)者并行處理，采用生產(chǎn)者-消費(fèi)者模式分離IO和計(jì)算操作。文件讀取時(shí)使用最大緩沖區(qū)和順序掃描：

await using var fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, 81920, FileOptions.SequentialScan);

關(guān)鍵優(yōu)化點(diǎn)包括：

使用異步文件讀取(FileStream異步API)
預(yù)分配列表內(nèi)存(new List(chunkSize))
采用ClickHouse批量并行提交而非單條插入
設(shè)置chunkSize為10萬條
使用BoundedChannel進(jìn)行管道容量控制

關(guān)于管道容量控制的作用：

防止內(nèi)存溢出：限制管道中積壓的未處理數(shù)據(jù)量，避免生產(chǎn)者過快導(dǎo)致內(nèi)存問題
實(shí)現(xiàn)自動(dòng)背壓(Backpressure)：管道滿時(shí)生產(chǎn)者會(huì)自動(dòng)阻塞或丟棄數(shù)據(jù)（取決于配置），直到消費(fèi)者處理部分?jǐn)?shù)據(jù)

選擇10萬而非150萬的原因：

原因1：內(nèi)存考量

150萬條數(shù)據(jù)（假設(shè)每條20字節(jié)）單批次占用約30MB
并行處理時(shí)多個(gè)批次會(huì)顯著增加內(nèi)存壓力，可能觸發(fā)GC影響性能
10萬條僅占用約2MB，對GC更友好，適合長期批處理

原因2：ClickHouse性能

BulkInsert在5萬~10萬批次時(shí)吞吐量最佳

原因3：并行處理效率

10萬條/批次能更好分配任務(wù)到多線程/核心
150萬條/批次可能導(dǎo)致任務(wù)分配不均

原因4：I/O優(yōu)化

大數(shù)據(jù)批次可能導(dǎo)致I/O阻塞，CPU空閑等待
小批次(10萬)能更好重疊I/O和計(jì)算，提升吞吐量

private async Task InsertContentsAsync(string filePath, BlackListCreateDto dto, ulong maxId, long batchId)
 {
     var stopwatch = Stopwatch.StartNew();
     var regex = RegexConst.Phone();
     var userId = CurrentUser.Id.ToString();
     var isTxt = filePath.EndsWith(".txt", StringComparison.OrdinalIgnoreCase);
     const int chunkSize = 100_000; 
     var totalProcessed = 0L;

     // 并行處理管道
     var processingChannel = Channel.CreateBounded<List<string>>(new BoundedChannelOptions(5)
     {
         SingleWriter = true,
         SingleReader = false,
         FullMode = BoundedChannelFullMode.Wait
     });

     // 啟動(dòng)并行消費(fèi)者
     var processingTasks = Enumerable.Range(0, Environment.ProcessorCount)
         .Select(_ => Task.Run(async () =>
         {
             await foreach (var batch in processingChannel.Reader.ReadAllAsync())
             {
                 await ProcessBatchAsync(batch);
                 Interlocked.Add(ref totalProcessed, batch.Count);
             }
         })).ToArray();

     try
     {
         if (isTxt)
         {
             await ProcessTextFileAsync(filePath, regex, processingChannel.Writer, chunkSize);
         }
         else
         {
             await ProcessExcelFileAsync(filePath, processingChannel.Writer, chunkSize);
         }
     }
     finally
     {
         processingChannel.Writer.Complete();
         await Task.WhenAll(processingTasks);
         stopwatch.Stop();

         Logger.LogInformation("文件路徑:{FilePath}, {Total}條數(shù)據(jù), 耗時(shí):{Elapsed}ms",
             filePath, totalProcessed, stopwatch.ElapsedMilliseconds);
     }

     async Task ProcessBatchAsync(List<string> batch)
     {
         var entities = batch
             .Select(phone => CreateBlackList(dto, batchId, ++maxId, userId, phone))
             .ToList();

         await _clickHouseClientRepository.BulkInsertAsync(
             entities,
             _clickHouseTables.Value.SMS_BLACK_LIST_TABLE);
     }
 }

 private async Task ProcessTextFileAsync(string filePath, Regex regex, ChannelWriter<List<string>> writer, int chunkSize)
 {
     await using var fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, 81920, FileOptions.SequentialScan);
     using var sr = new StreamReader(fs);

     var batch = new List<string>(chunkSize);
     string line;

     while ((line = await sr.ReadLineAsync()) != null)
     {
         if (regex.IsMatch(line))
         {
             batch.Add(line);

             if (batch.Count >= chunkSize)
             {
                 await writer.WriteAsync(batch);
                 batch = new List<string>(chunkSize);
             }
         }
     }

     if (batch.Count > 0)
     {
         await writer.WriteAsync(batch);
     }
 }

 private async Task ProcessExcelFileAsync(string filePath, ChannelWriter<List<string>> writer, int chunkSize)
 {
     await using var stream = File.Open(filePath, FileMode.Open, FileAccess.Read);
     using var reader = MiniExcel.QueryAsDataTable(stream);
     var batch = new List<string>(chunkSize);
     foreach (DataRow row in reader.Rows)
     {
         var cell = row[0].ToString();
         if (cell is string str && !string.IsNullOrWhiteSpace(str))
         {
             batch.Add(str);

             if (batch.Count >= chunkSize)
             {
                 await writer.WriteAsync(batch);
                 batch = new List<string>(chunkSize);
             }
         }
     }

     if (batch.Count > 0)
     {
         await writer.WriteAsync(batch);
     }
 }

總結(jié)

到此這篇關(guān)于C#實(shí)現(xiàn)千萬數(shù)據(jù)秒級(jí)導(dǎo)入的代碼的文章就介紹到這了,更多相關(guān)C#千萬數(shù)據(jù)秒級(jí)導(dǎo)入內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: