// 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);
});

Aqui está o problema, ele inicia mais de 1000 solicitações da web simultâneas. Existe uma maneira fácil de limitar a quantidade simultânea dessas solicitações http assíncronas? Para que não mais do que 20 páginas da web sejam baixadas a qualquer momento. Como fazer da maneira mais eficiente?

Definitivamente, você pode fazer isso nas versões mais recentes do async para .NET, usando .NET 4.5 Beta. O post anterior de 'usr' aponta para um bom artigo escrito por Stephen Toub, mas a notícia menos anunciada é que o semáforo assíncrono realmente chegou à versão Beta do .NET 4.5

Se você olhar para nossa amada SemaphoreSlimclasse (que você deveria usar, pois tem mais desempenho do que a original Semaphore), ela agora possui uma WaitAsync(...)série de sobrecargas, com todos os argumentos esperados - intervalos de tempo limite, tokens de cancelamento, todos os seus amigos de programação usuais: )

Stephen também escreveu uma postagem no blog mais recente sobre os novos recursos do .NET 4.5 que saíram com o beta, consulte Novidades para paralelismo no .NET 4.5 Beta .

Por último, aqui está um exemplo de código sobre como usar SemaphoreSlim para limitação de método assíncrono:

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)
}

Por último, mas provavelmente uma menção digna é uma solução que usa agendamento baseado em TPL. Você pode criar tarefas vinculadas a delegados no TPL que ainda não foram iniciadas e permitir que um agendador de tarefas personalizado limite a simultaneidade. Na verdade, há uma amostra do MSDN aqui:

Veja também TaskScheduler .

Se você tiver um IEnumerable (ou seja, strings de URL s) e quiser fazer uma operação de ligação de E / S com cada um deles (ou seja, fazer uma solicitação assíncrona de http) simultaneamente E, opcionalmente, também deseja definir o número máximo de simultâneos Solicitações de E / S em tempo real, veja como você pode fazer isso. Dessa forma, você não usa pool de threads e outros, o método usa semáforoslim para controlar o máximo de solicitações de E / S simultâneas, semelhante a um padrão de janela deslizante que uma solicitação completa, deixa o semáforo e a próxima entra.

uso: 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);
    }

Infelizmente, o .NET Framework está perdendo os combinadores mais importantes para orquestrar tarefas assíncronas paralelas. Não existe tal coisa embutida.

Veja a classe AsyncSemaphore construída pelo mais respeitável Stephen Toub. O que você quer é chamado de semáforo e precisa de uma versão assíncrona dele.

Existem muitas armadilhas e o uso direto de um semáforo pode ser complicado em casos de erro, então eu sugeriria usar o Pacote NuGet AsyncEnumerator em vez de reinventar a roda:

// 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);

O exemplo de Theo Yaung é bom, mas há uma variante sem lista de tarefas em espera.

 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){/*....*/}
}

O SemaphoreSlim pode ser muito útil aqui. Aqui está o método de extensão que criei.

    /// <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)));
        }
    }

Uso de amostra:

await enumerable.ForEachAsyncConcurrent(
    async item =>
    {
        await SomeAsyncMethod(item);
    },
    5);

Velha pergunta, nova resposta. @vitidev tinha um bloco de código que foi reutilizado quase intacto em um projeto que analisei. Depois de discutir com alguns colegas, um perguntou "Por que você simplesmente não usa os métodos TPL embutidos?" ActionBlock parece ser o vencedor. https://msdn.microsoft.com/en-us/library/hh194773(v=vs.110).aspx . Provavelmente não acabará mudando nenhum código existente, mas definitivamente tentará adotar esse nuget e reutilizar a prática recomendada do Sr. Softy para paralelismo limitado.

Aqui está uma solução que tira proveito da natureza preguiçosa do LINQ. É funcionalmente equivalente à resposta aceita ), mas usa tarefas de trabalho em vez de a SemaphoreSlim, reduzindo assim o consumo de memória de toda a operação. A princípio vamos fazer funcionar sem estrangulamento. O primeiro passo é converter nossos urls em uma lista de tarefas.

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));
});

A segunda etapa é fazer awaittodas as tarefas simultaneamente usando o Task.WhenAllmétodo:

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

Resultado:

Url: https://stackoverflow.com , 105.574 chars
Url: https://superuser.com , 126.953 chars
Url: https://serverfault.com , 125.963 chars
Url: https://meta.stackexchange.com , 185.276 chars
...

A implementação de MicrosoftTask.WhenAll materializa instantaneamente o enumerável fornecido a um array, fazendo com que todas as tarefas sejam iniciadas de uma vez. Não queremos isso, porque queremos limitar o número de operações assíncronas simultâneas. Portanto, precisaremos implementar uma alternativa WhenAllque enumerará nosso enumerável suave e lentamente. Faremos isso criando uma série de tarefas de trabalho (igual ao nível desejado de simultaneidade), e cada tarefa de trabalho irá enumerar nossa tarefa enumerável por vez, usando um bloqueio para garantir que cada tarefa de url será processada por apenas um trabalhador-tarefa. Em seguida, concluímos awaittodas as tarefas do trabalhador e, finalmente, retornamos os resultados. Aqui está a implementação:

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();
}

... e aqui está o que devemos mudar em nosso código inicial, para atingir a limitação desejada:

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

Há uma diferença quanto ao tratamento das exceções. O nativo Task.WhenAllespera que todas as tarefas sejam concluídas e agrega todas as exceções. A implementação acima termina imediatamente após a conclusão da primeira tarefa com falha.

Embora 1000 tarefas possam ser enfileiradas muito rapidamente, a biblioteca de Tarefas Paralelas só pode lidar com tarefas simultâneas iguais à quantidade de núcleos de CPU na máquina. Isso significa que se você tiver uma máquina de quatro núcleos, apenas 4 tarefas serão executadas em um determinado momento (a menos que você diminua o MaxDegreeOfParallelism).

Cálculos paralelos devem ser usados ​​para acelerar as operações vinculadas à CPU. Aqui, estamos falando sobre operações vinculadas a E / S. Sua implementação deve ser puramente assíncrona , a menos que você esteja sobrecarregando o único núcleo ocupado em sua CPU multi-core.

EDITAR Eu gosto da sugestão feita por usr para usar um "semáforo assíncrono" aqui.

Use MaxDegreeOfParallelism, que é uma opção que você pode especificar em 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
        });

Essencialmente, você vai querer criar uma ação ou tarefa para cada URL que deseja acessar, colocá-los em uma lista e, em seguida, processar essa lista, limitando o número que pode ser processado em paralelo.

Minha postagem do blog mostra como fazer isso com Tarefas e com Ações e fornece um projeto de amostra que você pode baixar e executar para ver os dois em ação.

Com ações

Se estiver usando Actions, você pode usar a função integrada .Net Parallel.Invoke. Aqui, limitamos a execução de no máximo 20 threads em paralelo.

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());

Com Tarefas

Com o Tarefas, não há função incorporada. No entanto, você pode usar o que eu disponibilizo no meu blog.

    /// <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());
        }
    }

E, em seguida, criar sua lista de tarefas e chamar a função para executá-las, digamos, no máximo 20 simultâneas por vez, você pode fazer o seguinte:

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);

esta não é uma boa prática, pois altera uma variável global. também não é uma solução geral para assíncrono. mas é fácil para todas as instâncias de HttpClient, se é isso que você está procurando. você pode simplesmente tentar:

System.Net.ServicePointManager.DefaultConnectionLimit = 20;