T:System.Threading.Tasks.Task

タスクベースの非同期パターン (TAP) を使用して非同期操作を操作する場合は、コールバックを使用して、ブロックせずに待機を実現できます。タスクの場合、このパターンでは Task.ContinueWith などのメソッドが使用されます。言語ベースの非同期サポートでは、通常の制御フロー内で非同期操作を待機できるようにすることでコールバックを非表示にし、コンパイラによって生成されたコードでも同じ API レベルのサポートが提供されます。

Await を使用した実行の中断

C# の await キーワードと Visual Basic の Await 演算子を使用して、 Task オブジェクトと Task<TResult> オブジェクトを非同期的に待機できます。 Taskを待機すると、await式はvoid型になります。 Task<TResult>を待機すると、await式はTResult型になります。 await式は、非同期メソッドの本体内で行う必要があります。

待機機能は、隠れた状態で継続を使用してタスクにコールバックをインストールします。このコールバックは、中断の時点で非同期メソッドを再開します。非同期メソッドが再開されると、待機中の操作が正常に完了し、 Task<TResult>であった場合は、その TResult が返されます。待っていた Task または Task<TResult> が Canceled 状態で終わった場合、OperationCanceledException 例外がスローされます。待っていた Task または Task<TResult> が Faulted 状態で終わった場合、エラーの原因となった例外がスローされます。 Taskは複数の例外の結果としてエラーが発生する可能性がありますが、これらの例外の 1 つだけが伝達されます。ただし、 Task.Exception プロパティは、すべてのエラーを含む AggregateException 例外を返します。

同期コンテキスト (SynchronizationContext オブジェクト) が、中断時に非同期メソッドを実行していたスレッドに関連付けられている場合 (たとえば、 SynchronizationContext.Current プロパティが nullされていない場合)、非同期メソッドはコンテキストの Post メソッドを使用して、同じ同期コンテキストで再開します。それ以外の場合は、中断時に現在のタスクスケジューラ (TaskScheduler オブジェクト) に依存します。通常、これはスレッドプールを対象とする既定のタスクスケジューラ (TaskScheduler.Default) です。このタスクスケジューラは、待機中の非同期操作が完了した場所で再開するか、再開をスケジュールするかを決定します。通常、既定のスケジューラでは、待機中の操作が完了したスレッドで継続を実行できます。

非同期メソッドを呼び出すと、関数の本体は、完了していない待機可能なインスタンスで最初の await 式に達するまで同期的に実行され、その時点で呼び出しは呼び出し元に戻ります。非同期メソッドが voidを返さない場合は、進行中の計算を表す Task または Task<TResult> オブジェクトを返します。 void 以外の非同期メソッドでは、return ステートメントが検出された場合、またはメソッド本体の末尾に達した場合、タスクは RanToCompletion 最終状態で完了します。ハンドルされない例外によって制御が非同期メソッドの本体から離れる場合、タスクは Faulted 状態で終了します。その例外が OperationCanceledExceptionの場合、タスクは代わりに Canceled 状態で終了します。この方法では、結果または例外が最終的に公開されます。

この動作にはいくつかの重要なバリエーションがあります。パフォーマンス上の理由から、タスクが待機するまでにタスクが既に完了している場合、制御は生成されず、関数は引き続き実行されます。さらに、元のコンテキストに戻すことは常に目的の動作ではなく、変更できます。この動作については、次のセクションで詳しく説明します。

Yield と ConfigureAwait を使用した中断と再開の構成

いくつかのメソッドは、非同期メソッドの実行をより詳細に制御します。たとえば、 Task.Yield メソッドを使用して、非同期メソッドに yield ポイントを導入できます。

public class Task : …
{
    public static YieldAwaitable Yield();
    …
}

このメソッドは、現在のコンテキストに非同期的にポストまたはスケジュールを戻すのと同じです。

public static async Task YieldLoopExample()
{
    await Task.Run(async delegate
    {
        for (int i = 0; i < 1000000; i++)
        {
            await Task.Yield(); // fork the continuation into a separate work item
        }
    });
}

Public Async Function YieldLoopExample() As Task
    Await Task.Run(Async Function()
                       For i As Integer = 0 To 999999
                           Await Task.Yield() ' fork the continuation into a separate work item
                       Next
                   End Function)
End Function

また、 Task.ConfigureAwait メソッドを使用して、非同期メソッドの中断と再開をより適切に制御することもできます。前述のように、既定では、現在のコンテキストは非同期メソッドが中断された時点でキャプチャされ、キャプチャされたコンテキストは再開時に非同期メソッドの継続を呼び出すために使用されます。多くの場合、これは必要な正確な動作です。それ以外の場合は、継続コンテキストを気にせず、そのような投稿を元のコンテキストに戻さないようにすることで、パフォーマンスを向上させることができます。この動作を有効にするには、 Task.ConfigureAwait メソッドを使用して、コンテキストでキャプチャおよび再開するのではなく、待機中の非同期操作が完了した場所で実行を続行するように await 操作に通知します。

await someTask.ConfigureAwait(continueOnCapturedContext:false);

Awaitables、ConfigureAwait、SynchronizationContext

await は、のみではなく、Taskを満たす任意の型で動作します。型は、互換性のあるGetAwaiterメソッドを提供し、そのメソッドがIsCompleted、OnCompleted、およびGetResultメンバーを持つ型を返す場合、待機可能です。ほとんどのパブリック API では、 Task、 Task<TResult>、 ValueTask、または ValueTask<TResult>を返します。特殊なシナリオでのみ、カスタム待機可能値を使用します。

継続に呼び出し元のコンテキストが必要ない場合は、 ConfigureAwait を使用します。 UI を更新するアプリコードでは、多くの場合、コンテキストキャプチャが必要です。再利用可能なライブラリコードでは、不要なコンテキストホップを回避し、ブロックする呼び出し元のデッドロックリスクを軽減するため、通常、 ConfigureAwait(false) をお勧めします。

ConfigureAwait(false) はフローではなく、継続スケジュールを変更しますExecutionContext 。コンテキストの動作の詳細については、 ExecutionContext と SynchronizationContext に関する記事を参照してください。

非同期操作の取り消し

取り消しをサポートする TAP メソッドは、キャンセルトークン (CancellationToken オブジェクト) を受け入れるオーバーロードを少なくとも 1 つ提供します。

キャンセルトークンソース (CancellationTokenSource オブジェクト) を使用してキャンセルトークンを作成します。ソースの Token プロパティは、ソースの Cancel メソッドが呼び出されたときに通知するキャンセルトークンを返します。

var cts = new CancellationTokenSource();
string result = await DownloadStringTaskAsync(url, cts.Token);
… // at some point later, potentially on another thread
cts.Cancel();

たとえば、1 つの Web ページをダウンロードして操作を取り消すことができるようにするには、 CancellationTokenSource オブジェクトを作成し、そのトークンを TAP メソッドに渡し、操作を取り消す準備ができたらソースの Cancel メソッドを呼び出します。

