标签: parallel.foreach

谁能解释一下,为什么这个程序返回sqrt_min的正确值？

int n = 1000000;

double[] myArr = new double[n];
for(int i = n-1 ; i>= 0; i--){ myArr[i] = (double)i;}

// sqrt_min contains minimal sqrt-value
double sqrt_min = double.MaxValue;

Parallel.ForEach(myArr, num =>
{
double sqrt = Math.Sqrt(num); // some time consuming calculation that should be parallized
if(sqrt < sqrt_min){ sqrt_min = sqrt;}
});
Console.WriteLine("minimum: "+sqrt_min);

我刚开始使用TPL,我希望对Web服务进行多次并行调用.从我可以收集的内容中,我看到了两种方法.

要么Parallel.ForEach:

List<ServiceMemberBase> list = new List<ServiceMemberBase>(); //Take list from somewhere.
        Parallel.ForEach(list, member =>
            {
                var result = Proxy.Invoke(member);
                //...
                //Do stuff with the result
                //...
            });

或者Task<T>:

List<ServiceMemberBase> list = new List<ServiceMemberBase>(); //Take list from somewhere.
        ForEach(var member in list)
        {
            Task<MemberResult>.Factory.StartNew(() => proxy.Invoke(member));
        }

        //Wait for all tasks to finish.
        //Process the result objects.

无论语法是否正确,这些都是等同的吗？

他们会产生相同的结果吗？如果没有,为什么？哪个更好？

我正在使用Moq对包含Parallel.foreach循环的一些代码进行单元测试.

的排列相套最多4个例外的环内被抛出,然后包装在一个AggregateException.

这传递了我的i7处理器,我检查了代码.

之后,一位同事抱怨说他没有过世.事实证明Parallel.foreach,在轰炸之前,他的Core2duo上只产生了2个线程,因此只有2个例外被包含在内AggregateException.

问题是如何处理这个问题,以便单元测试不依赖于处理器架构？几点想法: -

1. 有一篇关于手动添加例外的微软文章,AggregateException但我们并不热衷于这样做,因为如果出现问题,循环应尽早退出.
2. ParallelOptions.MaxDegreeOfParallelism可以对使用的线程数设置上限.但除非将其调低为1(这似乎更像是作弊而不是正确的单元测试),单元测试如何知道实际使用了多少线程,从而设置了安排和断言阶段？

我的代码中有类似的东西:

Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
{
    Process(item);
});

问题是我在Process()方法中做了很多事情(连接到文件共享,解析文件,保存到数据库等),我担心在这个过程中可能会出现问题,使迭代永远不会完成...... 这可能是曾经发生过？

有没有办法为方法设置超时,Process()以避免最终拥有僵尸线程？

更新:

我发现设置超时的最简单方法是在任务上添加毫秒CancellationTokenSource或调用Wait()方法.

选项1

Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
{
    var cts = new CancellationTokenSource(2000);
    Task task = Task.Factory.StartNew(() => Process(item), cts.Token);
});

选项#2

Parallel.ForEach(myList, new ParallelOptions { MaxDegreeOfParallelism = 4 }, item =>
{
    Task task = Task.Factory.StartNew(() => Process(item));
    task.Wait(2000);
});

问题是这些选项都不能取消该Process()方法.我需要检查Process()方法中的某些内容吗？

我有一个Parallel.ForEach循环做一些处理.但是第一个操作是在未包含密钥的情况下在字典中添加值.添加它时出错,它表示密钥已经在字典中.我猜这个密钥是在这个线程的.Contains检查之后但是在添加之前由并行进程添加的.除了将该行放在try-catch中之外,还有另一个我可以用来防止该错误的简单解决方案吗？

Parallel.ForEach(branchFixes, b =>
{
  Parallel.ForEach(b.Value, t =>
  {
    var team = t.Key;
    if (!resultTeamDict.ContainsKey(team))
    {
      resultTeamDict.Add(team, new Dictionary<FixItem, Dictionary<BranchInfo, bool>>());
    }
  });
});

所以情况就是这样:我需要拨打一个开始搜索的网站.这种搜索会持续一段时间,我知道搜索完成的唯一方法是定期查询网站,查看网站上是否有"下载数据"链接(它在javascript上使用了一些奇怪的ajax调用)用于检查后端和更新页面的计时器,我认为).

所以这就是诀窍:我需要搜索数百个项目,一次一个.所以我有一些看起来有点像这样的代码:

