在并行处理期间维护 DataTable 行的顺序

Question

以下是当前代码：

 Parallel.ForEach(dataTable.AsEnumerable(),row => {

     // Code to process the data row to Dictionary<object,object>
     // Unique Column name is the Dictionary Key
     // ConcurrentDictionary is used for thread safety      
     });

这里我使用Parallel.ForEach将DataTable的行处理成Dictionary<object,object>类型的对象，最终结果是List<Dictionary<object,object>>类型，使用中间线程安全结构ConcurrentQueue<Dictionary<object,object>>实现的，@的来源987654327@ 以给定的顺序对数据进行排序，但在并行处理期间总是会丢失。由于顺序很重要，所以我想出了以下解决方法：

Parallel.For(0,RowCount,index => {

  int rowIndex = index;

  // Access the rows using the Index
  // Final structure will be of type ConcurrentDictionary<Custom>, 
  // with the RowIndex assigned based on original index
});

Class Custom
{
  public int RowIndex { get; set; }

  public Dictionary<object,object> DataDictionary {get; set;}
}

ConcurrentQueue<Dictionary<Custom>> customObj 类型的最终结果使用以下代码进行处理：

customObj.OrderBy(x=>x.RowIndex).Select(y=>y.DataDictionary).ToList()

以下是我的问题：

1. 有没有更好的方法来实现相同的并行处理，我可以保持原始顺序，这是最重要的业务需求

2. 在最终解决方案中我是否需要局部变量rowIndex，我的理解是index 是并行循环的一部分，不会导致关闭问题

任何指针？

Answer 1

您可以将PLINQ 与ParallelEnumerable.AsOrdered 扩展方法一起使用

允许将数据源视为已排序，覆盖默认的无序。

在您的示例中，您可以按以下方式使用它：

var result = dataTable.AsEnumerable().AsParallel().AsOrdered()
                      .Select(/*Process the row to dictionary*/).ToList();

Answer 2

这个呢

var items = new ConcurrentDictionary<DataRow, Dictionary<object,object>>;

Parallel.ForEach(dataTable.AsEnumerable(),row => {
    var result = ...; 
    items.Add(row, result);
});

var finalResult = dataTable.Rows.Cast<DataRow>().Select(r => items[r]).ToList());

Answer 3

首先，您可以在 Parallel.ForEach 中获取索引，而不是使用 Parallel.For

Parallel.ForEach(dataTable.AsEnumerable(), (line, state, index) =>
{
    Console.WriteLine("{0} : {1}", index, line);
});

如我所见，主要目的是避免 OrderBy。 要实现这一点，请在 ForLoop 之前创建您的

var lines =  new YourClass[NumberOfElemnts] ;

在此之后，您可以在此列表中填写您想要的任何循环。让我们使用 Parallel.For

Parallel.For(0, NumberOfElemnts, i =>
    {
        lines[i]=dataTable[i];
    });

根据@Panagiotis Kanavos 的评论进行编辑