var cts = new CancellationTokenSource();
    IList<string> results = await Task.WhenAll(from url in urls select DownloadStringTaskAsync(url, cts.Token));
    // at some point later, potentially on another thread
    …
    cts.Cancel();

または、同じトークンを操作の選択的なサブセットに渡すことができます。

var cts = new CancellationTokenSource();
    byte [] data = await DownloadDataAsync(url, cts.Token);
    await SaveToDiskAsync(outputPath, data, CancellationToken.None);
    … // at some point later, potentially on another thread
    cts.Cancel();

Von Bedeutung

どのスレッドでもキャンセル要求を開始できます。

取り消しトークンを受け取る任意のメソッドに CancellationToken.None 値を渡して、取り消しが要求されていないことを示すことができます。この値により、 CancellationToken.CanBeCanceled プロパティは falseを返し、呼び出されたメソッドはそれに応じて最適化できます。テストのためには、トークンが既に取り消された状態か取り消し不可能な状態で開始するかどうかを示すブール値を受け取るコンストラクターを使用してインスタンス化した、事前に取り消されたキャンセルトークンを渡すこともできます。

キャンセルに対するこのアプローチには、いくつかの利点があります。

同じキャンセルトークンを任意の数の非同期操作と同期操作に渡すことができます。
同じキャンセル要求を任意の数のリスナーに送信できます。
非同期 API の開発者は、取り消しを要求できるかどうかと、いつ有効になるかを完全に制御できます。
API を活用しているコードは、キャンセル要求が届けられる非同期呼び出しを選択的に決定できます。

進行状況の監視

一部の非同期メソッドは、非同期メソッドに渡す進行状況インターフェイスを介して進行状況を公開します。たとえば、テキストの文字列を非同期的にダウンロードし、その過程で、これまでに完了したダウンロードの割合を含む進行状況の更新を発生させる関数を考えてみましょう。このようなメソッドは、次のようにWindows Presentation Foundation (WPF) アプリケーションで使用できます。

private async void btnDownload_Click(object sender, RoutedEventArgs e)
{
    btnDownload.IsEnabled = false;
    try
    {
        txtResult.Text = await DownloadStringTaskAsync(txtUrl.Text,
            new Progress<int>(p => pbDownloadProgress.Value = p));
    }
    finally { btnDownload.IsEnabled = true; }
}

組み込みのタスクベースの組み合わせ子の使用

System.Threading.Tasks名前空間には、タスクを作成および操作するためのいくつかのメソッドが含まれています。

注

このセクションのいくつかのコードサンプルでは、Bitmap を使用します。これには System.Drawing.Common パッケージが必要であり、Windowsでのみサポートされています。彼らが示す非同期パターンはすべてのプラットフォームに適用されます。Windows以外のターゲットでは、クロスプラットフォームのイメージングライブラリを使用することに置き換えてください。

Task.Run

Task クラスには、Runとして作業を簡単にオフロードしたり、スレッドプールにTaskしたりできる、いくつかのTask<TResult>メソッドが含まれています。例えば次が挙げられます。

public static async Task TaskRunBasicExample()
{
    int answer = 42;
    string result = await Task.Run(() =>
    {
        // … do compute-bound work here
        return answer.ToString();
    });
    Console.WriteLine(result);
}

Public Async Function TaskRunBasicExample() As Task
    Dim answer As Integer = 42
    Dim result As String = Await Task.Run(Function()
                                              ' … do compute-bound work here
                                              Return answer.ToString()
                                          End Function)
    Console.WriteLine(result)
End Function

これらの Run メソッドの一部 ( Task.Run(Func<Task>) オーバーロードなど) は、 TaskFactory.StartNew メソッドの短縮形として存在します。このオーバーロードにより、オフロードされた作業内で await を使用できます。例えば次が挙げられます。

public static async Task TaskRunAsyncExample()
{
    Bitmap image = await Task.Run(async () =>
    {
        using Bitmap bmp1 = await Stubs.DownloadFirstImageAsync();
        using Bitmap bmp2 = await Stubs.DownloadSecondImageAsync();
        return Stubs.Mashup(bmp1, bmp2);
    });
}

Public Async Function TaskRunAsyncExample() As Task
    Dim image As Bitmap = Await Task.Run(Async Function()
                                             Using bmp1 As Bitmap = Await Stubs.DownloadFirstImageAsync()
                                                 Using bmp2 As Bitmap = Await Stubs.DownloadSecondImageAsync()
                                                     Return Stubs.Mashup(bmp1, bmp2)
                                                 End Using
                                             End Using
                                         End Function)
End Function

このようなオーバーロードは、タスク並列ライブラリのTaskFactory.StartNew拡張メソッドと組み合わせてUnwrap メソッドを使用することと論理的に同等です。

Task.FromResult

データが既に利用でき、単に FromResult にリフトされたタスクを返すメソッドから返す必要があるシナリオでは、Task<TResult> メソッドを使用します。

public static Task<int> GetValueAsync(string key)
{
    int cachedValue;
    return Stubs.TryGetCachedValue(out cachedValue) ?
        Task.FromResult(cachedValue) :
        GetValueAsyncInternal(key);
}

static async Task<int> GetValueAsyncInternal(string key)
{
    await Task.Delay(1);
    return 0;
}

Public Function GetValueAsync(key As String) As Task(Of Integer)
    Dim cachedValue As Integer
    If Stubs.TryGetCachedValue(cachedValue) Then
        Return Task.FromResult(cachedValue)
    Else
        Return GetValueAsyncInternal(key)
    End If
End Function

Private Async Function GetValueAsyncInternal(key As String) As Task(Of Integer)
    Await Task.Delay(1)
    Return 0
End Function

Task.WhenAll

WhenAll メソッドを使用して、タスクとして表される複数の非同期操作を非同期的に待機します。このメソッドには、一連の非ジェネリックタスクまたは一様でない一連の汎用タスクをサポートする複数のオーバーロードがあります (たとえば、複数の void を返す操作を非同期的に待機したり、各値が異なる可能性がある複数の値を返すメソッドを非同期的に待機したりします)、一様な一連のジェネリックタスク (複数の TResultを返すメソッドを非同期的に待機するなど) をサポートします。

複数の顧客に電子メールメッセージを送信するとします。 1 つのメッセージが完了するのを待ってから次のメッセージを送信しないように、メッセージの送信を重複させることができます。送信操作が完了したタイミングと、エラーが発生したかどうかを確認することもできます。

IEnumerable<Task> asyncOps = from addr in addrs select SendMailAsync(addr);
await Task.WhenAll(asyncOps);

このコードは発生する可能性がある例外を明示的に処理せず、await の結果のタスクの WhenAll から例外を反映します。例外を処理するには、次のようなコードを使用します。

