标签: parallel.foreach

我正在转换现有应用程序以利用多个处理器。我有一些嵌套循环，并且我已将最内层循环转换为 Parallel.Foreach 循环。在原始应用程序中，在最内层循环内，代码将调用DataTable.NewRow()实例化适当布局的新 DataRow、填充列并将填充的 DataRow 添加到带有 的 DataTable 中DataTable.Add()。但由于 DataTable 仅对于读取操作是线程安全的，因此我已转换处理以将填充的 DataRow 对象添加到对象中ConcurrentBag<DataRow>。然后，一旦 Parallel.Foreach 循环完成，我将迭代 ConcurrentBag 并将 DataRow 对象添加到 DataTable 中。它看起来像这样......

DataTable MyDataTable = new DataTable()
// Add columns to the data table

For(int OuterLoop = 1; OuterLoop < MaxValue; OuterLoop++)
{
    //Do Stuff...

    ConcurrentBag<DataRow> CB = new ConcurrentBag<DataRow>();

    Parallel.Foreach(MyCollectionToEnumerate, x => 
    {
        //Do Stuff

        DataRow dr = MyDataTable.NewRow();
        // Populate dr...
        CB.Add(dr);
    {);

    ForEach(DataRow d in CB)
        MyDataTable.Add(d);
}

因此，当运行时，我看到“索引超出了数组的范围”。调用时出现异常MyDataTable.NewRow()。但是 NewRow() …

我有一个项目列表，对于每个项目，我需要执行几个异步 API 请求并处理响应，我已经看到了几个实现，它们在执行时间方面都很相似，但我想知道它们之间的区别。

方法一

Parallel.ForEach(items, new ParallelOptions { MaxDegreeOfParallelism = 10 }, item =>
        {
            var response = item.propertyOne.GetAsync().GetAwaiter().GetResult();
            //process it
            var response = item.propertyTwo.GetAsync().GetAwaiter().GetResult();
            //process it
            var response = item.propertyThree.GetAsync().GetAwaiter().GetResult();
            //process it
        });

方法2

Parallel.ForEach(items, new ParallelOptions { MaxDegreeOfParallelism = 10 }, item =>
        {
            Task.Run(async () =>
            {
                var response = await item.propertyOne.GetAsync();
            }).GetAwaiter().GetResult();
            //process it
            Task.Run(async () =>
            {
                var response = await item.propertyTwo.GetAsync();
            }).GetAwaiter().GetResult();
            //process it
            Task.Run(async () =>
            {
                var response = await item.propertyThreee.GetAsync();
            }).GetAwaiter().GetResult(); …

我有一个文本文件，我将其读取为字符串content。为了识别我想要进一步处理的文本主体，我获取字符串中关键字的索引，然后将“起始”索引设置为找到的最小索引。

我尝试过这个Parallel.ForEach...

ConcurrentBag<int> indexes = new();
int index;

switch (Case)
{
    case 1:
        Parallel.ForEach(KeywordTypes.GetImplementedNamedObjects(), inos =>
        {
            index = content.IndexOf($"/begin {inos}");
            index = index == -1 ? content.Length : index;
            indexes.Add(index);
        });
        index = indexes.Min();
        return index;

... 与foreach：

foreach (string inos in KeywordTypes.GetImplementedNamedObjects())
{
    index = content.IndexOf($"/begin {inos}");
    index = index == -1 ? content.Length : index;
    indexes.Add(index);
}

index = indexes.Min();
return index;

其中foreach产生预期结果但Parallel.ForEach没有产生。

为什么我的代码不是线程安全的？

该财产的文件ParallelOptions.MaxDegreeOfParallelism指出：

该属性会影响传递此实例的方法调用MaxDegreeOfParallelism运行的并发操作数。正的属性值将并发操作的数量限制为设定值。如果为-1，则并发运行的操作数没有限制。ParallelParallelOptions

默认情况下，For将ForEach利用底层调度程序提供的线程数量，因此更改MaxDegreeOfParallelism默认值只会限制将使用的并发任务数量。

我试图理解“无限制”在这种情况下意味着什么。根据以上文档摘录，我的期望是Parallel.Invoke配置的操作MaxDegreeOfParallelism = -1将立即开始并行执行所有提供的actions. 但事实并非如此。这是一个包含 12 个操作的实验：

int concurrency = 0;
Action action = new Action(() =>
{
    var current = Interlocked.Increment(ref concurrency);
    Console.WriteLine(@$"Started an action at {DateTime
        .Now:HH:mm:ss.fff} on thread #{Thread
        .CurrentThread.ManagedThreadId} with concurrency {current}");
    Thread.Sleep(1000);
    Interlocked.Decrement(ref concurrency);
});
Action[] actions = Enumerable.Repeat(action, 12).ToArray();
var options = new ParallelOptions() { MaxDegreeOfParallelism = -1 };
Parallel.Invoke(options, …

当我调试时Parallel.ForEach，我可以发现有字段CurrentIteration，但我找不到它ParallelLoopState。

如何获取CurrentIteration的值？

我试图使用dict中的值副本为每个循环生成线程.

我最初的理解是,这foreach将创造一个新的范围,并导致:

Dictionary<string, string> Dict = new Dictionary<string, string>() { { "sr1", "1" }, { "sr2", "2" } };
foreach (KeyValuePair<string, string> record in Dict) {
    new System.Threading.Timer(_ =>
    {
        Console.WriteLine(record.Value);
    }, null, TimeSpan.Zero, new TimeSpan(0, 0, 5));
}

写道

1
2
2
2

而不是(预期):

1
2
1
2

所以我尝试在foreach中克隆kvp:

KeyValuePair<string, string> tmp = new KeyValuePair<string, string>(record.Key, record.Value);

但这会产生相同的结果.

我也试过了,System.Parallel.ForEach但似乎需要非动态的价值,这对我的字典来说有点火车粉碎.

如何用线程迭代我的字典？

使用关于ParallelOptions.MaxDegreeOfParallelism 的MSDN文章中的代码,我尝试了以下内容......

ParallelOptions po = new ParallelOptions {
  MaxDegreeOfParallelism = 2
};
Parallel.ForEach(files, (currentFile) => {
  String filename = System.IO.Path.GetFileName(currentFile);
  Bitmap bitmap = new Bitmap(currentFile);
  bitmap.RotateFlip(RotateFlipType.Rotate180FlipNone);
  bitmap.Save(Path.Combine(newDir, filename));
  Console.WriteLine("Processing {0} on thread {1}", filename, Thread.CurrentThread.ManagedThreadId);
});

但是,查看输出的线程ID,我可以看到,如果我设置了MaxDegreeOfParallelism,它根本没有任何区别.看着我的CPU监视器,即使我将MaxDegreeOfParallelism设置为2,我也能看到所有核心在运行.

我错过了这里的观点吗？我以为这个想法是限制线程的数量？

C#中的以下代码块是否引入了竞争条件:

    Parallel.ForEach(guidDictionary, (dictionaryItem) =>
    {
        var fileName = dictionaryItem.Key;
        var fileText = File.ReadAllText(fileName, Encoding.ASCII);
        Parallel.ForEach(guidDictionary, (guidObj) =>
        {
            fileText = fileText.Replace(guidObj.Value.OldGuid, guidObj.Value.NewGuid);
        });

        File.WriteAllText(fileName, fileText);
    });

？

我已经完成了基本foreach循环,XmlNodeList如下所示.

示例XML文件(books.xml)

XmlDocument doc = new XmlDocument();
doc.Load("books.xml");
XmlNodeList xnList = doc.SelectNodes("catalog/book");
foreach (XmlNode node in xnList)
{
   Console.WriteLine(node["author"].InnerText);
} 

如何将此循环转换为Parallel.ForEach？

我试过这个代码.但它不起作用.

Parallel.ForEach(xnList, (XmlNode node) =>
{
   Console.WriteLine(node["author"].InnerText);
});

这是错误2

参数1:无法转换System.Xml.XmlNodeList为 System.Collections.Generic.IEnumerable<System.Xml.XmlNode>

我们有一个对象(XML或JSON),我们成功地将它映射到DTO,在我们的数据库中插入需要太长时间(5~7分钟),所以我们经历了Parallel.ForEach,但最终,我们注意到有一些数据输入不正确,就像Category所有具有相同名称的项目一样,但其他不同的属性是100%正确的,在其他情况下,我们注意到所有数据在一个类别中是相同的,但是,提供的JSON对象没有.

我承认它是如此之快,它需要不到一分钟,但插入错误,看看下面使用的代码:

JSON

[
  {
    "CategoryId": 1,
    "CategoryName": "Drinks",
    "SortOrder": 1,
    "Products": [
      {
        "ProductId": 100,
        "ProductName": "Black Tea",
        "SortOrder": 1,
        "Price": 5,
        "Choices": []
      },
      {
        "ProductId": 101,
        "ProductName": "Turkish Coffee",
        "SortOrder": 2,
        "Price": 7.5,
        "Choices": []
      },
      {
        "ProductId": 102,
        "ProductName": "Green Tea",
        "SortOrder": 3,
        "Price": 6,
        "Choices": []
      },
      {
        "ProductId": 103,
        "ProductName": "Café Latte Medium",
        "SortOrder": 4,
        "Price": 10,
        "Choices": []
      },
      {
        "ProductId": 104,
        "ProductName": "Orange Juice",
        "SortOrder": 5,
        "Price": 11, …