Блокировка Task.Result внутри Parallel.ForEach из-за медленного запроса HttpClient - PullRequest
Новая
       15

Блокировка Task.Result внутри Parallel.ForEach из-за медленного запроса HttpClient

2 голосов
/ 03 июля 2019

Я понимаю последствия использования асинхронной лямбды с Parallel.ForEach, поэтому я не использую ее здесь.Затем это заставляет меня использовать .Result для каждой из моих задач, которые делают запросы Http.Тем не менее, выполнение этого простого скребка через профилировщик производительности показывает, что .Result имеет истекшее исключительное время% ~ 98%, что, очевидно, связано с блокирующим характером вызова.

Мой вопрос: есть лиВозможность оптимизировать это для того, чтобы он все еще был асинхронным?Я не уверен, что это поможет в этом случае, так как для извлечения HTML / XML может потребоваться много времени.

Я использую 4-ядерный процессор с 8 логическими ядрами (отсюда MaxDegreesOfParallelism = 8.Сейчас я смотрю около 2,5 часов, чтобы загрузить и проанализировать ~ 51 000 HTML / XML-страниц простых финансовых данных.

Я склонялся к использованию XmlReader вместо Linq2XML для ускорения анализа, но, похоже,Узкое место при вызове .Result.

И хотя здесь это не должно иметь значения, SEC ограничивает очистку до 10 запросов / сек. 

public class SECScraper
{
    public event EventHandler<ProgressChangedEventArgs> ProgressChangedEvent;

    public SECScraper(HttpClient client, FinanceContext financeContext)
    {
        _client = client;
        _financeContext = financeContext;
    }

    public void Download()
    {
        _numDownloaded = 0;
        _interval = _financeContext.Companies.Count() / 100;

        Parallel.ForEach(_financeContext.Companies, new ParallelOptions {MaxDegreeOfParallelism = 8},
            company =>
            {
                RetrieveSECData(company.CIK);
            });
    }

    protected virtual void OnProgressChanged(ProgressChangedEventArgs e)
    {
        ProgressChangedEvent?.Invoke(this, e);
    }

    private void RetrieveSECData(int cik)
    {
        // move this url elsewhere
        var url = "https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK=" + cik +
                  "&type=10-q&dateb=&owner=include&count=100";

        var srBody = ReadHTML(url).Result; // consider moving this to srPage
        var srPage = new SearchResultsPage(srBody);

        var reportLinks = srPage.GetAllReportLinks();

        foreach (var link in reportLinks)
        {
            url = SEC_HOSTNAME + link;

            var fdBody = ReadHTML(url).Result;
            var fdPage = new FilingDetailsPage(fdBody);

            var xbrlLink = fdPage.GetInstanceDocumentLink();

            var xbrlBody = ReadHTML(SEC_HOSTNAME + xbrlLink).Result;
            var xbrlDoc = new XBRLDocument(xbrlBody);
            var epsData = xbrlDoc.GetAllEPSData();

            //foreach (var eps in epsData)
            //    Console.WriteLine($"{eps.StartDate} to {eps.EndDate} -- {eps.EPS}");
        }

        IncrementNumDownloadedAndNotify();
    }

    private async Task<string> ReadHTML(string url)
    {
        using var response = await _client.GetAsync(url);
        return await response.Content.ReadAsStringAsync();
    }
}

Ответы [ 2 ]

3 голосов
/ 03 июля 2019

Задача связана не с процессором, а с сетью, поэтому нет необходимости использовать несколько потоков.

Выполните несколько асинхронных вызовов в одном потоке. просто не ждите их. Поместите задачи в список. Когда вы наберете определенную сумму (скажем, хотите 10) за один раз, начните ждать окончания первого (для получения дополнительной информации посмотрите «задача, когда»).

Затем добавьте больше :-) Затем вы можете контролировать размер задач с помощью # / секунды, используя другой код.

1 голос
/ 03 июля 2019

есть ли возможность оптимизировать его, чтобы он все еще был асинхронным?

Да.Я не уверен, почему вы используете Parallel в первую очередь;похоже, это неправильное решение для такого рода проблемУ вас есть асинхронная работа с набором элементов, поэтому лучше всего подойдет асинхронный параллелизм;это делается с помощью Task.WhenAll: 

public class SECScraper
{
  public async Task DownloadAsync()
  {
    _numDownloaded = 0;
    _interval = _financeContext.Companies.Count() / 100;

    var tasks = _financeContext.Companies.Select(company => RetrieveSECDataAsync(company.CIK)).ToList();
    await Task.WhenAll(tasks);
  }

  private async Task RetrieveSECDataAsync(int cik)
  {
    var url = "https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany&CIK=" + cik +
        "&type=10-q&dateb=&owner=include&count=100";

    var srBody = await ReadHTMLAsync(url);
    var srPage = new SearchResultsPage(srBody);

    var reportLinks = srPage.GetAllReportLinks();

    foreach (var link in reportLinks)
    {
      url = SEC_HOSTNAME + link;

      var fdBody = await ReadHTMLAsync(url);
      var fdPage = new FilingDetailsPage(fdBody);

      var xbrlLink = fdPage.GetInstanceDocumentLink();

      var xbrlBody = await ReadHTMLAsync(SEC_HOSTNAME + xbrlLink);
      var xbrlDoc = new XBRLDocument(xbrlBody);
      var epsData = xbrlDoc.GetAllEPSData();
    }

    IncrementNumDownloadedAndNotify();
  }

  private async Task<string> ReadHTMLAsync(string url)
  {
    using var response = await _client.GetAsync(url);
    return await response.Content.ReadAsStringAsync();
  }
}

Кроме того, я рекомендую использовать IProgress<T> для сообщения о прогрессе.

...