public static async Task WhenAllWithCatch()
{
    IEnumerable<Task> asyncOps = from addr in Stubs.addrs select Stubs.SendMailAsync(addr);
    try
    {
        await Task.WhenAll(asyncOps);
    }
    catch (Exception exc)
    {
        Console.WriteLine(exc);
    }
}

Public Async Function WhenAllWithCatch() As Task
    Dim asyncOps As IEnumerable(Of Task) = From addr In Stubs.addrs Select Stubs.SendMailAsync(addr)
    Try
        Await Task.WhenAll(asyncOps)
    Catch exc As Exception
        Console.WriteLine(exc)
    End Try
End Function

この場合、非同期操作が失敗した場合、すべての例外がAggregateException例外に統合されます。この例外は、Task メソッドから返されるWhenAllに格納されます。ただし、 await キーワードによって反映されるのは、これらの例外の 1 つだけです。すべての例外を調べる場合は、前のコードを次のように書き直すことができます。

public static async Task WhenAllExamineExceptions()
{
    Task[] asyncOps = (from addr in Stubs.addrs select Stubs.SendMailAsync(addr)).ToArray();
    try
    {
        await Task.WhenAll(asyncOps);
    }
    catch (Exception exc)
    {
        foreach (Task faulted in asyncOps.Where(t => t.IsFaulted))
        {
            Console.WriteLine($"Faulted: {faulted.Exception}");
        }
    }
}

Public Async Function WhenAllExamineExceptions() As Task
    Dim asyncOps As Task() = (From addr In Stubs.addrs Select Stubs.SendMailAsync(addr)).ToArray()
    Try
        Await Task.WhenAll(asyncOps)
    Catch exc As Exception
        For Each faulted As Task In asyncOps.Where(Function(t) t.IsFaulted)
            Console.WriteLine($"Faulted: {faulted.Exception}")
        Next
    End Try
End Function

Web から複数のファイルを非同期にダウンロードする例を考えてみましょう。この場合、すべての非同期操作には同種の結果の種類があり、結果に簡単にアクセスできます。

string [] pages = await Task.WhenAll(
    from url in urls select DownloadStringTaskAsync(url));

前の void を返すシナリオで説明したのと同じ例外処理手法を使用できます。

public static async Task WhenAllDownloadPagesExceptions()
{
    Task<string>[] asyncOps =
        (from url in Stubs.urls select Stubs.DownloadStringTaskAsync(url)).ToArray();
    try
    {
        string[] pages = await Task.WhenAll(asyncOps);
        Console.WriteLine(pages.Length);
    }
    catch (Exception exc)
    {
        foreach (Task<string> faulted in asyncOps.Where(t => t.IsFaulted))
        {
            Console.WriteLine($"Faulted: {faulted.Exception}");
        }
    }
}

Public Async Function WhenAllDownloadPagesExceptions() As Task
    Dim asyncOps As Task(Of String)() =
        (From url In Stubs.urls Select Stubs.DownloadStringTaskAsync(url)).ToArray()
    Try
        Dim pages As String() = Await Task.WhenAll(asyncOps)
        Console.WriteLine(pages.Length)
    Catch exc As Exception
        For Each faulted As Task(Of String) In asyncOps.Where(Function(t) t.IsFaulted)
            Console.WriteLine($"Faulted: {faulted.Exception}")
        Next
    End Try
End Function

Task.WhenAny

WhenAny メソッドを使用して、タスクとして表される複数の非同期操作のいずれかが完了するまで非同期的に待機します。このメソッドは、次の 4 つの主要なユースケースに対応します。

冗長性: 操作を複数回実行し、最初に完了した操作を選択します (たとえば、1 つの結果を返す複数の株価情報 Web サービスに連絡し、最も速く完了するものを選択するなど)。
インターリーブ: 複数の操作を起動してそのすべてが完了するのを待機するが、それらの操作を完了時に処理する。
スロットリング: 他の操作が完了すると、追加の操作が開始されるように許可します。このシナリオは、インターリーブシナリオの拡張機能です。
早期の救済: たとえば、タスク t1 で表される操作は、別のタスク t2 を持つ WhenAny タスクにグループ化でき、 WhenAny タスクを待機できます。タスク t2 は、タイムアウト、取り消し、または t1 が完了する前に WhenAny タスクが完了する原因となるその他のシグナルを表している可能性があります。

冗長性

株式を買うかどうかを決定する場合を考えてみましょう。信頼できるストックレコメンデーション Web サービスがいくつか存在しますが、毎日の負荷によっては、各サービスが異なる時間に遅くなる可能性があります。 WhenAnyメソッドを使用して、操作が完了したときに通知を受け取ります。

public static async Task WhenAnyRedundancy(string symbol)
{
    var recommendations = new List<Task<bool>>()
    {
        Stubs.GetBuyRecommendation1Async(symbol),
        Stubs.GetBuyRecommendation2Async(symbol),
        Stubs.GetBuyRecommendation3Async(symbol)
    };
    Task<bool> recommendation = await Task.WhenAny(recommendations);
    if (await recommendation) Stubs.BuyStock(symbol);
}

Public Async Function WhenAnyRedundancy(symbol As String) As Task
    Dim recommendations As New List(Of Task(Of Boolean)) From {
        Stubs.GetBuyRecommendation1Async(symbol),
        Stubs.GetBuyRecommendation2Async(symbol),
        Stubs.GetBuyRecommendation3Async(symbol)
    }
    Dim recommendation As Task(Of Boolean) = Await Task.WhenAny(recommendations)
    If Await recommendation Then Stubs.BuyStock(symbol)
End Function

正常に完了したすべてのタスクのラップされていない結果を返す WhenAllとは異なり、 WhenAny は完了したタスクを返します。タスクが失敗した場合は、失敗したことを把握することが重要であり、タスクが成功した場合は、戻り値が関連付けられているタスクを把握することが重要です。そのため、この例に示すように、返されたタスクの結果にアクセスするか、さらに待機する必要があります。

WhenAllと同様に、例外に対応できる必要があります。完了したタスクを受け取るので、適切に try/catch で囲んで、エラーが反映されて返されるタスクを待機できます。次に例を示します。

public static async Task WhenAnyRetryOnException(string symbol)
{
    Task<bool>[] allRecommendations = new Task<bool>[]
    {
        Stubs.GetBuyRecommendation1Async(symbol),
        Stubs.GetBuyRecommendation2Async(symbol),
        Stubs.GetBuyRecommendation3Async(symbol)
    };
    var remaining = allRecommendations.ToList();
    while (remaining.Count > 0)
    {
        Task<bool> recommendation = await Task.WhenAny(remaining);
        try
        {
            if (await recommendation) Stubs.BuyStock(symbol);
            break;
        }
        catch (WebException)
        {
            remaining.Remove(recommendation);
        }
    }
}

