.NET Core 3.1中Task.Run与RunSynchronously处理GBQ批量插入的问题

一、代码问题根源 #

1. 闭包变量捕获陷阱 #

你用Task.Run时出现批次ID混乱、数据丢失，核心是闭包捕获了循环中的可变变量：

• 循环里的i、start、size是可变的，而Task.Run的lambda是异步执行的，当任务真正启动时，循环已经执行多轮甚至结束，此时捕获到的变量值早已不是创建任务时的原值。
• 比如批次ID显示为18，是因为循环结束后i的值变成了17，所有异步任务都捕获了这个最终值，i+1就成了18；数据丢失（94、95行）是因为最后一个批次修改了size为4，前面的任务捕获到修改后的size，本该取6条数据的批次只取了4条，导致中间两行遗漏。

2. 未等待任务完成 #

代码创建了Task数组，但没有等待所有任务执行完毕，Main方法可能在任务运行结束前就退出，加剧了数据丢失的问题。

3. 异步方法未await #

GbqTable.InsertRowsAsync是异步方法，直接调用不await会导致插入操作还没完成，任务就提前结束，可能出现数据插入不完整或失败。

二、修复后的代码示例 #

using System;
using System.Collections;
using System.Threading.Tasks;

public class Program
{
    public static async Task Main()
    {
        ArrayList forecasts = new ArrayList();
        for (var k = 0; k < 100; k++)
        {
            forecasts.Add(k);
        }

        int batchSize = 6;
        var taskNum = (int)Math.Ceiling(forecasts.Count / (double)batchSize);
        Console.WriteLine("task number:" + taskNum);
        Console.WriteLine("item number:" + forecasts.Count);
        Task[] tasks = new Task[taskNum];

        for (int i = 0; i < taskNum; i++)
        {
            // 创建局部变量保存当前批次的参数，避免闭包陷阱
            int currentBatchId = i + 1;
            int currentStart = i * batchSize;
            int currentSize = Math.Min(batchSize, forecasts.Count - currentStart);

            tasks[i] = Task.Run(async () =>
            {
                var batchedforecastRows = forecasts.GetRange(currentStart, currentSize);
                // await异步插入方法，确保操作完成
                await GbqTable.InsertRowsAsync(batchedforecastRows);
                Console.WriteLine($"batchID:{currentBatchId} Inserted:[{string.Join(",", batchedforecastRows.ToArray())}]");
            });
        }

        // 等待所有任务执行完成
        await Task.WhenAll(tasks);
        Console.WriteLine("所有批次插入完成");
    }

    // 模拟GBQ插入方法
    private static class GbqTable
    {
        public static Task InsertRowsAsync(ArrayList rows)
        {
            // 实际GBQ插入逻辑
            return Task.Delay(10);
        }
    }
}

三、Task.Run是否适合你的场景？ #

完全适合，但要注意以下细节：

• 控制并发数：3-5亿行数据如果开过多任务，会触发GBQ限流或网络拥堵，建议用SemaphoreSlim限制同时执行的插入任务数（比如10-20个），避免超出配额。
• 优先用原生异步API：如果InsertRowsAsync是真正的IO异步方法，不需要用Task.Run包裹，直接调用并await即可；Task.Run更适合处理CPU密集型同步操作。
• 优化批次大小：GBQ建议每个批次1-10万行（或数据量不超10MB），根据Avro行大小调整批次，避免过大导致超时、过小导致请求过多。
• 添加错误处理：在任务中加入try-catch捕获异常，记录失败批次信息方便后续重试：

  tasks[i] = Task.Run(async () =>
  {
      try
      {
          var batchedforecastRows = forecasts.GetRange(currentStart, currentSize);
          await GbqTable.InsertRowsAsync(batchedforecastRows);
          Console.WriteLine($"batchID:{currentBatchId} 插入成功");
      }
      catch (Exception ex)
      {
          Console.WriteLine($"batchID:{currentBatchId} 插入失败: {ex.Message}");
          // 记录失败日志，后续可重试
      }
  });
  

四、超大规模数据的优化建议 #

针对3-5亿行的量级，API批量插入效率有限，推荐更高效的方案：

• GCS+GBQ批量导入：先将Avro文件上传到Google Cloud Storage，再通过GBQ的导入作业从GCS加载数据，这是超大规模数据导入的最优方式。
• 使用官方SDK的流式/批量工具：Google.Cloud.BigQuery.V2库提供BigQueryClient.CreateInsertStream等专用批量插入接口，比手动拆分批次更可靠高效。

内容的提问来源于stack exchange，提问作者Zichen Ma