C# 如何将 Async 与 ForEach 结合使用？

How can I use Async with ForEach?

提问by James Jeffery

Is it possible to use Async when using ForEach? Below is the code I am trying:

使用 ForEach 时是否可以使用 Async？下面是我正在尝试的代码：

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

I am getting the error:

我收到错误：

The name 'Async' does not exist in the current context

当前上下文中不存在名称“Async”

The method the using statement is enclosed in is set to async.

using 语句所包含的方法设置为 async。

回答by James Jeffery

The problem was that the asynckeyword needs to appear before the lambda, not before the body:

问题是async关键字需要出现在 lambda 之前，而不是正文之前：

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

回答by Stephen Cleary

List<T>.ForEachdoesn't play particularly well with async(neither does LINQ-to-objects, for the same reasons).

List<T>.ForEach不太好用async（LINQ-to-objects 也不是，出于同样的原因）。

In this case, I recommend projectingeach element into an asynchronous operation, and you can then (asynchronously) wait for them all to complete.

在这种情况下，我建议将每个元素投影到异步操作中，然后您可以（异步）等待它们全部完成。

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

The benefits of this approach over giving an asyncdelegate to ForEachare:

这种方法比给予async委托的好处ForEach是：

1. Error handling is more proper. Exceptions from async voidcannot be caught with catch; this approach will propagate exceptions at the await Task.WhenAllline, allowing natural exception handling.
2. You know that the tasks are complete at the end of this method, since it does an await Task.WhenAll. If you use async void, you cannot easily tell when the operations have completed.
3. This approach has a natural syntax for retrieving the results. GetAdminsFromGroupAsyncsounds like it's an operation that produces a result (the admins), and such code is more natural if such operations can returntheir results rather than setting a value as a side effect.

1. 错误处理更妥当。异常async void不能被捕获catch；这种方法将在await Task.WhenAll生产线上传播异常，允许自然的异常处理。
2. 您知道任务在此方法结束时完成，因为它执行await Task.WhenAll. 如果使用async void，则无法轻易判断操作何时完成。
3. 这种方法具有用于检索结果的自然语法。GetAdminsFromGroupAsync听起来这是一个产生结果的操作（管理员），如果此类操作可以返回其结果而不是将值设置为副作用，则此类代码更自然。

回答by JD Courtoy

This little extension method should give you exception-safe async iteration:

这个小扩展方法应该为您提供异常安全的异步迭代：

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

Since we're changing the return type of the lambda from voidto Task, exceptions will propagate up correctly. This will allow you to write something like this in practice:

由于我们将 lambda 的返回类型从void更改为Task，因此异常将正确传播。这将允许您在实践中编写如下内容：

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

回答by superlogical

Add this extension method

添加此扩展方法

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

And then use like so:

然后像这样使用：

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

回答by mrogunlana

Here is an actual working version of the above async foreach variants with sequential processing:

这是上述带有顺序处理的异步 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);
}

Here is the implementation:

这是实现：

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

What's the key difference? .ConfigureAwait(false);which keeps the context of main thread while async sequential processing of each task.

关键区别是什么？.ConfigureAwait(false);它在每个任务的异步顺序处理时保持主线程的上下文。

回答by RubberDuck

The simple answer is to use the foreachkeyword instead of the ForEach()method of List().

简单的答案是使用foreach关键字而不是 的ForEach()方法List()。

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

回答by Andrei Krasutski

Starting with C# 8.0, you can create and consume streams asynchronously.

从 开始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;
        }
    }

More

更多的