Public Async Function WhenAnyRetryOnException(symbol As String) As Task
    Dim allRecommendations As Task(Of Boolean)() = {
        Stubs.GetBuyRecommendation1Async(symbol),
        Stubs.GetBuyRecommendation2Async(symbol),
        Stubs.GetBuyRecommendation3Async(symbol)
    }
    Dim remaining As List(Of Task(Of Boolean)) = allRecommendations.ToList()
    While remaining.Count > 0
        Dim recommendation As Task(Of Boolean) = Await Task.WhenAny(remaining)
        Try
            If Await recommendation Then Stubs.BuyStock(symbol)
            Exit While
        Catch ex As WebException
            remaining.Remove(recommendation)
        End Try
    End While
End Function

さらに、最初のタスクが正常に完了した場合でも、後続のタスクが失敗する可能性があります。この時点で、例外を処理するためのオプションがいくつかあります。起動されたすべてのタスクが完了するまで待機できます。その場合は、 WhenAll メソッドを使用することも、すべての例外が重要であり、ログに記録する必要があると判断することもできます。このシナリオでは、タスクが非同期的に完了したときに継続を使用して通知を受け取ることができます。

foreach(Task recommendation in recommendations)
{
    var ignored = recommendation.ContinueWith(
        t => { if (t.IsFaulted) Log(t.Exception); });
}

または:

foreach(Task recommendation in recommendations)
{
    var ignored = recommendation.ContinueWith(
        t => Log(t.Exception), TaskContinuationOptions.OnlyOnFaulted);
}

または、以下のように指定できます。

private static async void LogCompletionIfFailed(IEnumerable<Task> tasks)
{
    foreach (var task in tasks)
    {
        try { await task; }
        catch (Exception exc) { Stubs.Log(exc); }
    }
}

Private Async Sub LogCompletionIfFailed(tasks As IEnumerable(Of Task))
    For Each task In tasks
        Try
            Await task
        Catch exc As Exception
            Stubs.Log(exc)
        End Try
    Next
End Sub

最後に、残りのすべての操作を取り消したい場合があります。

public static async Task WhenAnyCancelRemainder(string symbol)
{
    var cts = new CancellationTokenSource();
    var recommendations = new List<Task<bool>>()
    {
        Stubs.GetBuyRecommendation1Async(symbol, cts.Token),
        Stubs.GetBuyRecommendation2Async(symbol, cts.Token),
        Stubs.GetBuyRecommendation3Async(symbol, cts.Token)
    };

    Task<bool> recommendation = await Task.WhenAny(recommendations);
    cts.Cancel();
    if (await recommendation) Stubs.BuyStock(symbol);
}

Public Async Function WhenAnyCancelRemainder(symbol As String) As Task
    Dim cts As New CancellationTokenSource()
    Dim recommendations As New List(Of Task(Of Boolean)) From {
        Stubs.GetBuyRecommendation1Async(symbol, cts.Token),
        Stubs.GetBuyRecommendation2Async(symbol, cts.Token),
        Stubs.GetBuyRecommendation3Async(symbol, cts.Token)
    }

    Dim recommendation As Task(Of Boolean) = Await Task.WhenAny(recommendations)
    cts.Cancel()
    If Await recommendation Then Stubs.BuyStock(symbol)
End Function

インターリーブ

Web からイメージをダウンロードし、各イメージを処理する場合を考えてみましょう (たとえば、UI コントロールに画像を追加する)。 UI スレッドでイメージを順番に処理しますが、可能な限り同時にイメージをダウンロードする必要があります。また、イメージがすべてダウンロードされるまで、UI へのイメージの追加を保留する必要はありません。代わりに、それらを完了時に追加したいと考えています。

public static async Task WhenAnyInterleaving(string[] imageUrls)
{
    List<Task<Bitmap>> imageTasks =
        (from imageUrl in imageUrls select Stubs.GetBitmapAsync(imageUrl)).ToList();
    while (imageTasks.Count > 0)
    {
        try
        {
            Task<Bitmap> imageTask = await Task.WhenAny(imageTasks);
            imageTasks.Remove(imageTask);

            Bitmap image = await imageTask;
            Console.WriteLine($"Got image: {image.Width}x{image.Height}");
        }
        catch { }
    }
}

Public Async Function WhenAnyInterleaving(imageUrls As String()) As Task
    Dim imageTasks As List(Of Task(Of Bitmap)) =
        (From imageUrl In imageUrls Select Stubs.GetBitmapAsync(imageUrl)).ToList()
    While imageTasks.Count > 0
        Try
            Dim imageTask As Task(Of Bitmap) = Await Task.WhenAny(imageTasks)
            imageTasks.Remove(imageTask)

            Dim image As Bitmap = Await imageTask
            Console.WriteLine($"Got image: {image.Width}x{image.Height}")
        Catch
        End Try
    End While
End Function

ダウンロードしたイメージの ThreadPool に対して計算負荷の高い処理を伴うシナリオにインターリーブを適用することもできます。次に例を示します。

public static async Task WhenAnyInterleavingWithProcessing(string[] imageUrls)
{
    List<Task<Bitmap>> imageTasks =
        (from imageUrl in imageUrls
         select Stubs.GetBitmapAsync(imageUrl)
             .ContinueWith(t => Stubs.ConvertImage(t.Result))).ToList();
    while (imageTasks.Count > 0)
    {
        try
        {
            Task<Bitmap> imageTask = await Task.WhenAny(imageTasks);
            imageTasks.Remove(imageTask);

            Bitmap image = await imageTask;
            Console.WriteLine($"Got image: {image.Width}x{image.Height}");
        }
        catch { }
    }
}

Public Async Function WhenAnyInterleavingWithProcessing(imageUrls As String()) As Task
    Dim imageTasks As List(Of Task(Of Bitmap)) =
        (From imageUrl In imageUrls
         Select Stubs.GetBitmapAsync(imageUrl).ContinueWith(Function(t) Stubs.ConvertImage(t.Result))).ToList()
    While imageTasks.Count > 0
        Try
            Dim imageTask As Task(Of Bitmap) = Await Task.WhenAny(imageTasks)
            imageTasks.Remove(imageTask)

            Dim image As Bitmap = Await imageTask
            Console.WriteLine($"Got image: {image.Width}x{image.Height}")
        Catch
        End Try
    End While
End Function

Throttling

インターリーブの例を考えてみます。ただし、ユーザーが非常に多くの画像をダウンロードしているため、ダウンロード速度をスロットリングしなければならないという点は例外です。たとえば、特定の数のダウンロードのみを同時に実行する必要があります。この目標を達成するには、非同期操作のサブセットを開始します。操作が完了したら、追加の操作を開始して代わりに実行できます。

