标签: parallel.foreach

我有代码：

public void FindMatches(string source)
{
   ...
   var matchCollections = new List<MatchCollection>();
   Parallel.ForEach(patterns,
      pattern =>
         {
            var regex = new Regex(pattern);
            MatchCollection matches = regex.Matches(source, 0);
            matchCollections.Add(matches);
         }
      );   
   foreach (MatchCollection matches in matchCollections)
   {
      if (matches.Count > 0) //NullReferenceException
      {
         foreach (Match match in matches)
         {
            ...
         }
      }
   } 
   ...
}

有时我在第 15 行遇到 NullreferenceException。如果在插入“matchCollections”之前检查“matches”不为空，则仍然会抛出异常。什么问题？

这是一个失败的测试.如何确认循环运行的次数是否正确？

    public Random Randomator { get; set; }
    public const int TimesToRun = 1000000;

    [TestMethod]
    public void ThrowTheDice()
    {
        Randomator = new Random();

        var resultsParallel = new Dictionary<int, int>
        {
            {1, 0}, {2, 0}, {3, 0}, {4, 0}, {5, 0}, {6, 0}
        };

        var resultsParallelForEach = new Dictionary<int, int>
        {
            {1, 0}, {2, 0}, {3, 0}, {4, 0}, {5, 0}, {6, 0}
        };

        var stopwatch = new Stopwatch();
        stopwatch.Start();
        Parallel.For(0, TimesToRun, ctr =>
        {
            var val = ThrowDice();
            if (!resultsParallel.ContainsKey(val)) …

我有一个方法,它读取一个文本文件,每行包含一个int值,为了使读取速度更快,我使用Parallel.ForEach,但我所看到的行为是意外的,我在文件中有800行但是当我运行这个方法时,每个时间它返回不同的HashSet计数,我在搜索后读取的是Parallel.ForEach生成多个线程并且当所有线程完成其工作时返回结果,但是我的代码执行矛盾,或者我在这里遗漏了一些重要的东西？

这是我的方法:

private HashSet<int> GetKeyItemsProcessed()
{
   HashSet<int> keyItems = new HashSet<int>();

   if (!File.Exists(TrackingFilePath))
     return keyItems;

     // normal foreach works fine

     //foreach(var keyItem in File.ReadAllLines(TrackingFilePath))
     //{
     //    keyItems.Add(int.Parse(keyItem));
     //}


     // this does not return right number of hashset rows
     Parallel.ForEach(File.ReadAllLines(TrackingFilePath).AsParallel(), keyItem =>
     {
         keyItems.Add(int.Parse(keyItem));
     });


    return keyItems;

}

我必须在批处理作业中处理大量数据

程序流程

var inputDataLst = contextObj.GetData(); //More that 10000 rows I will retrieve

foreach(var item in inpDataLst)
{
     //logic
}

Call context.SaveMethod(inpuDataLst);

我想做点什么

var tsklst = inputDataLst.Select(async pItem => 
{
   //Logic
});

await Task.WhenAll(taskList);

编译器发出警告,Resharper建议我制作同步方法.

任何人都可以建议我如何处理这个,因为我将有大量的数据,我想做一个异步操作..

实施@bruno建议

Parallel.ForEach(taskList, item => {
 item.StatusId = 2; //Completed
 LastUpdateUser = "Batch";
});

循环完成后我得到taskList为NULL,

我有一些过程可以按顺序迭代一组大文件并从中提取信息来写入\更新我们的数据库。这些文件通常每个文件都有几千行，因此我构建了一个 Parallel.ForEach 来同时处理给定文件中的多行（由于需要按顺序应用每个文件，因此一次只处理一个文件）。现在我需要了解当前文件的大约处理量，以便我可以向管理层提供剩余运行时间的指示。到目前为止我有以下内容

public void MyProcess(FileItem file)
{
    List<string> lines = file.GetLines(); //some process to get the lines to handle
    long cntr = 0; //The counter to track

    Parallel.ForEach(lines, crntLine =>
    {
       Console.Writeline(String.Format("Currently finished {0} out of {1} lines",cntr,lines.Count());
       InterLocked.Increment(ref cntr);

       //...Code to process crntLine here

    });
}

我根本不关心已处理哪些行，只关心总共处理了多少行，这样就可以回答它在当前文件中的位置有多远的问题。这会给我带来我正在寻找的可靠的东西吗？

我已经编写了3种不同的计算整数数组之和的方法,但是,我得到了第三种方法的不同结果.

初始化:

        int n = 100;
        int[] mArray = new int[n];
        for (int i = 0; i < mArray.Length; i++)          
            mArray[i] = 1;

第一:

        int sum1 = mArray.Sum();
        Console.WriteLine("sum1 " + sum1);

第二:

        int sum2 = 0;
        for (int i = 0; i < mArray.Length; i++)
            sum2 += mArray[i];
        Console.WriteLine("sum2 " + sum2);

第三:

        int sum3 = 0;
        Parallel.ForEach(mArray, item =>
        {
            sum3 += item;
        });
        Console.WriteLine("sum3 " + sum3);

显然,3种方法给出了如下所示的相同输出:

然而,当n增加(例如,n = 30000)时,第三种方法给出了令人惊讶的错误结果

注意:我使用ConcurrentBag测试了这些方法,这是一个线程安全的集合.我想,没有溢出问题.代码在Windows 10 x64计算机(Intel核心I-7 @ 3.30ghz)上进行测试

理解为什么Parallel.For的行为不同会很有趣.

我有一种方法可以并行加载并运行报表布局。所有报告将使用相同的baselayout.xml。由于线程每次尝试访问同一资源时都会因异常而失败，因此我使用了a lock来锁定文件。

public static XmlTextReader LoadReport(string reportName)
{
    object _locker = new object();
    object reportData;
    lock (_locker)
    {
        reportData = Resources.ResourceManager.GetObject(reportName);
    }
    return new XmlTextReader(new MemoryStream((byte[])reportData));
}

并行方法如下所示：

private void RunReportsParallel(List<ReportObject> coverterList)
{
    try
    {
        Parallel.ForEach(coverterList, (currentObject) => {
            currentObject.Convert();
        });    
    }
    catch (Exception e)
    {
        smlLogger.Error(Helper.SetLogLine(e.Message, processId));
        throw;
    }
}

Conver将运行以下代码：

public override SectionReport GetMainReport()
{
    SectionReport mainReport = new SectionReport();
    XMLDataSource datasource = new XMLDataSource(null, "//AkontoRechnung");
    datasource.LoadXML(rechnungsdaten.ToString());
    mainReport = new ReportAkontorechnung(datasource, reportConfiguration, Language, NoPrintOut);
    try …

我在列表中有几个对象。每分钟，对于他们每个人，我都必须下载api信息并将接收到的数据保存到数据库中。

列表中的对象彼此独立。每个对象的方法可以单独调用。根据 api 的响应时间，单个方法的总执行时间通常为 ~0.5 到 5 秒。

按照 stackoverflow 上其他线程中的建议，我创建了以下代码：

private async void DownloadAndSaveDataForMyObjects(object state)
{
   try
   {
      await Task.Run(() => Parallel.ForEach(MyObjectsList, ServiceSingleObjectMethod));
   }
}


private void ServiceSingleObjectMethod(RehDeviceSmsRequest myObjectFromList)
{
   var apiInfo = GetInfoFromApi(myObjectFromList);
   SaveInfoToDatabase(myObjectFromList);
}

在我看来，代码是异步的。但是……我有些怀疑。Logger 通知我应用程序运行期间执行的操作。对象的方法调用之间的间隔非常大。请看这个：

[13:32:46 DBG][MyService]->MyMethod => Update object id: 54
[13:32:47 DBG][MyService]->MyMethod => Update object id: 9
[13:32:50 DBG][MyService]->MyMethod => Update object id: 47
[13:32:51 DBG][MyService]->MyMethod => Update object id: 21
[13:32:51 DBG][MyService]->MyMethod => Update object id: 53
[13:32:53 DBG][MyService]->MyMethod => Update object id: …

我有一个长期运行的工作，我需要为集合中的每个项目运行一次。

我想同时做这些工作，虽然整个程序可以等待所有工作完成。（稍后我可能会更改它，但就目前而言，我让问题保持简单）

根据已经提供的一些帮助，我得到了以下模式：

public static void DoWork()
{
    //Get a collection of items
    var items = GetMyItems();

}


private async void DoStuffWithItems(ICollection<MyItem> items)
{
    var tasks = items.Select (i => DoStuffWithItem(i));
    await Task.WhenAll(tasks);
}

private Task DoStuffWithItem(MyItem item)
{
    //LongRunningTask
    return Task.Run(async () =>
            {
                var returnObject = await LongRunningAsyncMethod(item);
            });
}

如果我理解正确的话，这仍然一次完成每项任务 - 毫无意义。

有人建议我将 Parallel.ForEach 与 async await 结合使用 - Parallel-ForEach 模式很简单：

public static void DoWork()
{
    //Get a collection of items
    var items = GetMyItems();
    
    Parallel.ForEach(items, (item) …

我有一个需要并行运行的循环，因为每次迭代都很慢并且需要大量处理器，但我还需要调用异步方法作为循环中每次迭代的一部分。

我见过关于如何在循环中处理异步方法的问题，但没有看到关于异步和同步的组合的问题，这就是我所得到的。

我的（简化的）代码如下 - 我知道由于异步操作被传递给 foreach，这将无法正常工作。

protected IDictionary<int, ReportData> GetReportData()
{
    var results = new ConcurrentDictionary<int, ReportData>();
      
    Parallel.ForEach(requestData, async data =>
    {
        // process data synchronously
        var processedData = ProcessData(data);

        // get some data async
        var reportRequest = await BuildRequestAsync(processedData);

        // synchronous building
        var report = reportRequest.BuildReport();

        results.TryAdd(data.ReportId, report);
     });

     // This needs to be populated before returning
     return results;
}

当操作必须异步才能等待单个异步调用时，有什么方法可以并行执行操作。

将同步函数转换为异步函数并不是一个实用的选择。

我不想将操作拆分并使用 Parallel.ForEach ，然后使用 WhenAll 和另一个 Parallel.ForEach 进行异步调用，因为每个阶段的速度在不同迭代之间可能差异很大，因此拆分操作效率较低，因为速度越快那些人会等待较慢的人然后再继续。

我确实想知道是否可以使用 PLINQ ForAll 来代替 Parallel.ForEach，但从未使用过 PLINQ，并且不确定它是否会在返回之前等待所有迭代完成，即任务是否仍在运行结束的过程。

我需要在DataTable中的URI上进行多个Web请求。之前我有以下代码。但我意识到，这使得同步调用为的await会等到GET / POST调用完成，并已处理的应答，然后前进到下一个迭代。

foreach (DataRow dr in dt.Rows)
{
    activeTasks.Add(SendRequestAsync(dr));
    Task.WhenAll(activeTasks).Wait();
}

private async Task<string> SendRequestAsync(DataRow dr)
{
    using (var client = new HttpClient())
    {
        string reqMethod = (dr["RequestMethod"] != null && dr["RequestMethod"].ToString() != "") ? dr["RequestMethod"].ToString() : "GET";
        client.BaseAddress = new Uri(dr["URL"].ToString());
        client.DefaultRequestHeaders.Accept.Clear();
        string reqContentType = (dr["RequestContentType"] != null && dr["RequestContentType"].ToString() != "") ? dr["RequestContentType"].ToString() : "text/xml";
        client.DefaultRequestHeaders.Accept.Add(new MediaTypeWithQualityHeaderValue(reqContentType));

        HttpResponseMessage response = null;
        try
        {
            if (reqMethod == "GET")
                response = await client.GetAsync(client.BaseAddress.AbsoluteUri);
            else
                response = await …

在Directory.EnumerateFiles或Directory.GetFiles中允许使用.take()？

我有这个并且有效

    Parallel.ForEach(Directory.EnumerateFiles(sourceFolder, "*", 
               SearchOption.AllDirectories), file =>
    {
         //DO SOMETHING

    });

我只想在例如100个文件中进行循环.所以:

    Parallel.ForEach(Directory.EnumerateFiles(sourceFolder, "*",
              SearchOption.AllDirectories).Take(100), file =>
    {
         //DO SOMETHING

    });

但是这个周期并没有像我预期的那样被记录100次.

为什么？