Как я могу использовать Async с ForEach?

Ответ 1

List<T>.ForEach не особенно хорошо работает с async (также как и LINQ-to-objects по тем же причинам).

В этом случае я рекомендую проецировать каждый элемент в асинхронную операцию, и затем вы можете (асинхронно) дождаться их завершения.

using (DataContext db = new DataLayer.DataContext())
{
    var tasks = db.Groups.ToList().Select(i => GetAdminsFromGroupAsync(i.Gid));
    var results = await Task.WhenAll(tasks);
}

Преимущества такого подхода по сравнению с передачей async делегата ForEach:

1. Обработка ошибок более правильная. Исключения из async void нельзя отловить с помощью catch; этот подход будет распространять исключения в строке await Task.WhenAll, обеспечивая естественную обработку исключений.
2. Вы знаете, что задачи завершены в конце этого метода, поскольку он выполняет await Task.WhenAll. Если вы используете async void, вы не можете легко определить, когда завершены операции.
3. Этот подход имеет естественный синтаксис для получения результатов. GetAdminsFromGroupAsync звучит как операция, которая выдает результат (администраторы), и такой код более естественен, если такие операции могут возвращать свои результаты, а не устанавливать значение в качестве побочного эффекта.

Ответ 2

Этот небольшой метод расширения должен предоставить вам безопасную асинхронную итерацию:

public static async Task ForEachAsync<T>(this List<T> list, Func<T, Task> func)
{
    foreach (var value in list)
    {
        await func(value);
    }
}

Поскольку мы меняем тип возвращаемого значения лямбда от void до Task, исключения будут корректно распространяться. Это позволит вам написать что-то подобное на практике:

await db.Groups.ToList().ForEachAsync(async i => {
    await GetAdminsFromGroup(i.Gid);
});

Ответ 3

Вот фактическая рабочая версия вышеперечисленных вариантов async foreach с последовательной обработкой:

public static async Task ForEachAsync<T>(this List<T> enumerable, Action<T> action)
{
    foreach (var item in enumerable)
        await Task.Run(() => { action(item); }).ConfigureAwait(false);
}

Вот реализация:

public async void SequentialAsync()
{
    var list = new List<Action>();

    Action action1 = () => {
        //do stuff 1
    };

    Action action2 = () => {
        //do stuff 2
    };

    list.Add(action1);
    list.Add(action2);

    await list.ForEachAsync();
}

Какое ключевое различие? .ConfigureAwait(false);, который сохраняет контекст основного потока при асинхронной последовательной обработке каждой задачи.

Ответ 4

Добавьте этот метод расширения

public static class ForEachAsyncExtension
{
    public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body)
    {
        return Task.WhenAll(from partition in Partitioner.Create(source).GetPartitions(dop) 
            select Task.Run(async delegate
            {
                using (partition)
                    while (partition.MoveNext())
                        await body(partition.Current).ConfigureAwait(false);
            }));
    }
}

Затем используйте так:

Task.Run(async () =>
{
    var s3 = new AmazonS3Client(Config.Instance.Aws.Credentials, Config.Instance.Aws.RegionEndpoint);
    var buckets = await s3.ListBucketsAsync();

    foreach (var s3Bucket in buckets.Buckets)
    {
        if (s3Bucket.BucketName.StartsWith("mybucket-"))
        {
            log.Information("Bucket => {BucketName}", s3Bucket.BucketName);

            ListObjectsResponse objects;
            try
            {
                objects = await s3.ListObjectsAsync(s3Bucket.BucketName);
            }
            catch
            {
                log.Error("Error getting objects. Bucket => {BucketName}", s3Bucket.BucketName);
                continue;
            }

            // ForEachAsync (4 is how many tasks you want to run in parallel)
            await objects.S3Objects.ForEachAsync(4, async s3Object =>
            {
                try
                {
                    log.Information("Bucket => {BucketName} => {Key}", s3Bucket.BucketName, s3Object.Key);
                    await s3.DeleteObjectAsync(s3Bucket.BucketName, s3Object.Key);
                }
                catch
                {
                    log.Error("Error deleting bucket {BucketName} object {Key}", s3Bucket.BucketName, s3Object.Key);
                }
            });

            try
            {
                await s3.DeleteBucketAsync(s3Bucket.BucketName);
            }
            catch
            {
                log.Error("Error deleting bucket {BucketName}", s3Bucket.BucketName);
            }
        }
    }
}).Wait();

Ответ 5

Проблема заключалась в том, что ключевое слово async должно появляться перед лямбдой, а не перед телом:

db.Groups.ToList().ForEach(async (i) => {
    await GetAdminsFromGroup(i.Gid);
});

Ответ 6

Простой ответ заключается в использовании ключевого слова foreach вместо метода ForEach() объекта List().

using (DataContext db = new DataLayer.DataContext())
{
    foreach(var i in db.Groups)
    {
        await GetAdminsFromGroup(i.Gid);
    }
}

Ответ 7

Начиная с C# 8.0, вы можете создавать и потреблять потоки асинхронно.

    private async void button1_Click(object sender, EventArgs e)
    {
        IAsyncEnumerable<int> enumerable = GenerateSequence();

        await foreach (var i in enumerable)
        {
            Debug.WriteLine(i);
        }
    }

    public static async IAsyncEnumerable<int> GenerateSequence()
    {
        for (int i = 0; i < 20; i++)
        {
            await Task.Delay(100);
            yield return i;
        }
    }

Подробнее