public static async Task WhenAnyThrottling(Uri[] uriList)
{
    const int CONCURRENCY_LEVEL = 15;
    int nextIndex = 0;
    var imageTasks = new List<Task<Bitmap>>();
    while (nextIndex < CONCURRENCY_LEVEL && nextIndex < uriList.Length)
    {
        imageTasks.Add(Stubs.GetBitmapAsync(uriList[nextIndex].ToString()));
        nextIndex++;
    }

    while (imageTasks.Count > 0)
    {
        try
        {
            Task<Bitmap> imageTask = await Task.WhenAny(imageTasks);
            imageTasks.Remove(imageTask);

            Bitmap image = await imageTask;
            Console.WriteLine($"Got image: {image.Width}x{image.Height}");
        }
        catch (Exception exc) { Stubs.Log(exc); }

        if (nextIndex < uriList.Length)
        {
            imageTasks.Add(Stubs.GetBitmapAsync(uriList[nextIndex].ToString()));
            nextIndex++;
        }
    }
}

Public Async Function WhenAnyThrottling(uriList As Uri()) As Task
    Const CONCURRENCY_LEVEL As Integer = 15
    Dim nextIndex As Integer = 0
    Dim imageTasks As New List(Of Task(Of Bitmap))
    While nextIndex < CONCURRENCY_LEVEL AndAlso nextIndex < uriList.Length
        imageTasks.Add(Stubs.GetBitmapAsync(uriList(nextIndex).ToString()))
        nextIndex += 1
    End While

    While imageTasks.Count > 0
        Try
            Dim imageTask As Task(Of Bitmap) = Await Task.WhenAny(imageTasks)
            imageTasks.Remove(imageTask)

            Dim image As Bitmap = Await imageTask
            Console.WriteLine($"Got image: {image.Width}x{image.Height}")
        Catch exc As Exception
            Stubs.Log(exc)
        End Try

        If nextIndex < uriList.Length Then
            imageTasks.Add(Stubs.GetBitmapAsync(uriList(nextIndex).ToString()))
            nextIndex += 1
        End If
    End While
End Function

初期のエスケープ

ユーザーの取り消し要求 (たとえば、ユーザーがキャンセルボタンをクリックした場合など) に同時に応答しながら、操作が完了するまで非同期的に待機しているとします。次のコードは、このシナリオを示しています。

class EarlyBailoutUI
{
    private CancellationTokenSource? m_cts;

    public void btnCancel_Click(object sender, EventArgs e)
    {
        if (m_cts != null) m_cts.Cancel();
    }

    public async void btnRun_Click(object sender, EventArgs e)
    {
        m_cts = new CancellationTokenSource();
        try
        {
            Task<Bitmap> imageDownload = Stubs.GetBitmapAsync("url");
            await Examples.UntilCompletionOrCancellation(imageDownload, m_cts.Token);
            if (imageDownload.IsCompleted)
            {
                Bitmap image = await imageDownload;
                Stubs.Log(image);
            }
            else imageDownload.ContinueWith(t => Stubs.Log(t));
        }
        finally { }
    }
}

Class EarlyBailoutUI
    Private m_cts As CancellationTokenSource

    Public Sub btnCancel_Click(sender As Object, e As EventArgs)
        If m_cts IsNot Nothing Then m_cts.Cancel()
    End Sub

    Public Async Sub btnRun_Click(sender As Object, e As EventArgs)
        m_cts = New CancellationTokenSource()
        Try
            Dim imageDownload As Task(Of Bitmap) = Stubs.GetBitmapAsync("url")
            Await Examples.UntilCompletionOrCancellation(imageDownload, m_cts.Token)
            If imageDownload.IsCompleted Then
                Dim image As Bitmap = Await imageDownload
                Stubs.Log(image)
            Else
                imageDownload.ContinueWith(Sub(t) Stubs.Log(t))
            End If
        Finally
        End Try
    End Sub
End Class

ユーザーがキャンセルを選択したらすぐに、ユーザーインターフェイスは再度有効になります。ただし、基になる非同期操作は取り消されません。別の方法として、キャンセルしようと決めた場合に、保留中の操作を取り消すことが考えられます。ただし、その際、キャンセル要求により操作が早期終了する可能性があるので、操作が完了するまではユーザーインターフェースを復元しないことも検討すべきです。

class EarlyBailoutWithTokenUI
{
    private CancellationTokenSource? m_cts;

    public async void btnRun_Click(object sender, EventArgs e)
    {
        m_cts = new CancellationTokenSource();
        try
        {
            Task<Bitmap> imageDownload = Stubs.GetBitmapAsync("url", m_cts.Token);
            await Examples.UntilCompletionOrCancellation(imageDownload, m_cts.Token);
            Bitmap image = await imageDownload;
            Stubs.Log(image);
        }
        catch (OperationCanceledException) { }
        finally { }
    }
}

Class EarlyBailoutWithTokenUI
    Private m_cts As CancellationTokenSource

    Public Async Sub btnRun_Click(sender As Object, e As EventArgs)
        m_cts = New CancellationTokenSource()
        Try
            Dim imageDownload As Task(Of Bitmap) = Stubs.GetBitmapAsync("url", m_cts.Token)
            Await Examples.UntilCompletionOrCancellation(imageDownload, m_cts.Token)
            Dim image As Bitmap = Await imageDownload
            Stubs.Log(image)
        Catch ex As OperationCanceledException
        Finally
        End Try
    End Sub
End Class

早期救済のもう 1 つの例として、次のセクションで説明するように、 WhenAny メソッドを Delay メソッドと組み合わせて使用する必要があります。

Task.Delay (タスクの遅延)

Task.Delay メソッドを使用して、非同期メソッドの実行に一時停止を追加します。この一時停止は、ポーリングループの構築や、ユーザー入力の処理の遅延など、さまざまな種類の機能に役立ちます。 Task.Delay メソッドをTask.WhenAnyと共に使用して、awaits にタイムアウトを実装することもできます。

大規模な非同期操作 (たとえば、ASP.NET Web サービス) の一部であるタスクの完了に時間がかかりすぎる場合は、操作全体が低下する可能性があります(特に、完了に失敗した場合)。このため、非同期操作を待機しているときにタイムアウトできることが重要です。同期 Task.Wait、 Task.WaitAll、および Task.WaitAny のメソッドはタイムアウト値を受け取りますが、対応する TaskFactory.ContinueWhenAll/TaskFactory.ContinueWhenAny と前述の Task.WhenAll/Task.WhenAny メソッドでは受け入れられません。代わりに、 Task.Delay と Task.WhenAny を一緒に使用してタイムアウトを実装します。

たとえば、UI アプリケーションで、イメージをダウンロードし、イメージのダウンロード中に UI を無効にするとします。ただし、ダウンロードに時間がかかりすぎる場合は、UI を再度有効にしてダウンロードを破棄する必要があります。

public static async Task<Bitmap?> DownloadWithTimeout(string url)
{
    Task<Bitmap> download = Stubs.GetBitmapAsync(url);
    if (download == await Task.WhenAny(download, Task.Delay(3000)))
    {
        return await download;
    }
    else
    {
        var ignored = download.ContinueWith(
            t => Trace($"Task finally completed: {t.Status}"));
        return null;
    }
}

