Как ограничить количество одновременных операций асинхронного ввода-вывода?

115
// let's say there is a list of 1000+ URLs
string[] urls = { "http://google.com", "http://yahoo.com", ... };

// now let's send HTTP requests to each of these URLs in parallel
urls.AsParallel().ForAll(async (url) => {
    var client = new HttpClient();
    var html = await client.GetStringAsync(url);
});

Проблема в том, что он запускает более 1000 одновременных веб-запросов. Есть ли простой способ ограничить одновременное количество этих асинхронных HTTP-запросов? Таким образом, в любой момент времени загружается не более 20 веб-страниц. Как сделать это наиболее эффективно?

Кодер скорби
источник
2
Чем это отличается от вашего предыдущего вопроса ?
svick
1
stackoverflow.com/questions/9290498/… С параметром ParallelOptions.
Крис Дисли
4
@ChrisDisley, это только распараллеливает запуск запросов.
Спендер
@svick прав, чем он отличается? кстати, мне нравится ответ на него stackoverflow.com/a/10802883/66372
eglasius
3
Кроме того, HttpClientесть IDisposable, и вы должны избавиться от него, особенно если вы собираетесь использовать более 1000 штук. HttpClientможет использоваться как синглтон для нескольких запросов.
Шимми Вайцхандлер

Ответы:

161

Вы определенно можете сделать это в последних версиях async для .NET, используя .NET 4.5 Beta. Предыдущий пост от 'usr' указывает на хорошую статью, написанную Стивеном Тубом, но менее анонсированная новость заключается в том, что семафор async фактически попал в бета-версию .NET 4.5.

Если вы посмотрите на наш любимый SemaphoreSlimкласс (который вам следует использовать, поскольку он более производительный, чем исходный Semaphore), теперь он может похвастаться WaitAsync(...)серией перегрузок со всеми ожидаемыми аргументами - интервалами тайм-аута, токенами отмены, всеми вашими обычными друзьями по планированию: )

Стивен также написал в своем недавнем блоге сообщение о новых возможностях .NET 4.5, которые появились в бета-версии, см. Что нового для параллелизма в .NET 4.5 Beta .

Наконец, вот несколько примеров кода о том, как использовать SemaphoreSlim для регулирования асинхронного метода:

public async Task MyOuterMethod()
{
    // let's say there is a list of 1000+ URLs
    var urls = { "http://google.com", "http://yahoo.com", ... };

    // now let's send HTTP requests to each of these URLs in parallel
    var allTasks = new List<Task>();
    var throttler = new SemaphoreSlim(initialCount: 20);
    foreach (var url in urls)
    {
        // do an async wait until we can schedule again
        await throttler.WaitAsync();

        // using Task.Run(...) to run the lambda in its own parallel
        // flow on the threadpool
        allTasks.Add(
            Task.Run(async () =>
            {
                try
                {
                    var client = new HttpClient();
                    var html = await client.GetStringAsync(url);
                }
                finally
                {
                    throttler.Release();
                }
            }));
    }

    // won't get here until all urls have been put into tasks
    await Task.WhenAll(allTasks);

    // won't get here until all tasks have completed in some way
    // (either success or exception)
}

И последнее, но, вероятно, заслуживающее упоминания - решение, использующее планирование на основе TPL. Вы можете создавать задачи с привязкой к делегатам в TPL, которые еще не были запущены, и разрешить настраиваемому планировщику задач ограничивать параллелизм. Фактически, здесь есть образец MSDN:

См. Также TaskScheduler .

