如何异步执行DataRow转POCO并添加至List<T>的操作？

问题场景 #

查询返回约11000行数据存入DataTable后，通过自定义ORM的ToCustomObject<T>方法将每行映射为POCO对象，这一步循环操作耗时数秒（远慢于查询本身的0.5秒）。现有代码是同步遍历添加到List，想通过并行/异步方式提速，尝试过Task.WhenAll但List.Add不是异步操作，对Parallel.ForEach的适用性存疑。

解决方案 #

针对CPU密集型的POCO映射操作，分两种可行方案实现并行处理：

方案1：使用Parallel.ForEach（推荐，CPU密集型场景更高效）

Parallel.ForEach专为CPU密集的并行任务设计，会自动根据系统核心数调整并发度，避免过度线程切换。注意普通List不是线程安全的，可以用ConcurrentBag或者本地分区的方式避免线程冲突：

基于ConcurrentBag的简化版

public async Task<List<Foo>> GetFoosAsync()
{
    string query = "select t.Field1, t.Field2 from Table";
    DataTable results = await customDbHandlerThatsNotRelevant.ExecuteQueryAsync(query);

    var concurrentFoos = new ConcurrentBag<Foo>();
    Parallel.ForEach(results.Rows.Cast<DataRow>(), row =>
    {
        concurrentFoos.Add(row.ToCustomObject<Foo>());
    });

    return concurrentFoos.ToList();
}

本地分区优化版（减少锁竞争，性能更好）

通过给每个线程分配本地列表，最后再合并，大幅降低线程间的锁开销：

public async Task<List<Foo>> GetFoosAsync()
{
    string query = "select t.Field1, t.Field2 from Table";
    DataTable results = await customDbHandlerThatsNotRelevant.ExecuteQueryAsync(query);

    // 预先分配容量，避免扩容开销
    var foos = new List<Foo>(results.Rows.Count);
    
    Parallel.ForEach(results.Rows.Cast<DataRow>(), 
        // 每个工作线程初始化本地列表
        () => new List<Foo>(),
        // 处理单条数据，添加到本地列表
        (row, _, localList) =>
        {
            localList.Add(row.ToCustomObject<Foo>());
            return localList;
        },
        // 合并本地列表到全局列表
        localList =>
        {
            lock (foos)
            {
                foos.AddRange(localList);
            }
        });

    return foos;
}

方案2：使用Task.WhenAll（适合习惯异步编程模型的场景）

先把所有映射操作包装成Task，等待全部完成后一次性生成List，无需处理线程安全问题：

public async Task<List<Foo>> GetFoosAsync()
{
    string query = "select t.Field1, t.Field2 from Table";
    DataTable results = await customDbHandlerThatsNotRelevant.ExecuteQueryAsync(query);

    // 将每行的映射操作包装为Task
    var tasks = results.Rows.Cast<DataRow>()
        .Select(row => Task.Run(() => row.ToCustomObject<Foo>()))
        .ToList();

    // 等待所有任务完成，返回结果数组转List
    Foo[] fooArray = await Task.WhenAll(tasks);
    return fooArray.ToList();
}

关键注意点 #

• 如果ToCustomObject<T>有对应的异步版本（比如ToCustomObjectAsync<T>），优先使用异步版本配合Task.WhenAll，这才是真正的IO/CPU混合场景异步，不会占用线程池线程。
• 并行处理仅适用于纯CPU密集型操作，如果映射过程中包含IO操作（比如额外查库），需要调整为异步优先的方案。
• 不要在并行循环中直接调用List.Add，会引发线程安全问题，导致数据丢失或异常。

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