static void Trace(string message) => Console.WriteLine(message);

Public Async Function DownloadWithTimeout(url As String) As Task(Of Bitmap)
    Dim download As Task(Of Bitmap) = Stubs.GetBitmapAsync(url)
    If download Is Await Task.WhenAny(download, Task.Delay(3000)) Then
        Return Await download
    Else
        Dim ignored = download.ContinueWith(Sub(t) TraceMsg($"Task finally completed: {t.Status}"))
        Return Nothing
    End If
End Function

WhenAllはタスクを返すので、同じ原則が複数のダウンロードに適用されます。

public static async Task<Bitmap[]?> DownloadMultipleWithTimeout(string[] imageUrls)
{
    Task<Bitmap[]> downloads =
        Task.WhenAll(from url in imageUrls select Stubs.GetBitmapAsync(url));
    if (downloads == await Task.WhenAny(downloads, Task.Delay(3000)))
    {
        return await downloads;
    }
    else
    {
        downloads.ContinueWith(t => Stubs.Log(t));
        return null;
    }
}

Public Async Function DownloadMultipleWithTimeout(imageUrls As String()) As Task(Of Bitmap())
    Dim downloads As Task(Of Bitmap()) =
        Task.WhenAll(From url In imageUrls Select Stubs.GetBitmapAsync(url))
    If downloads Is Await Task.WhenAny(downloads, Task.Delay(3000)) Then
        Return Await downloads
    Else
        downloads.ContinueWith(Sub(t) Stubs.Log(t))
        Return Nothing
    End If
End Function

タスクベースの組み合わせ子の構築

タスクは完全に非同期操作を表し、操作との結合、結果の取得などの同期機能と非同期機能を提供できるため、タスクを構成する組み合わせ子の便利なライブラリを構築して、より大きなパターンを構築できます。前のセクションで説明したように、.NET には複数の組み込みコンバイネータが含まれていますが、独自にビルドすることもできます。次のセクションでは、組み合わせ子のメソッドと型の例をいくつか示します。

RetryOnFault

多くの場合、以前の試行が失敗した場合に操作を再試行する必要があります。同期コードの場合は、次の例の RetryOnFault などのヘルパーメソッドをビルドして、このタスクを実行できます。

public static T RetryOnFault<T>(Func<T> function, int maxTries)
{
    for (int i = 0; i < maxTries; i++)
    {
        try { return function(); }
        catch { if (i == maxTries - 1) throw; }
    }
    return default(T)!;
}

Public Function RetryOnFaultSync(Of T)(func As Func(Of T), maxTries As Integer) As T
    For i As Integer = 0 To maxTries - 1
        Try
            Return func()
        Catch
            If i = maxTries - 1 Then Throw
        End Try
    Next
    Return Nothing
End Function

TAP を使用して実装され、タスクを返す非同期操作用に、ほぼ同じヘルパーメソッドを構築できます。

public static async Task<T> RetryOnFault<T>(Func<Task<T>> function, int maxTries)
{
    for (int i = 0; i < maxTries; i++)
    {
        try { return await function().ConfigureAwait(false); }
        catch { if (i == maxTries - 1) throw; }
    }
    return default(T)!;
}

Public Async Function RetryOnFault(Of T)(func As Func(Of Task(Of T)), maxTries As Integer) As Task(Of T)
    For i As Integer = 0 To maxTries - 1
        Try
            Return Await func().ConfigureAwait(False)
        Catch
            If i = maxTries - 1 Then Throw
        End Try
    Next
    Return Nothing
End Function

その後、このコンバイネーターを使用して、再試行をアプリケーションのロジックにエンコードできます。例えば次が挙げられます。

// Download the URL, trying up to three times in case of failure
string pageContents = await RetryOnFault(
    () => DownloadStringTaskAsync(url), 3);

RetryOnFault関数をさらに拡張できます。たとえば、関数は再試行の間に呼び出す別の Func<Task> を受け入れて、操作を再試行するタイミングを決定できます。例えば次が挙げられます。

public static async Task<T> RetryOnFaultWithDelay<T>(
    Func<Task<T>> function, int maxTries, Func<Task> retryWhen)
{
    for (int i = 0; i < maxTries; i++)
    {
        try { return await function().ConfigureAwait(false); }
        catch { if (i == maxTries - 1) throw; }
        await retryWhen().ConfigureAwait(false);
    }
    return default(T)!;
}

Public Async Function RetryOnFaultWithDelay(Of T)(
    func As Func(Of Task(Of T)), maxTries As Integer, retryWhen As Func(Of Task)) As Task(Of T)
    For i As Integer = 0 To maxTries - 1
        Try
            Return Await func().ConfigureAwait(False)
        Catch
            If i = maxTries - 1 Then Throw
        End Try
        Await retryWhen().ConfigureAwait(False)
    Next
    Return Nothing
End Function

その後、次のように関数を使用して、操作を再試行する前に 1 秒間待機できます。

// Download the URL, trying up to three times in case of failure,
// and delaying for a second between retries
string pageContents = await RetryOnFault(
    () => DownloadStringTaskAsync(url), 3, () => Task.Delay(1000));

NeedOnlyOne

場合によっては、冗長性を利用して、操作の待機時間と成功の可能性を向上させることができます。株価情報を提供する複数の Web サービスを検討しますが、1 日のさまざまな時間帯に、各サービスによって異なるレベルの品質と応答時間が提供される場合があります。これらの変動に対処するために、すべての Web サービスに要求を発行し、応答が返されたらすぐに残りの要求を取り消します。ヘルパー関数を実装すると、複数の操作を起動し、任意の操作を待機してから、残りの操作を取り消すという一般的なパターンを簡単に実装できます。次の例の NeedOnlyOne 関数は、このシナリオを示しています。

public static async Task<T> NeedOnlyOne<T>(
    params Func<CancellationToken, Task<T>>[] functions)
{
    var cts = new CancellationTokenSource();
    var tasks = (from function in functions
                 select function(cts.Token)).ToArray();
    var completed = await Task.WhenAny(tasks).ConfigureAwait(false);
    cts.Cancel();
    foreach (var task in tasks)
    {
        var ignored = task.ContinueWith(
            t => Stubs.Log(t), TaskContinuationOptions.OnlyOnFaulted);
    }
    return await completed;
}

Public Async Function NeedOnlyOne(Of T)(
    ParamArray functions As Func(Of CancellationToken, Task(Of T))()) As Task(Of T)
    Dim cts As New CancellationTokenSource()
    Dim tasks As Task(Of T)() = (From func In functions Select func(cts.Token)).ToArray()
    Dim completed As Task(Of T) = Await Task.WhenAny(tasks).ConfigureAwait(False)
    cts.Cancel()
    For Each task In tasks
        Dim ignored = task.ContinueWith(
            Sub(tsk) Stubs.Log(tsk), TaskContinuationOptions.OnlyOnFaulted)
    Next
    Return Await completed