Тео Яунг
источник
3
Разве parallel.foreach с ограниченной степенью параллелизма не лучший подход? msdn.microsoft.com/en-us/library/…
GreyCloud
2
Почему бы тебе не избавиться от тебяHttpClient
Шимми Вайцхандлер
4
@GreyCloud: Parallel.ForEachработает с синхронным кодом. Это позволяет вызывать асинхронный код.
Джош Ноу
2
@TheMonarch , ты ошибаешься . Кроме того , это всегда хорошая привычка , чтобы обернуть все IDisposableс в usingили try-finallyзаявлениях, а также обеспечить их утилизацию.
Шимми Вайцхандлер
29
Учитывая, насколько популярен этот ответ, стоит отметить, что HttpClient может и должен быть одним общим экземпляром, а не экземпляром для каждого запроса.
Руперт Роунсли
15

Если у вас есть IEnumerable (например, строки URL-адресов), и вы хотите выполнить операцию ввода-вывода с каждым из них (например, выполнить асинхронный HTTP-запрос) одновременно И, возможно, вы также хотите установить максимальное количество одновременных Запросы ввода-вывода в реальном времени. Вот как это можно сделать. Таким образом, вы не используете пул потоков и др., Метод использует semaphoreslim для управления максимальным количеством одновременных запросов ввода-вывода, аналогичных шаблону скользящего окна, один запрос завершает, оставляет семафор, а следующий входит.

использование: await ForEachAsync (urlStrings, YourAsyncFunc, optionalMaxDegreeOfConcurrency);

public static Task ForEachAsync<TIn>(
        IEnumerable<TIn> inputEnumerable,
        Func<TIn, Task> asyncProcessor,
        int? maxDegreeOfParallelism = null)
    {
        int maxAsyncThreadCount = maxDegreeOfParallelism ?? DefaultMaxDegreeOfParallelism;
        SemaphoreSlim throttler = new SemaphoreSlim(maxAsyncThreadCount, maxAsyncThreadCount);

        IEnumerable<Task> tasks = inputEnumerable.Select(async input =>
        {
            await throttler.WaitAsync().ConfigureAwait(false);
            try
            {
                await asyncProcessor(input).ConfigureAwait(false);
            }
            finally
            {
                throttler.Release();
            }
        });

        return Task.WhenAll(tasks);
    }
Догу Арслан
источник
нет, вам не нужно явно удалять SemaphoreSlim в этой реализации и использовании, поскольку он используется внутри метода, и метод не имеет доступа к своему свойству AvailableWaitHandle, и в этом случае нам пришлось бы либо удалить, либо обернуть его в блок using.
Dogu Arslan
1
Просто думайте о лучших практиках и уроках, которые мы преподаем другим людям. Было usingбы неплохо.
AgentFire
ну, этому примеру я могу следовать, но пытаюсь понять, как лучше всего это сделать, в основном есть дроссель, но мой Func вернет список, который я в конечном итоге хочу в окончательном списке всего завершенного, когда будет сделано ... что может требуется заблокировано в списке, у вас есть предложения.
Seabizkit
вы можете немного обновить метод, чтобы он возвращал список актуальных задач, а вы ожидали Task.WhenAll внутри вызывающего кода. После завершения Task.WhenAll вы можете перечислить каждую задачу в списке и добавить ее список в окончательный список. Измените подпись метода на 'public static IEnumerable <Task <TOut>> ForEachAsync <TIn, TOut> (IEnumerable <TIn> inputEnumerable, Func <TIn, Task <TOut>> asyncProcessor, int? MaxDegreeOfParallelism = null)'
Догу Арслан,
7

К сожалению, в .NET Framework отсутствуют наиболее важные комбинаторы для организации параллельных асинхронных задач. Встроенного такого нет.

Взгляните на класс AsyncSemaphore, созданный самым уважаемым Стивеном Тубом. То, что вам нужно, называется семафором, и вам нужна его асинхронная версия.

