Параллельный цикл foreach с асинхронным кодом не завершается правильно по неизвестной причине - PullRequest
Новая
       69

Параллельный цикл foreach с асинхронным кодом не завершается правильно по неизвестной причине

0 голосов
/ 06 ноября 2019

Я пытаюсь переписать цикл foreach, чтобы использовать Parallel.ForEach, поскольку каждый документ, который мне нужно обработать, может быть обработан как отдельный объект, поэтому нет никаких зависимостей.

Код довольнопрямо как показано ниже:

  • Запрос к базе данных
  • Считывание каждого документа в цикле
  • Для каждого документа выполните два веб-вызова и добавьте результаты в документ
  • Добавить обновленный документ в список
  • Список BulkImport в БД

Поскольку вызовы веб-API являются самой медленной частью из-за задержки в сети, я хотел обрабатывать их параллельно ссэкономьте время, чтобы я написал этот код 

private async Task<List<String>> FetchDocumentsAndBuildList(string brand)
{
    using (var client = new DocumentClient(new Uri(cosmosDBEndpointUrl), cosmosDBPrimaryKey))
    {
        List<string> formattedList = new List<string>();
        FeedOptions queryOptions = new FeedOptions
        {
            MaxItemCount = -1,
            PartitionKey = new PartitionKey(brand)
        };

        var query = client.CreateDocumentQuery<Document>(UriFactory.CreateDocumentCollectionUri(cosmosDBName, cosmosDBCollectionNameRawData), $"SELECT TOP 2 * from c where c.brand = '{brand}'", queryOptions).AsDocumentQuery();

        while(query.HasMoreResults)
        {
            var options = new ParallelOptions { MaxDegreeOfParallelism = Environment.ProcessorCount * 10 };

            Parallel.ForEach(await query.ExecuteNextAsync<Document>(), options, async singleDocument =>
            {
                JObject originalData = singleDocument.GetPropertyValue<JObject>("BasicData");

                if (originalData != null)
                {
                    var artNo = originalData.GetValue("artno");
                    if (artNo != null)
                    {
                        string strArtNo = artNo.ToString();
                        string productNumber = strArtNo.Substring(0, 7);
                        string colorNumber = strArtNo.Substring(7, 3);
                        string HmGoeUrl = $"https://xxx,xom/Online/{strArtNo}/en";
                        string sisApiUrl = $"https:/yyy.com/{productNumber}/{colorNumber}?&maxnumberofstores=10&brand=000&channel=02";
                        string HttpFetchMethod = "GET";

                        JObject detailedDataResponse = await DataFetcherAsync(HmGoeUrl, HttpFetchMethod);
                        JObject inventoryData = await DataFetcherAsync(sisApiUrl, HttpFetchMethod);

                        if (detailedDataResponse != null)
                        {
                            JObject productList = (JObject)detailedDataResponse["product"];

                            if (productList != null)
                            {
                                var selectedIndex = productList["articlesList"].Select((x, index) => new { code = x.Value<string>("code"), Node = x, Index = index })
                                .Single(x => x.code == strArtNo)
                                .Index;

                                detailedDataResponse = (JObject)productList["articlesList"][selectedIndex];
                            }
                        }

                        singleDocument.SetPropertyValue("DetailedData", detailedDataResponse);
                        singleDocument.SetPropertyValue("InventoryData", inventoryData);
                        singleDocument.SetPropertyValue("consumer", "NWS");
                    }
                }
                formattedList.Add(Newtonsoft.Json.JsonConvert.SerializeObject(singleDocument));
            });


            //foreach (Document singleDocument in await query.ExecuteNextAsync<Document>())
            //{
            //    JObject originalData = singleDocument.GetPropertyValue<JObject>("BasicData");

            //    if(originalData != null)
            //    {
            //        var artNo = originalData.GetValue("artno");
            //        if(artNo != null)
            //        {
            //            string strArtNo = artNo.ToString();
            //            string productNumber = strArtNo.Substring(0, 7);
            //            string colorNumber = strArtNo.Substring(7, 3);
            //            string HmGoeUrl = $"https:/xxx.xom/Online/{strArtNo}/en";
            //            string sisApiUrl = $"https://yyy.xom&maxnumberofstores=10&brand=000&channel=02";
            //            string HttpFetchMethod = "GET";

            //            JObject detailedDataResponse = await DataFetcherAsync(HmGoeUrl, HttpFetchMethod);
            //            JObject inventoryData = await DataFetcherAsync(sisApiUrl, HttpFetchMethod);

            //            if(detailedDataResponse != null)
            //            {
            //                JObject productList = (JObject)detailedDataResponse["product"];

            //                if(productList != null)
            //                {
            //                    var selectedIndex = productList["articlesList"].Select((x, index) => new { code = x.Value<string>("code"), Node = x, Index = index })
            //                    .Single(x => x.code == strArtNo)
            //                    .Index;

            //                    detailedDataResponse = (JObject)productList["articlesList"][selectedIndex];
            //                }
            //            }

            //            singleDocument.SetPropertyValue("DetailedData", detailedDataResponse);
            //            singleDocument.SetPropertyValue("InventoryData", inventoryData);
            //            singleDocument.SetPropertyValue("consumer", "NWS");
            //        }
            //    }
            //    formattedList.Add(Newtonsoft.Json.JsonConvert.SerializeObject(singleDocument));
            //}
        }
        return formattedList;
    }
}

Если я добавлю точку останова в цикле, я вижу, что для каждой переменной назначены правильные значения, но по какой-то причине возвращаемое значение formattedList всегда равно 0 иЯ не могу понять, почему.

Закомментирован оригинальный цикл foreach, который прекрасно работает, но имеет значение slooooow

--- EDIT --- Вот как я вызываю этот код изродительский метод 

   log.LogInformation($"Starting creation of DocumentList for BulkImport at: {DateTime.Now}");

   var documentsToImportInBatch = await FetchDocumentsAndBuildList(brand);

   log.LogInformation($"BulkExecutor DocumentList has: {documentsToImportInBatch.Count} entries, created at: {DateTime.Now}");

1 Ответ

1 голос
/ 08 ноября 2019

Проблема здесь в том, что Parallel.ForEach не понимает, что каждый вызов вашей лямбды, возвращающий Task, следует ожидать, прежде чем ForEach можно будет считать завершенным.

В результате,продолжение после await не вызывается до выхода из вашей функции, и именно поэтому formattedList содержит ноль элементов.

Вы можете легко доказать это с помощью примера кода, такого как: 

Parallel.ForEach(Enumerable.Range(0, 100), async singleDocument => await Task.Delay(9999));
Console.WriteLine("Done!");

Done будет напечатан почти сразу.

Для параллелизма, связанного с вводом / выводом, вы могли бы вместо этого использовать Task.WhenAll для распараллеливания асинхронных вызовов на основе затирания 

var myDocuments = await query.ExecuteNextAsync<Document>();
var myScrapingTasks = myDocuments.Select(async singleDocument =>
{
       // ... all of your web scraping code here
       // return the amended (mutated) document
       return JsonConvert.SerializeObject(singleDocument);
});
var results = await Task.WhenAll(myScrapingTasks);
formattedList.AddRange(results);

wrt MaxDegreeOfParallelism, если вы обнаружите, что вам нужно ограничить количество одновременных вызовов вызовов, проще всего будет сгруппировать входящие документы в управляемые порции и обрабатывать меньшие порции за раз - перегрузка Select(x, i) иGroupBy чудеса творчества.

...