End Function

その後、次のようにこの関数を使用できます。

double currentPrice = await NeedOnlyOne(
    ct => GetCurrentPriceFromServer1Async("msft", ct),
    ct => GetCurrentPriceFromServer2Async("msft", ct),
    ct => GetCurrentPriceFromServer3Async("msft", ct));

インターリーブされた操作

WhenAnyメソッドを使用してインターリーブシナリオをサポートすると、大規模なタスクセットを操作するときにパフォーマンスの問題が発生する可能性があります。 WhenAny を呼び出すために、各タスクの継続処理が登録されます。タスクが、N 数の場合、このプロセスは、インターリーブ操作の実行期間全体で O(N²) の継続処理を作成します。多数のタスクセットを使用している場合は、コンバイネータ (次の例のInterleaved ) を使用してパフォーマンスの問題に対処します。

public static IEnumerable<Task<T>> Interleaved<T>(IEnumerable<Task<T>> tasks)
{
    var inputTasks = tasks.ToList();
    var sources = (from _ in Enumerable.Range(0, inputTasks.Count)
                   select new TaskCompletionSource<T>()).ToList();
    int nextTaskIndex = -1;
    foreach (var inputTask in inputTasks)
    {
        inputTask.ContinueWith(completed =>
        {
            var source = sources[Interlocked.Increment(ref nextTaskIndex)];
            if (completed.IsFaulted)
                source.TrySetException(completed.Exception!.InnerExceptions);
            else if (completed.IsCanceled)
                source.TrySetCanceled();
            else
                source.TrySetResult(completed.Result);
        }, CancellationToken.None,
           TaskContinuationOptions.ExecuteSynchronously,
           TaskScheduler.Default);
    }
    return from source in sources
           select source.Task;
}

Public Function Interleaved(Of T)(tasks As IEnumerable(Of Task(Of T))) As IEnumerable(Of Task(Of T))
    Dim inputTasks As List(Of Task(Of T)) = tasks.ToList()
    Dim sources As List(Of TaskCompletionSource(Of T)) =
        (From _i In Enumerable.Range(0, inputTasks.Count) Select New TaskCompletionSource(Of T)()).ToList()
    Dim indexRef As Integer() = {-1}
    For Each inputTask In inputTasks
        inputTask.ContinueWith(Sub(completed)
                                   Dim idx = Interlocked.Increment(indexRef(0))
                                   Dim source = sources(idx)
                                   If completed.IsFaulted Then
                                       source.TrySetException(completed.Exception.InnerExceptions)
                                   ElseIf completed.IsCanceled Then
                                       source.TrySetCanceled()
                                   Else
                                       source.TrySetResult(completed.Result)
                                   End If
                               End Sub,
                               CancellationToken.None,
                               TaskContinuationOptions.ExecuteSynchronously,
                               TaskScheduler.Default)
    Next
    Return From source In sources Select source.Task
End Function

コンバイネーターを使用して、タスクの完了時に結果を処理します。例えば次が挙げられます。

IEnumerable<Task<int>> tasks = ...;
foreach(var task in Interleaved(tasks))
{
    int result = await task;
    …
}

WhenAllOrFirstException

特定のスキャッター/ギャザーシナリオでは、そのうちの 1 つにエラーが発生しない限り、セット内のすべてのタスクを待機する必要がある場合があります。その場合は、例外が発生したらすぐに待機を停止します。この動作は、次の例の WhenAllOrFirstException などのコンバイネータメソッドを使用して実現できます。

public static Task<T[]> WhenAllOrFirstException<T>(IEnumerable<Task<T>> tasks)
{
    var inputs = tasks.ToList();
    var ce = new CountdownEvent(inputs.Count);
    var tcs = new TaskCompletionSource<T[]>();

    Action<Task> onCompleted = (Task completed) =>
    {
        if (completed.IsFaulted)
            tcs.TrySetException(completed.Exception!.InnerExceptions);
        if (ce.Signal() && !tcs.Task.IsCompleted)
            tcs.TrySetResult(inputs.Select(t => ((Task<T>)t).Result).ToArray());
    };

    foreach (var t in inputs) t.ContinueWith(onCompleted);
    return tcs.Task;
}

Public Function WhenAllOrFirstException(Of T)(tasks As IEnumerable(Of Task(Of T))) As Task(Of T())
    Dim inputs As List(Of Task(Of T)) = tasks.ToList()
    Dim ce As New CountdownEvent(inputs.Count)
    Dim tcs As New TaskCompletionSource(Of T())()

    Dim onCompleted As Action(Of Task) = Sub(completed As Task)
                                             If completed.IsFaulted Then
                                                 tcs.TrySetException(completed.Exception.InnerExceptions)
                                             End If
                                             If ce.Signal() AndAlso Not tcs.Task.IsCompleted Then
                                                 tcs.TrySetResult(inputs.Select(Function(taskItem) DirectCast(taskItem, Task(Of T)).Result).ToArray())
                                             End If
                                         End Sub

    For Each t In inputs
        t.ContinueWith(onCompleted)
    Next
    Return tcs.Task
End Function

タスクベースのデータ構造の構築

カスタムタスクベースの組み合わせ子を構築する機能に加えて、非同期操作の結果と、それに結合するために必要な同期の両方を表すデータ構造を Task と Task<TResult> することで、非同期シナリオで使用するカスタムデータ構造を構築するための強力な型になります。

AsyncCache

タスクの重要な側面の 1 つは、複数のコンシューマーに渡すことができるということです。すべてのコンシューマーは、それを待機したり、継続処理を登録したり、結果または例外 ( Task<TResult> の場合) を取得したりできます。この側面により、 Task と Task<TResult> 非同期キャッシュインフラストラクチャでの使用に完全に適しています。 Task<TResult>に基づいて構築された、小さくて強力な非同期キャッシュの例を次に示します。

public class AsyncCache<TKey, TValue> where TKey : notnull
{
    private readonly Func<TKey, Task<TValue>> _valueFactory;
    private readonly ConcurrentDictionary<TKey, Lazy<Task<TValue>>> _map;

    public AsyncCache(Func<TKey, Task<TValue>> valueFactory)
    {
        if (valueFactory == null) throw new ArgumentNullException(nameof(valueFactory));
        _valueFactory = valueFactory;
        _map = new ConcurrentDictionary<TKey, Lazy<Task<TValue>>>();
    }

    public Task<TValue> this[TKey key]
    {
        get
        {
            if (key == null) throw new ArgumentNullException(nameof(key));
            return _map.GetOrAdd(key, toAdd =>
                new Lazy<Task<TValue>>(() => _valueFactory(toAdd))).Value;
        }
    }
}

