С#.net 4.5 async/multithread?

Ответ 1

Если вы хотите использовать async и await (хотя вам это не обязательно, но в .NET 4.5 все упрощается), вы сначала захотите изменить свой метод ScrapeData, чтобы вернуть Task<T> instance, используя ключевое слово async, например:

async Task<DataSet> ScrapeDataAsync(Uri url)
{
    // Create the HttpClientHandler which will handle cookies.
    var handler = new HttpClientHandler();

    // Set cookies on handler.

    // Await on an async call to fetch here, convert to a data
    // set and return.
    var client = new HttpClient(handler);

    // Wait for the HttpResponseMessage.
    HttpResponseMessage response = await client.GetAsync(url);

    // Get the content, await on the string content.
    string content = await response.Content.ReadAsStringAsync();

    // Process content variable here into a data set and return.
    DataSet ds = ...;

    // Return the DataSet, it will return Task<DataSet>.
    return ds;
}

Обратите внимание, что вы, вероятно, захотите отойти от класса WebClient, поскольку он не поддерживает Task<T> по своей сути в своих асинхронных операциях. Лучшим выбором в .NET 4.5 является класс HttpClient. Я решил использовать HttpClient выше. Кроме того, посмотрите HttpClientHandler класс, в частности CookieContainer свойство, которое вы будете использовать для отправки файлов cookie с каждым запросом.

Однако это означает, что вам, скорее всего, придется использовать ключевое слово await для ожидания другой операции async, которая в этом случае скорее всего будет загружать страницу. Вам придется настраивать свои вызовы, загружающие данные, для использования асинхронных версий и await на них.

Как только это будет завершено, вы обычно вызываете await, но вы не можете сделать этого в этом сценарии, потому что вы бы await для переменной. В этом случае вы запускаете цикл, поэтому переменная будет reset с каждой итерацией. В этом случае лучше просто сохранить Task<T> в массиве следующим образом:

DataSet alldata = ...;

var tasks = new List<Task<DataSet>>();

foreach(var url in the8000urls)
{
    // ScrapeData downloads the html from the url with 
    // WebClient.DownloadString
    // and scrapes the data into several datatables which 
    // it returns as a dataset.
    tasks.Add(ScrapeDataAsync(url));
}

Существует вопрос о объединении данных в allData. С этой целью вы хотите вызвать метод ContinueWith в возвращаемом экземпляре Task<T> и выполнить задачу добавления данных в allData:

DataSet alldata = ...;

var tasks = new List<Task<DataSet>>();

