将线程安全集合转换为 DataTable 的最佳方法？

Question

所以这里是场景：

我必须获取一组数据，对其进行处理，构建一个对象，然后将这些对象插入到数据库中。

为了提高性能，我使用并行循环对数据进行多线程处理，并将对象存储在 CollectionBag 列表中。

那部分工作正常。但是，这里的问题是我现在需要获取该列表，将其转换为 DataTable 对象并将数据插入数据库。这很丑陋，我觉得我没有以最好的方式做到这一点（下面的伪）：

ConcurrentBag<FinalObject> bag = new ConcurrentBag<FinalObject>();

ParallelOptions parallelOptions = new ParallelOptions();
parallelOptions.MaxDegreeOfParallelism = Environment.ProcessorCount;

Parallel.ForEach(allData, parallelOptions, dataObj =>
{   
    .... Process data ....

    bag.Add(theData);

    Thread.Sleep(100);
});

DataTable table = createTable();
foreach(FinalObject moveObj in bag) {
    table.Rows.Add(moveObj.x);
}

Answer 1

这是 PLINQ（或 Rx - 我将专注于 PLINQ，因为它是基类库的一部分）的理想选择。

IEnumerable<FinalObject> bag = allData
    .AsParallel()
    .WithDegreeOfParallelism(Environment.ProcessorCount)
    .Select(dataObj =>
    {
        FinalObject theData = Process(dataObj);

        Thread.Sleep(100);

        return theData;
    });

DataTable table = createTable();

foreach (FinalObject moveObj in bag)
{
    table.Rows.Add(moveObj.x);
}

实际上，您不应通过Thread.Sleep 限制循环，而应进一步限制最大并行度，直到将 CPU 使用率降至所需水平。

免责声明：以下所有内容仅供娱乐，尽管它确实确实有效。

当然，您总是可以将它提升一个档次并生成一个完整的异步Parallel.ForEach 实现，允许您并行处理输入并异步进行节流，而不会阻塞任何线程池线程。

async Task ParallelForEachAsync<TInput, TResult>(IEnumerable<TInput> input,
                                                 int maxDegreeOfParallelism,
                                                 Func<TInput, Task<TResult>> body,
                                                 Action<TResult> onCompleted)
{
    Queue<TInput> queue = new Queue<TInput>(input);

    if (queue.Count == 0) {
        return;
    }

    List<Task<TResult>> tasksInFlight = new List<Task<TResult>>(maxDegreeOfParallelism);

    do
    {
        while (tasksInFlight.Count < maxDegreeOfParallelism && queue.Count != 0)
        {
            TInput item = queue.Dequeue();
            Task<TResult> task = body(item);

            tasksInFlight.Add(task);
        }

        Task<TResult> completedTask = await Task.WhenAny(tasksInFlight).ConfigureAwait(false);

        tasksInFlight.Remove(completedTask);

        TResult result = completedTask.GetAwaiter().GetResult(); // We know the task has completed. No need for await.

        onCompleted(result);
    }
    while (queue.Count != 0 || tasksInFlight.Count != 0);
}

用法（full Fiddle here）：

async Task<DataTable> ProcessAllAsync(IEnumerable<InputObject> allData)
{
    DataTable table = CreateTable();
    int maxDegreeOfParallelism = Environment.ProcessorCount;

    await ParallelForEachAsync(
        allData,
        maxDegreeOfParallelism,
        // Loop body: these Tasks will run in parallel, up to {maxDegreeOfParallelism} at any given time.
        async dataObj =>
        {
            FinalObject o = await Task.Run(() => Process(dataObj)).ConfigureAwait(false); // Thread pool processing.

            await Task.Delay(100).ConfigureAwait(false); // Artificial throttling.

            return o;
        },
        // Completion handler: these will be executed one at a time, and can safely mutate shared state.
        moveObj => table.Rows.Add(moveObj.x)
    );

    return table;
}

struct InputObject
{
    public int x;
}

struct FinalObject
{
    public int x;
}

FinalObject Process(InputObject o)
{
    // Simulate synchronous work.
    Thread.Sleep(100);

    return new FinalObject { x = o.x };
}

相同的行为，但没有 Thread.Sleep 和 ConcurrentBag<T>。

Answer 2

听起来您通过使所有内容并行运行使事情变得相当复杂，但是如果您将DataRow obejcts 存储在您的包中而不是普通对象，最后您可以使用DataTableExtensions 创建一个DataTable 很容易来自通用集合：

var dataTable = bag.CopyToDataTable();

只需在您的项目中添加对System.Data.DataSetExtensions 的引用即可。

Answer 3

我认为这样的东西应该提供更好的性能，看起来 object[] 是比 DataRow 更好的选择，因为您需要 DataTable 来获取 DataRow 对象。

ConcurrentBag<object[]> bag = new ConcurrentBag<object[]>();

Parallel.ForEach(allData, 
    new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount }, 
    dataObj =>
{
    object[] row = new object[colCount];

    //do processing

    bag.Add(row);

    Thread.Sleep(100);
});

DataTable table = createTable();
foreach (object[] row in bag)
{
    table.Rows.Add(row);
}