usr
источник
12
Обратите внимание: «К сожалению, в .NET Framework отсутствуют наиболее важные комбинаторы для оркестровки параллельных асинхронных задач. Встроенной такой вещи нет». больше не является правильным с версии .NET 4.5 Beta. SemaphoreSlim теперь предлагает функцию WaitAsync (...) :)
Тео Яунг,
Должен ли SemaphoreSlim (с его новыми асинхронными методами) быть предпочтительнее AsyncSemphore, или реализация Toub все еще имеет некоторое преимущество?
Todd Menier
На мой взгляд, следует предпочесть встроенный тип, потому что он, вероятно, будет хорошо протестирован и хорошо спроектирован.
usr
4
Стивен добавил комментарий в ответ на вопрос в своем сообщении в блоге, подтверждающий, что использование SemaphoreSlim для .NET 4.5, как правило, будет правильным решением.
jdasilva
7

Есть много подводных камней, и прямое использование семафора может быть непростым в случае ошибок, поэтому я бы предложил использовать пакет AsyncEnumerator NuGet вместо того, чтобы заново изобретать колесо:

// let's say there is a list of 1000+ URLs
string[] urls = { "http://google.com", "http://yahoo.com", ... };

// now let's send HTTP requests to each of these URLs in parallel
await urls.ParallelForEachAsync(async (url) => {
    var client = new HttpClient();
    var html = await client.GetStringAsync(url);
}, maxDegreeOfParalellism: 20);
Серж Семенов
источник
4

Пример Тео Яунга хорош, но есть вариант без списка ожидающих задач.

 class SomeChecker
 {
    private const int ThreadCount=20;
    private CountdownEvent _countdownEvent;
    private SemaphoreSlim _throttler;

    public Task Check(IList<string> urls)
    {
        _countdownEvent = new CountdownEvent(urls.Count);
        _throttler = new SemaphoreSlim(ThreadCount); 

        return Task.Run( // prevent UI thread lock
            async  () =>{
                foreach (var url in urls)
                {
                    // do an async wait until we can schedule again
                    await _throttler.WaitAsync();
                    ProccessUrl(url); // NOT await
                }
                //instead of await Task.WhenAll(allTasks);
                _countdownEvent.Wait();
            });
    }

    private async Task ProccessUrl(string url)
    {
        try
        {
            var page = await new WebClient()
                       .DownloadStringTaskAsync(new Uri(url)); 
            ProccessResult(page);
        }
        finally
        {
            _throttler.Release();
            _countdownEvent.Signal();
        }
    }

    private void ProccessResult(string page){/*....*/}
}
Витидев
источник
4
Обратите внимание, что есть одна опасность использования этого подхода - любые исключения, возникающие в ProccessUrlили его подфункциях, фактически игнорируются. Они будут записаны в Задачи, но не будут возвращены исходному вызывающему объекту Check(...). Лично я до сих пор использую Задачи и их комбинаторные функции, такие как WhenAllи WhenAny- чтобы улучшить распространение ошибок. :)
Тео Яунг
3

SemaphoreSlim может быть здесь очень полезным. Вот созданный мной метод расширения.

    /// <summary>
    /// Concurrently Executes async actions for each item of <see cref="IEnumerable<typeparamref name="T"/>
    /// </summary>
    /// <typeparam name="T">Type of IEnumerable</typeparam>
    /// <param name="enumerable">instance of <see cref="IEnumerable<typeparamref name="T"/>"/></param>
    /// <param name="action">an async <see cref="Action" /> to execute</param>
    /// <param name="maxActionsToRunInParallel">Optional, max numbers of the actions to run in parallel,
    /// Must be grater than 0</param>
    /// <returns>A Task representing an async operation</returns>
    /// <exception cref="ArgumentOutOfRangeException">If the maxActionsToRunInParallel is less than 1</exception>
    public static async Task ForEachAsyncConcurrent<T>(
        this IEnumerable<T> enumerable,
        Func<T, Task> action,
        int? maxActionsToRunInParallel = null)
    {
        if (maxActionsToRunInParallel.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxActionsToRunInParallel.Value, maxActionsToRunInParallel.Value))
            {
                var tasksWithThrottler = new List<Task>();

                foreach (var item in enumerable)
                {
                    // Increment the number of currently running tasks and wait if they are more than limit.
                    await semaphoreSlim.WaitAsync();

                    tasksWithThrottler.Add(Task.Run(async () =>
                    {
                        await action(item).ContinueWith(res =>
                        {
                            // action is completed, so decrement the number of currently running tasks
                            semaphoreSlim.Release();
                        });
                    }));
                }

                // Wait for all of the provided tasks to complete.
                await Task.WhenAll(tasksWithThrottler.ToArray());
            }
        }
        else
        {
            await Task.WhenAll(enumerable.Select(item => action(item)));
        }
    }