foreach(var url in the8000urls)
{
    // ScrapeData downloads the html from the url with 
    // WebClient.DownloadString
    // and scrapes the data into several datatables which 
    // it returns as a dataset.
    tasks.Add(ScrapeDataAsync(url).ContinueWith(t => {
        // Lock access to the data set, since this is
        // async now.
        lock (allData)
        {
             // Add the data.
        }
    });
}

Затем вы можете подождать по всем задачам, используя метод WhenAll на Task class и await на этом:

// After your loop.
await Task.WhenAll(tasks);

// Process allData

Однако обратите внимание, что у вас есть foreach, а WhenAll выполняет IEnumerable<T>. Это хороший показатель того, что это подходит для использования LINQ, что это:

DataSet alldata;

var tasks = 
    from url in the8000Urls
    select ScrapeDataAsync(url).ContinueWith(t => {
        // Lock access to the data set, since this is
        // async now.
        lock (allData)
        {
             // Add the data.
        }
    });

await Task.WhenAll(tasks);

// Process allData

Вы также можете не использовать синтаксис запроса, если хотите, в этом случае это не имеет значения.

Обратите внимание, что если содержащийся метод не помечен как async (поскольку вы находитесь в консольном приложении и должны ждать результатов до завершения приложения), вы можете просто вызвать Wait method в Task, возвращаемом при вызове WhenAll:

// This will block, waiting for all tasks to complete, all
// tasks will run asynchronously and when all are done, then the
// code will continue to execute.
Task.WhenAll(tasks).Wait();

// Process allData.

А именно, вы хотите собрать экземпляры Task в последовательность, а затем дождаться всей последовательности перед обработкой allData.

Однако я бы предложил попытаться обработать данные, прежде чем слить их в allData, если можно; если для обработки данных не требуется всего DataSet, вы получите еще больший выигрыш в производительности, обработав как можно больше данных, когда вы вернете их, а не ожидаете, пока все вернется.

Ответ 2

Вы также можете использовать TPL Dataflow, который подходит для этой проблемы.

В этом случае вы создаете "сетку потока данных", а затем ваши данные проходят через нее.

Это больше похоже на конвейер, чем на "сетку". Я делаю три шага: Загрузите (строковые) данные из URL-адреса; Разбирайте (string) данные в HTML и затем в DataSet; и слейте DataSet в мастер DataSet.

Сначала мы создаем блоки, которые будут перемещаться в сетке:

DataSet allData;
var downloadData = new TransformBlock<string, string>(
  async pageid =>
  {
    System.Net.WebClient webClient = null;
    var url = "https://domain.com?&id=" + pageid + "restofurl";
    return await webClient.DownloadStringTaskAsync(url);
  },
  new ExecutionDataflowBlockOptions
  {
    MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded,
  });
var parseHtml = new TransformBlock<string, DataSet>(
  html =>
  {
    var dsPageData = new DataSet();
    var doc = new HtmlDocument();
    doc.LoadHtml(html);

    // HTML Agility parsing

    return dsPageData;
  },
  new ExecutionDataflowBlockOptions
  {
    MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded,
  });
var merge = new ActionBlock<DataSet>(
  dataForOnePage =>
  {
    // merge dataForOnePage into allData
  });

Затем мы связываем три блока вместе, чтобы создать сетку:

downloadData.LinkTo(parseHtml);
parseHtml.LinkTo(merge);

Затем мы начинаем накачивать данные в сетку:

foreach (var pageid in the8000urls)
  downloadData.Post(pageid);

И, наконец, мы ждем завершения каждого шага в сетке (это также будет чисто распространять любые ошибки):

downloadData.Complete();
await downloadData.Completion;
parseHtml.Complete();
await parseHtml.Completion;
merge.Complete();
await merge.Completion;

Хорошая вещь о потоке данных TPL заключается в том, что вы можете легко контролировать, насколько параллельны каждая часть. На данный момент я установил для блоков загрузки и разбора Unbounded, но вы можете ограничить их. Блок слияния использует максимальный по умолчанию parallelism по умолчанию 1, поэтому при слиянии не требуется блокировок.

Ответ 3

Я рекомендую прочитать мое разумное полное введение в async/await.

Сначала сделайте все асинхронным, начиная с материала нижнего уровня:

public static async Task<DataSet> ScrapeDataAsync(string pageid)
{
  CookieAwareWebClient webClient = ...;
  var dsPageData = new DataSet();

  // DOWNLOAD HTML FOR THE REO PAGE AND LOAD IT INTO AN HTMLDOCUMENT
  string url = @"https://domain.com?&id=" + pageid + @"restofurl";
  string html = await webClient.DownloadStringTaskAsync(url).ConfigureAwait(false);
  var doc = new HtmlDocument();
  doc.LoadHtml(html);

  // A BUNCH OF PARSING WITH HTMLAGILITY AND STORING IN dsPageData 
  return dsPageData;
}

Затем вы можете использовать его следующим образом (используя async с LINQ):

DataSet alldata;
var tasks = the8000urls.Select(async url =>
{
  var dataForOnePage = await ScrapeDataAsync(url);

  //merge each table in dataForOnePage into allData

});
await Task.WhenAll(tasks);
PushAllDataToSql(alldata);

И используйте AsyncContext из моей библиотеки AsyncEx, так как это a консольное приложение:

class Program
{
  static int Main(string[] args)
  {
    try
    {
      return AsyncContext.Run(() => MainAsync(args));
    }
    catch (Exception ex)
    {
      Console.Error.WriteLine(ex);
      return -1;
    }
  }

  static async Task<int> MainAsync(string[] args)
  {
    ...
  }
}

Что это. Нет необходимости в блокировке или продолжении или любом из них.

Ответ 4

Я считаю, что вам здесь не нужны async и await. Они могут помочь в настольном приложении, где вам нужно переместить свою работу в поток без GUI. На мой взгляд, в вашем случае будет лучше использовать метод Parallel.ForEach. Что-то вроде этого:

    DataSet alldata;
    var bag = new ConcurrentBag<DataSet>();

    Parallel.ForEach(the8000urls, url =>
    {
        // ScrapeData downloads the html from the url with WebClient.DownloadString 
        // and scrapes the data into several datatables which it returns as a dataset. 
        DataSet dataForOnePage = ScrapeData(url);
        // Add data for one page to temp bag
        bag.Add(dataForOnePage);
    });

    //merge each table in dataForOnePage into allData from bag

    PushAllDataToSql(alldata);