var items = getItems();
Parallel.ForEach(items, item =>
{
   startSearch(item);
   var finished = isSearchFinished(item);
   while(finished == false)
   {
      finished = isSearchFinished(item); //<--- How do I delay this action 30 Secs?
   }
   downloadData(item);
}

现在,显然这不是真正的代码,因为可能存在导致isSearchFinished永远存在的事物false.

除了显而易见的无限循环危险之外,我将如何正确地isSearchFinished()避免一遍又一遍地调用,而是每次拨打30秒或1分钟？

我知道Thread.Sleep()这不是正确的解决方案,我认为解决方案可能是通过使用来完成的,Threading.Timer()但我对它并不是很熟悉,并且有太多的线程选项,我只是不确定使用哪个.

我有96个txt文件需要处理.现在我正在使用for循环并一次执行一个,这个过程非常慢.生成的96个文件,不需要合并.有没有办法使它们并行运行,ala Parallel.foreach在C#中？当前代码:

for src_name in glob.glob(source_dir+'/*.txt'):
   outfile = open (...)
   with open(...) as infile:
      for line in infile:
         --PROCESS--
   for --condition--:
      outfile.write(...)
   infile.close()
   outfile.close()

希望此进程并行运行source_dir中的所有文件.

我有一个任务,它逐行读取一个大文件,用它做一些逻辑,并返回一个我需要写入文件的字符串.输出的顺序无关紧要.但是,当我尝试下面的代码时,它会在读取我的文件的15-20k行后停止/变得非常慢.

public static Object FileLock = new Object();
...
Parallel.ForEach(System.IO.File.ReadLines(inputFile), (line, _, lineNumber) =>
{
    var output = MyComplexMethodReturnsAString(line);
    lock (FileLock)
    {
        using (var file = System.IO.File.AppendText(outputFile))
        {
            file.WriteLine(output);
        }
    }
});

为什么我的程序在运行一段时间后会变慢？有没有更正确的方法来执行此任务？

因此,我试图解决这个问题,这是我晚上最好的部分.

我很幸运昨天被介绍到了parallel.foreach,它的工作方式就像我想要的那样,除了一个细节.

我有以下内容:

        Parallel.ForEach(data, (d) =>
        {
            try
            {
                MyMethod(d, measurements);
            }
            catch (Exception e)
            {
              // logg
            }

        });

在方法"MyMethod"中我有很多逻辑可以完成并且大部分都很好但我在api调用我获取数据的地方我使用异步任务为了能够使用"await"以便代码等到直到特定部分被执行然后继续:

    private async void MyMethod(PimData pimData, IEnumerable<ProductMeasurements> measurements)
    {
        try
        {
           // alot of logic but most relevant part 

            await Task.WhenAll(ExecuteMeasurmentAndChartLogic(pimData.ProductNumber, entity));
            await Task.WhenAll(resourceImportManager.HandleEntityImageFiles(pimData.ProductType + pimData.ProductSize,SwepImageType.Png, ResourceFileTypes.ThreeD, entity, LinkTypeId.ProductResource));

            await Task.WhenAll(resourceImportManager.HandleEntityImageFiles(pimData.ProductSketch, SwepImageType.Png, ResourceFileTypes.Sketch, entity, LinkTypeId.ProductResource));

        }
        catch (Exception e)
        {
            // logg
        }
    }

问题:

1对于初学者,循环在所有代码完成之前完成

2第二个问题是我在许多api调用中得到"任务被取消"

3如上所述,代码不会等待每个方法完全执行.

在继续下一步之前,我无法让它在ExecuteMeasurmentAndChartLogic()方法中执行所有操作.

这给了我以下问题(更多问题):

在这个方法中,我创建了一个项目并将其添加到数据库中,这个项目需要我从在ExecuteMeasurmentAndChartLogic()内部完成的api调用获得的更多信息,但问题是几个项目变得疯狂并且必须等待其余数据不是我想要的.

侧面注意:我知道装箱物品并在所有数据出现之前添加到数据库中并不是最佳实践,但我正在整合PIM并且过程很简单

我想要运行几个线程但同时我希望在继续下一个方法之前为每个项执行fuill逻辑.

澄清:

几个项目正在运行

在继续下一部分代码之前,每个项目都需要handel 所需的所有逻辑,并且等待着这样做. …

我对async/await如何作为并行工作感到困惑所以我在这里制作了一个测试代码:我尝试发送6个用模拟模拟的任务.每个任务都将执行3个其他子任务.

你可以复制/粘贴测试

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Threading;

namespace ConsoleApplication1
{
    class Program
    {
         static void Main(string[] args)
        {
            //job simulation 
            Func<int, string, Tuple<int, string>> tc = Tuple.Create;
            var input = new List<Tuple<int, string>>{
                  tc( 6000, "task 1" ),
                  tc( 5000, "task 2" ),
                  tc( 1000, "task 3" ),
                  tc( 1000, "task 4" ),
                  tc( 1000, "task 5" ),
                  tc( 1000, "task 6" )
            };

            List<Tuple<int, string>> JobsList = new List<Tuple<int, string>>(input);

            //paralelism atempt …