Public Class AsyncCache(Of TKey, TValue)
    Private ReadOnly _valueFactory As Func(Of TKey, Task(Of TValue))
    Private ReadOnly _map As New ConcurrentDictionary(Of TKey, Lazy(Of Task(Of TValue)))()

    Public Sub New(valueFactory As Func(Of TKey, Task(Of TValue)))
        If valueFactory Is Nothing Then Throw New ArgumentNullException(NameOf(valueFactory))
        _valueFactory = valueFactory
    End Sub

    Default Public ReadOnly Property Item(key As TKey) As Task(Of TValue)
        Get
            If key Is Nothing Then Throw New ArgumentNullException(NameOf(key))
            Return _map.GetOrAdd(key, Function(toAdd) New Lazy(Of Task(Of TValue))(Function() _valueFactory(toAdd))).Value
        End Get
    End Property
End Class

AsyncCache<TKey,TValue> クラスは、TKeyを受け取り、Task<TResult>を返す関数をコンストラクターへのデリゲートとして受け取ります。内部ディクショナリは、キャッシュから以前にアクセスした値を格納し、 AsyncCache は、キャッシュが同時にアクセスされている場合でも、キーごとに 1 つのタスクのみを生成することを保証します。

たとえば、ダウンロードした Web ページのキャッシュを作成できます。

private AsyncCache<string,string> m_webPages =
    new AsyncCache<string,string>(DownloadStringTaskAsync);

その後、Web ページの内容が必要な場合は常に、非同期メソッドでこのキャッシュを使用できます。 AsyncCache クラスを使用すると、できるだけ少ないページをダウンロードし、結果をキャッシュできます。

static AsyncCache<string, string> m_webPages =
    new AsyncCache<string, string>(url => Stubs.DownloadStringTaskAsync(url));

public static async Task UseWebPageCache(string url)
{
    string contents = await m_webPages[url];
    Console.WriteLine(contents.Length);
}

Private m_webPages As New AsyncCache(Of String, String)(Function(url) Stubs.DownloadStringTaskAsync(url))

Public Async Function UseWebPageCache(url As String) As Task
    Dim contents As String = Await m_webPages(url)
    Console.WriteLine(contents.Length)
End Function

AsyncProducerConsumerCollection

また、タスクを使用して、非同期アクティビティを調整するためのデータ構造を構築することもできます。従来の並列設計パターンの 1 つであるプロデューサー/コンシューマーについて考えてみましょう。このパターンでは、プロデューサーはコンシューマーが使用するデータを生成し、プロデューサーとコンシューマーは並列で実行できます。たとえば、コンシューマーはアイテム 1 を処理します。これは、アイテム 2 を現在生産しているプロデューサーによって以前に生成されました。プロデューサー/コンシューマーパターンでは、プロデューサーによって作成された作業を格納して、コンシューマーに新しいデータの通知を受け取り、使用可能な場合にそれを見つけられるように、常にいくつかのデータ構造が必要です。

非同期メソッドをプロデューサーおよびコンシューマーとして使用できるようにする、タスクに基づいて構築された単純なデータ構造を次に示します。

public class AsyncProducerConsumerCollection<T>
{
    private readonly Queue<T> m_collection = new Queue<T>();
    private readonly Queue<TaskCompletionSource<T>> m_waiting =
        new Queue<TaskCompletionSource<T>>();

    public void Add(T item)
    {
        TaskCompletionSource<T>? tcs = null;
        lock (m_collection)
        {
            if (m_waiting.Count > 0) tcs = m_waiting.Dequeue();
            else m_collection.Enqueue(item);
        }
        if (tcs != null) tcs.TrySetResult(item);
    }

    public Task<T> Take()
    {
        lock (m_collection)
        {
            if (m_collection.Count > 0)
            {
                return Task.FromResult(m_collection.Dequeue());
            }
            else
            {
                var tcs = new TaskCompletionSource<T>();
                m_waiting.Enqueue(tcs);
                return tcs.Task;
            }
        }
    }
}

Public Class AsyncProducerConsumerCollection(Of T)
    Private ReadOnly m_collection As New Queue(Of T)()
    Private ReadOnly m_waiting As New Queue(Of TaskCompletionSource(Of T))()

    Public Sub Add(item As T)
        Dim tcs As TaskCompletionSource(Of T) = Nothing
        SyncLock m_collection
            If m_waiting.Count > 0 Then
                tcs = m_waiting.Dequeue()
            Else
                m_collection.Enqueue(item)
            End If
        End SyncLock
        If tcs IsNot Nothing Then tcs.TrySetResult(item)
    End Sub

    Public Function Take() As Task(Of T)
        SyncLock m_collection
            If m_collection.Count > 0 Then
                Return Task.FromResult(m_collection.Dequeue())
            Else
                Dim tcs As New TaskCompletionSource(Of T)()
                m_waiting.Enqueue(tcs)
                Return tcs.Task
            End If
        End SyncLock
    End Function
End Class

そのデータ構造を配置すると、次のようなコードを記述できます。

static AsyncProducerConsumerCollection<int> m_data = new();

public static async Task ConsumerAsync()
{
    while (true)
    {
        int nextItem = await m_data.Take();
        Stubs.ProcessNextItem(nextItem);
    }
}

public static void Produce(int data)
{
    m_data.Add(data);
}

Private m_data As New AsyncProducerConsumerCollection(Of Integer)()

Public Async Function ConsumerAsync() As Task
    While True
        Dim nextItem As Integer = Await m_data.Take()
        Stubs.ProcessNextItem(nextItem)
    End While
End Function

Public Sub Produce(data As Integer)
    m_data.Add(data)
End Sub

System.Threading.Tasks.Dataflow名前空間にはBufferBlock<T>型が含まれています。これは同様の方法で使用できますが、カスタムコレクション型を作成する必要はありません。

static BufferBlock<int> m_dataBlock = new();

public static async Task ConsumerAsyncBlock()
{
    while (true)
    {
        int nextItem = await m_dataBlock.ReceiveAsync();
        Stubs.ProcessNextItem(nextItem);
    }
}

public static void ProduceBlock(int data)
{
    m_dataBlock.Post(data);
}

Private m_dataBlock As New BufferBlock(Of Integer)()

Public Async Function ConsumerAsyncBlock() As Task
    While True
        Dim nextItem As Integer = Await m_dataBlock.ReceiveAsync()
        Stubs.ProcessNextItem(nextItem)
    End While
End Function

Public Sub ProduceBlock(data As Integer)
    m_dataBlock.Post(data)
End Sub

注

System.Threading.Tasks.Dataflow名前空間は、NuGet パッケージとして使用できます。 System.Threading.Tasks.Dataflow名前空間を含むアセンブリをインストールするには、Visual Studio でプロジェクトを開き、[プロジェクト] メニューから [NuGet パッケージの管理] を選択し、System.Threading.Tasks.Dataflow パッケージをオンラインで検索します。

こちらも参照ください

フィードバック

このページはお役に立ちましたか?

最終更新: 2026-08-06