Пример использования:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);
Джей Шах
источник
0

Старый вопрос, новый ответ. У @vitidev был блок кода, который был повторно использован почти без изменений в проекте, который я рассмотрел. После обсуждения с несколькими коллегами один спросил: «Почему бы вам просто не использовать встроенные методы TPL?» ActionBlock здесь выглядит победителем. https://msdn.microsoft.com/en-us/library/hh194773(v=vs.110).aspx . Вероятно, в конечном итоге не изменится какой-либо существующий код, но определенно будет стремиться принять этот nuget и повторно использовать лучшие практики г-на Софти для дросселирования параллелизма.

Нет возврата, нет возврата
источник
0

Вот решение, использующее ленивую природу LINQ. Он функционально эквивалентен принятому ответу ), но использует рабочие задачи вместо a SemaphoreSlim, уменьшая таким образом объем памяти, занимаемый всей операцией. Сначала заставим работать без троттлинга. Первый шаг - преобразовать наши URL-адреса в множество задач.

string[] urls =
{
    "https://stackoverflow.com",
    "https://superuser.com",
    "https://serverfault.com",
    "https://meta.stackexchange.com",
    // ...
};
var httpClient = new HttpClient();
var tasks = urls.Select(async (url) =>
{
    return (Url: url, Html: await httpClient.GetStringAsync(url));
});

Второй шаг - выполнить awaitвсе задачи одновременно с использованием Task.WhenAllметода:

var results = await Task.WhenAll(tasks);
foreach (var result in results)
{
    Console.WriteLine($"Url: {result.Url}, {result.Html.Length:#,0} chars");
}

Вывод:

URL: https://stackoverflow.com , 105,574 символов.
URL: https://superuser.com , 126,953 символов.
URL: https://serverfault.com , 125,963 символов.
URL: https://meta.stackexchange.com , 185,276 символов.
...

Реализация Microsoft, из Task.WhenAllматериализуешься мгновенно поставляется перечислима в массив, в результате чего все задачи на старты сразу. Нам этого не нужно, потому что мы хотим ограничить количество одновременных асинхронных операций. Итак, нам нужно реализовать альтернативу, WhenAllкоторая будет мягко и медленно перечислять наши перечисления. Мы сделаем это, создав несколько рабочих задач (равных желаемому уровню параллелизма), и каждая рабочая задача будет перечислять наши перечисляемые задачи по одной за раз, используя блокировку, чтобы гарантировать, что каждая задача url будет обработана. всего одним рабочим заданием. Затем мы awaitзавершаем все рабочие задачи и, наконец, возвращаем результаты. Вот реализация:

public static async Task<T[]> WhenAll<T>(IEnumerable<Task<T>> tasks,
    int concurrencyLevel)
{
    if (tasks is ICollection<Task<T>>) throw new ArgumentException(
        "The enumerable should not be materialized.", nameof(tasks));
    var locker = new object();
    var results = new List<T>();
    var failed = false;
    using (var enumerator = tasks.GetEnumerator())
    {
        var workerTasks = Enumerable.Range(0, concurrencyLevel)
        .Select(async _ =>
        {
            try
            {
                while (true)
                {
                    Task<T> task;
                    int index;
                    lock (locker)
                    {
                        if (failed) break;
                        if (!enumerator.MoveNext()) break;
                        task = enumerator.Current;
                        index = results.Count;
                        results.Add(default); // Reserve space in the list
                    }
                    var result = await task.ConfigureAwait(false);
                    lock (locker) results[index] = result;
                }
            }
            catch (Exception)
            {
                lock (locker) failed = true;
                throw;
            }
        }).ToArray();
        await Task.WhenAll(workerTasks).ConfigureAwait(false);
    }
    lock (locker) return results.ToArray();
}

