221

コレクションを並行して処理したいのですが、実装に問題があるため、助けを求めています。

並列ループのラムダ内で、C# で async とマークされたメソッドを呼び出したい場合に問題が発生します。例えば：

var bag = new ConcurrentBag<object>();
Parallel.ForEach(myCollection, async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
}
var count = bag.Count;

この問題はカウントが 0 の場合に発生します。これは、作成されたすべてのスレッドが事実上単なるバックグラウンド スレッドであり、Parallel.ForEach呼び出しが完了を待たないためです。async キーワードを削除すると、メソッドは次のようになります。

var bag = new ConcurrentBag<object>();
Parallel.ForEach(myCollection, item =>
{
  // some pre stuff
  var responseTask = await GetData(item);
  responseTask.Wait();
  var response = responseTask.Result;
  bag.Add(response);
  // some post stuff
}
var count = bag.Count;

動作しますが、await の賢さが完全に無効になり、手動で例外処理を行う必要があります.. (簡潔にするために削除されました)。

Parallel.ForEachラムダ内で await キーワードを使用するループを実装するにはどうすればよいですか? 出来ますか？

Parallel.ForEach メソッドのプロトタイプはAction<T>as パラメーターを取りますが、非同期ラムダを待機させたいと考えています。

4

11 に答える 11

272

単純な並列処理が必要な場合は、次のように実行できます。

var bag = new ConcurrentBag<object>();
var tasks = myCollection.Select(async item =>
{
  // some pre stuff
  var response = await GetData(item);
  bag.Add(response);
  // some post stuff
});
await Task.WhenAll(tasks);
var count = bag.Count;

より複雑なものが必要な場合は、StephenToubのForEachAsync投稿を確認してください。

于 2013-02-28T13:30:44.903 に答える
5

ParallelForEach 非同期の軽量実装。

特徴：

  1. スロットリング (並列処理の最大次数)。
  2. 例外処理 (完了時に集計例外がスローされます)。
  3. メモリ効率が良い (タスクのリストを保存する必要がない)。

public static class AsyncEx
{
    public static async Task ParallelForEachAsync<T>(this IEnumerable<T> source, Func<T, Task> asyncAction, int maxDegreeOfParallelism = 10)
    {
        var semaphoreSlim = new SemaphoreSlim(maxDegreeOfParallelism);
        var tcs = new TaskCompletionSource<object>();
        var exceptions = new ConcurrentBag<Exception>();
        bool addingCompleted = false;

        foreach (T item in source)
        {
            await semaphoreSlim.WaitAsync();
            asyncAction(item).ContinueWith(t =>
            {
                semaphoreSlim.Release();

                if (t.Exception != null)
                {
                    exceptions.Add(t.Exception);
                }

                if (Volatile.Read(ref addingCompleted) && semaphoreSlim.CurrentCount == maxDegreeOfParallelism)
                {
                    tcs.TrySetResult(null);
                }
            });
        }

        Volatile.Write(ref addingCompleted, true);
        await tcs.Task;
        if (exceptions.Count > 0)
        {
            throw new AggregateException(exceptions);
        }
    }
}

使用例:

await Enumerable.Range(1, 10000).ParallelForEachAsync(async (i) =>
{
    var data = await GetData(i);
}, maxDegreeOfParallelism: 100);
于 2018-10-17T11:01:56.637 に答える
1

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="maxDegreeOfParallelism">Optional, An integer that represents the maximum degree of parallelism,
    /// 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? maxDegreeOfParallelism = null)
    {
        if (maxDegreeOfParallelism.HasValue)
        {
            using (var semaphoreSlim = new SemaphoreSlim(
                maxDegreeOfParallelism.Value, maxDegreeOfParallelism.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 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);
于 2018-05-09T22:39:16.617 に答える
1

受け入れられた回答では、 ConcurrentBag は必要ありません。それなしの実装は次のとおりです。

var tasks = myCollection.Select(GetData).ToList();
await Task.WhenAll(tasks);
var results = tasks.Select(t => t.Result);

「// 前のもの」と「// 後のもの」のいずれも、GetData 実装 (または GetData を呼び出す別のメソッド) に入れることができます。

短いことを除けば、アンチパターンである「async void」ラムダは使用されません。

于 2021-01-18T10:30:23.080 に答える