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

Voici le problème, il démarre plus de 1000 requêtes Web simultanées. Existe-t-il un moyen simple de limiter le nombre simultané de ces requêtes http asynchrones? Ainsi, pas plus de 20 pages Web ne sont téléchargées à un moment donné. Comment le faire de la manière la plus efficace?

Vous pouvez certainement le faire dans les dernières versions d'async pour .NET, en utilisant .NET 4.5 Beta. Le post précédent de 'usr' pointe vers un bon article écrit par Stephen Toub, mais la nouvelle moins annoncée est que le sémaphore asynchrone est en fait entré dans la version bêta de .NET 4.5

Si vous regardez notre SemaphoreSlimclasse bien - aimée (que vous devriez utiliser car elle est plus performante que l'original Semaphore), elle possède désormais la WaitAsync(...)série de surcharges, avec tous les arguments attendus - intervalles de temporisation, jetons d'annulation, tous vos amis de planification habituels: )

Stephen a également écrit un article de blog plus récent sur les nouveaux goodies .NET 4.5 sortis avec la version bêta, voir Quoi de neuf pour le parallélisme dans .NET 4.5 bêta .

Enfin, voici un exemple de code sur l'utilisation de SemaphoreSlim pour la limitation des méthodes asynchrones:

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

Enfin, mais probablement une mention digne d'être mentionnée, une solution qui utilise la planification basée sur TPL. Vous pouvez créer des tâches liées aux délégués sur le TPL qui n'ont pas encore été démarrées et permettre à un planificateur de tâches personnalisé de limiter la concurrence. En fait, il existe un exemple MSDN pour cela ici:

Voir aussi TaskScheduler .

Si vous avez un IEnumerable (c.-à-d. Des chaînes d'URL) et que vous souhaitez effectuer une opération liée aux E / S avec chacun d'entre eux (c.-à-d. Faire une requête http asynchrone) simultanément ET éventuellement, vous souhaitez également définir le nombre maximal de Demandes d'E / S en temps réel, voici comment vous pouvez le faire. De cette façon, vous n'utilisez pas de pool de threads et autres, la méthode utilise semaphoreslim pour contrôler le nombre maximal de requêtes d'E / S simultanées, similaire à un modèle de fenêtre glissante qu'une requête se termine, quitte le sémaphore et la suivante entre.

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

Malheureusement, le .NET Framework manque les combinateurs les plus importants pour l'orchestration des tâches asynchrones parallèles. Il n'y a pas une telle chose intégrée.

Regardez la classe AsyncSemaphore construite par le plus respectable Stephen Toub. Ce que vous voulez s'appelle un sémaphore, et vous en avez besoin d'une version asynchrone.

Il y a beaucoup de pièges et l'utilisation directe d'un sémaphore peut être délicate dans les cas d'erreur, donc je suggérerais d'utiliser AsyncEnumerator NuGet Package au lieu de réinventer la roue:

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

L'exemple de Theo Yaung est sympa, mais il existe une variante sans liste de tâches en attente.

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

SemaphoreSlim peut être très utile ici. Voici la méthode d'extension que j'ai créée.

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

Exemple d'utilisation:

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

Ancienne question, nouvelle réponse. @vitidev avait un bloc de code qui a été réutilisé presque intact dans un projet que j'ai examiné. Après avoir discuté avec quelques collègues, l'un d'eux a demandé "Pourquoi n'utilisez-vous pas simplement les méthodes TPL intégrées?" ActionBlock ressemble au gagnant là-bas. https://msdn.microsoft.com/en-us/library/hh194773(v=vs.110).aspx . Probablement ne finira pas par changer de code existant, mais cherchera certainement à adopter ce nuget et à réutiliser les meilleures pratiques de M. Softy pour le parallélisme limité.

Voici une solution qui tire parti de la nature paresseuse de LINQ. Elle est fonctionnellement équivalente à la réponse acceptée ), mais utilise des tâches de travail au lieu de a SemaphoreSlim, réduisant ainsi l'empreinte mémoire de l'ensemble de l'opération. Au début, faisons en sorte que cela fonctionne sans étranglement. La première étape consiste à convertir nos URL en une liste de tâches.

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

La deuxième étape consiste à awaitexécuter toutes les tâches simultanément en utilisant la Task.WhenAllméthode:

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

Production:

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

L'implémentation de MicrosoftTask.WhenAll matérialise instantanément l'énumérable fourni dans un tableau, provoquant le démarrage simultané de toutes les tâches. Nous ne voulons pas de cela, car nous voulons limiter le nombre d'opérations asynchrones simultanées. Nous devrons donc implémenter une alternative WhenAllqui énumérera notre énumérable doucement et lentement. Nous le ferons en créant un certain nombre de tâches de travail (égal au niveau souhaité de concurrence), et chaque tâche de travail énumérera notre tâche énumérable une tâche à la fois, en utilisant un verrou pour garantir que chaque tâche d'URL sera traitée par une seule tâche de travail. Ensuite, nous awaitpour que toutes les tâches de travail soient terminées, et enfin nous renvoyons les résultats. Voici la mise en œuvre:

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

... et voici ce que nous devons changer dans notre code initial, pour atteindre la limitation souhaitée:

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

Il y a une différence concernant le traitement des exceptions. Le natif Task.WhenAllattend que toutes les tâches soient terminées et regroupe toutes les exceptions. L'implémentation ci-dessus se termine rapidement après l'achèvement de la première tâche défaillante.

Bien que 1000 tâches puissent être mises en file d'attente très rapidement, la bibliothèque de tâches parallèles ne peut gérer que des tâches simultanées égales à la quantité de cœurs de processeur de la machine. Cela signifie que si vous avez une machine à quatre cœurs, seules 4 tâches seront exécutées à un moment donné (sauf si vous réduisez le MaxDegreeOfParallelism).

Les calculs parallèles doivent être utilisés pour accélérer les opérations liées au processeur. Nous parlons ici d'opérations liées aux E / S. Votre implémentation doit être purement asynchrone , à moins que vous ne submergiez le cœur unique occupé sur votre processeur multicœur.

EDIT J'aime la suggestion faite par usr d'utiliser un "sémaphore asynchrone" ici.

Utilisez MaxDegreeOfParallelism, qui est une option que vous pouvez spécifier dans 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
        });

Essentiellement, vous allez vouloir créer une action ou une tâche pour chaque URL sur laquelle vous souhaitez accéder, les mettre dans une liste, puis traiter cette liste, en limitant le nombre qui peut être traité en parallèle.

Mon article de blog montre comment faire cela à la fois avec des tâches et avec des actions, et fournit un exemple de projet que vous pouvez télécharger et exécuter pour voir les deux en action.

Avec des actions

Si vous utilisez Actions, vous pouvez utiliser la fonction intégrée .Net Parallel.Invoke. Ici, nous le limitons à l'exécution d'au plus 20 threads en parallèle.

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

Avec des tâches

Avec les tâches, il n'y a pas de fonction intégrée. Cependant, vous pouvez utiliser celui que je propose sur mon 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());
        }
    }

Et puis en créant votre liste de tâches et en appelant la fonction pour les exécuter, avec un maximum de 20 tâches simultanées à la fois, vous pouvez le faire:

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

ce n'est pas une bonne pratique car cela modifie une variable globale. ce n'est pas non plus une solution générale pour async. mais c'est facile pour toutes les instances de HttpClient, si c'est tout ce que vous recherchez. vous pouvez simplement essayer:

System.Net.ServicePointManager.DefaultConnectionLimit = 20;