... и вот что мы должны изменить в нашем исходном коде, чтобы добиться желаемого регулирования:

var results = await WhenAll(tasks, concurrencyLevel: 2);

Есть разница в обработке исключений. Собственный код Task.WhenAllожидает завершения всех задач и собирает все исключения. Вышеупомянутая реализация завершается сразу же после завершения первой неисправной задачи.

Теодор Зулиас
источник
Реализацию AC # 8, которая возвращает, IAsyncEnumerable<T>можно найти здесь .
Теодор Зулиас,
-1

Хотя 1000 задач могут быть поставлены в очередь очень быстро, библиотека Parallel Tasks может обрабатывать только параллельные задачи, равные количеству ядер ЦП на машине. Это означает, что если у вас четырехъядерный компьютер, только 4 задачи будут выполняться одновременно (если вы не снизите MaxDegreeOfParallelism).

Скотт
источник
8
Да, но это не относится к операциям асинхронного ввода-вывода. Приведенный выше код запускает более 1000 одновременных загрузок, даже если он выполняется в одном потоке.
Grief Coder
Не видел там awaitключевого слова. Удаление должно решить проблему, верно?
scottm
2
Библиотека, безусловно, может обрабатывать больше задач, выполняемых (со Runningстатусом) одновременно, чем количество ядер. Это будет особенно актуально для задач, связанных с вводом-выводом.
svick
@svick: ага. Вы знаете, как эффективно контролировать максимальное количество одновременных задач TPL (не потоков)?
Grief Coder
-1

Параллельные вычисления следует использовать для ускорения операций, связанных с процессором. Здесь мы говорим о связанных операциях ввода-вывода. Ваша реализация должна быть чисто асинхронной , если только вы не перегружаете загруженное одно ядро ​​вашего многоядерного процессора.

ИЗМЕНИТЬ Мне нравится предложение usr использовать здесь «асинхронный семафор».

GregC
источник
Хорошая точка зрения! Хотя каждая задача здесь будет содержать асинхронный и синхронизирующий код (страница загружается асинхронно, а затем обрабатывается синхронно). Я пытаюсь распределить синхронизирующую часть кода между процессорами и в то же время ограничить количество одновременных операций асинхронного ввода-вывода.
Grief Coder
Зачем? Потому что одновременный запуск 1000+ HTTP-запросов может не соответствовать пропускной способности сети пользователя.
Спендер
Параллельные расширения также могут использоваться как способ мультиплексирования операций ввода-вывода без необходимости вручную реализовывать чисто асинхронное решение. Что, я согласен, можно считать небрежным, но пока вы сохраняете жесткий лимит на количество одновременных операций, это, вероятно, не будет слишком сильно напрягать пул потоков.
Шон У
3
Я не думаю, что этот ответ дает ответ. Здесь недостаточно просто асинхронности: мы действительно хотим ограничивать физические операции ввода-вывода неблокирующим образом.
usr
1
Хм ... не уверен, что согласен ... при работе над большим проектом, если слишком много разработчиков примет эту точку зрения, вы получите голод, даже если вклад каждого разработчика по отдельности недостаточен, чтобы перебить ситуацию. Учитывая, что существует только один пул ThreadPool, даже если вы относитесь к нему полууважительно ... если все остальные делают то же самое, могут возникнуть проблемы. Поэтому я всегда советую не запускать длинные файлы в ThreadPool.
Спендер
-1

Использование MaxDegreeOfParallelism, которое можно указать в Parallel.ForEach():

var options = new ParallelOptions { MaxDegreeOfParallelism = 20 };

Parallel.ForEach(urls, options,
    url =>
        {
            var client = new HttpClient();
            var html = client.GetStringAsync(url);
            // do stuff with html
        });
Шон Ю
источник
4
Я не думаю, что это работает. GetStringAsync(url)предназначен для вызова с помощью await. Если вы посмотрите на тип var html, то это будет, а Task<string>не результат string.
Нил Эхардт
2
@NealEhardt прав. Parallel.ForEach(...)предназначен для параллельного выполнения блоков синхронного кода (например, в разных потоках).
Тео Яунг
-1

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

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

С Действиями

Если вы используете Actions, вы можете использовать встроенную функцию .Net Parallel.Invoke. Здесь мы ограничиваем его параллельным запуском не более 20 потоков.

var listOfActions = new List<Action>();
foreach (var url in urls)
{
    var localUrl = url;
    // Note that we create the Task here, but do not start it.
    listOfTasks.Add(new Task(() => CallUrl(localUrl)));
}

var options = new ParallelOptions {MaxDegreeOfParallelism = 20};
Parallel.Invoke(options, listOfActions.ToArray());

С задачами

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

    /// <summary>
    /// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel.
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
    /// </summary>
    /// <param name="tasksToRun">The tasks to run.</param>
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
    /// <param name="cancellationToken">The cancellation token.</param>
    public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken())
    {
        await StartAndWaitAllThrottledAsync(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken);
    }

    /// <summary>
    /// Starts the given tasks and waits for them to complete. This will run the specified number of tasks in parallel.
    /// <para>NOTE: If a timeout is reached before the Task completes, another Task may be started, potentially running more than the specified maximum allowed.</para>
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para>
    /// </summary>
    /// <param name="tasksToRun">The tasks to run.</param>
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param>
    /// <param name="timeoutInMilliseconds">The maximum milliseconds we should allow the max tasks to run in parallel before allowing another task to start. Specify -1 to wait indefinitely.</param>
    /// <param name="cancellationToken">The cancellation token.</param>
    public static async Task StartAndWaitAllThrottledAsync(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken())
    {
        // Convert to a list of tasks so that we don't enumerate over it multiple times needlessly.
        var tasks = tasksToRun.ToList();

        using (var throttler = new SemaphoreSlim(maxTasksToRunInParallel))
        {
            var postTaskTasks = new List<Task>();

            // Have each task notify the throttler when it completes so that it decrements the number of tasks currently running.
            tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release())));

            // Start running each task.
            foreach (var task in tasks)
            {
                // Increment the number of tasks currently running and wait if too many are running.
                await throttler.WaitAsync(timeoutInMilliseconds, cancellationToken);

                cancellationToken.ThrowIfCancellationRequested();
                task.Start();
            }

            // Wait for all of the provided tasks to complete.
            // We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler's using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object.
            await Task.WhenAll(postTaskTasks.ToArray());
        }
    }

А затем, создав список задач и вызвав функцию для их запуска, скажем, не более 20 одновременно, вы можете сделать это:

var listOfTasks = new List<Task>();
foreach (var url in urls)
{
    var localUrl = url;
    // Note that we create the Task here, but do not start it.
    listOfTasks.Add(new Task(async () => await CallUrl(localUrl)));
}
await Tasks.StartAndWaitAllThrottledAsync(listOfTasks, 20);
deadlydog
источник
Я думаю, вы просто указываете initialCount для SemaphoreSlim, и вам нужно указать второй параметр, то есть maxCount, в конструкторе SemaphoreSlim.
Джей Шах
Я хочу, чтобы каждый ответ от каждой задачи был преобразован в список. Как я могу получить результат возврата или ответ
Venkat
-1

это не лучшая практика, поскольку она изменяет глобальную переменную. это также не общее решение для async. но это легко для всех экземпляров HttpClient, если это все, что вам нужно. вы можете просто попробовать:

System.Net.ServicePointManager.DefaultConnectionLimit = 20;
симбионт
источник