c# - 最も最適なTPLデータフロー設計？

Question

TPLデータフローを使用して最適なアーキテクチャを最適に設計する方法について意見を求めたいと思います。私はまだコードを書いていませんので、投稿できるサンプルコードはありません。私は（ボランティアでない限り）コードも探していませんが、設計の支援をいただければ幸いです。

要件は次のとおりです。

特定の方法で相互に依存している3つのコアデータブロックがあります。Datablock1は、タイプFoo1のオブジェクトを生成するプロデューサーです。Datablock2は（Datablock1からの）Foo1オブジェクトをサブスクライブすることになっており、潜在的に（特定の関数の対象となるすべてのFoo1にではなく）、他のデータブロックが消費するために出力キューに格納するFoo2オブジェクトを生成します。Datablock3は（Datablock1からの）Foo1オブジェクトも消費し、Datablock2が消費してFoo2オブジェクトに変換するFoo3オブジェクトを生成する可能性があります。

要約すると、データブロックとそれらがそれぞれ生成および消費するものは次のとおりです。

• Datablock1：Produces（Foo1）、Consumes（Nothing）
• Datablock2：Produces（Foo2）、Consumes（Foo1、Foo3）
• Datablock3：Produces（Foo3）、Consumes（Foo1）

追加の要件は、同じFoo1がDatablock2とDatablock3でほぼ同時に処理されることです。Foo1オブジェクトが最初にDatablock2によって消費され、次にDatablock2がその作業を完了すると、まったく同じFoo1オブジェクトがDatablock3に投稿されてその作業を実行する場合は問題ありません。Datablock2のFoo2オブジェクトは、Foo1オブジェクトまたはFoo3オブジェクトのいずれかの操作の結果として発生する可能性があります。

これが理にかなっていることを願っています。それでも不明な点がある場合は、さらに説明させていただきます。

私の最初のアイデアは、3つのデータブロックごとにTPLデータフローブロックを作成し、それらにさまざまなオブジェクトタイプの着信ストリームを処理させることでした。もう1つのアイデアは、データブロックを分割し、各データブロックが単一のオブジェクトタイプのストリームのみを処理するようにすることです。あなたは何をお勧めしますか、それともうまくいくかもしれないさらに良い解決策がありますか？

SvickはすでにDatablock1を支援しており、すでに運用されています。現在の環境（上記のとおり）をTPLDataflowに変換する方法に固執しています。

任意のアイデアやポインタは大歓迎です。

Answer 1

この問題を3つに分割し、それぞれを個別に解決してみましょう。

1つ目は、条件付きでアイテムを作成する方法です。TransformManyBlock最良のオプションは、関数が1つまたは0のアイテムを含むコレクションを返すようにすることだと思います。

もう1つのオプションは、2つのブロックを条件付きでリンクすることです。これにより、nullsは無視され、null何も生成したくないときに返されます。ただし、これを行う場合は、ソースをにリンクしNullTargetて、nullsが出力バッファに留まらないようにする必要もあります。

2番目の問題は、Foo1をブロック＃2とブロック＃3の両方に送信する方法です。私はここで2つの方法を見ることができます：

1. BroadcastBlock両方のターゲットブロック（＃2と＃3）にリンクして使用します。BroadcastBlock出力キューがないため、これに注意してください。ターゲットブロックがアイテムを延期すると、それはアイテムを処理しないことを意味します。このため、この場合BoundedCapacity、ブロック＃2と＃3を設定しないでください。そうしないと、延期されることはなく、すべてのメッセージが両方のブロックで処理されます。
2. Foo1をブロック＃2で処理した後、手動でPost()（またはより適切にSendAsync()）ブロック＃3に処理します。

「ほぼ同時に」が正確に何を意味するのかはわかりませんが、一般に、TPLDataflowは独立したブロックの処理の順序について保証しません。カスタムTaskSchedulerを使用してさまざまなブロックの優先度を変更できますが、それがここで役立つかどうかはわかりません。

最後の最も複雑な問題は、単一のブロックでさまざまなタイプのアイテムを処理する方法です。これを行うにはいくつかの方法がありますが、どれがあなたに最適かはわかりません。

1. それらを単一のブロックで処理しないでください。TransformBlock<Foo1, Foo2>1つと1つ持ってTransformBlock<Foo3, Foo2>います。次に、両方を1つにリンクできますBufferBlock<Foo2>。
2. 提案したように、1を使用BatchedJoinBlock<Foo1, Foo3>して、を使用しbatchSizeます。これは、結果Tuple<IList<Foo1>, IList<Foo3>>に常に1つFoo1または1つのが含まれることを意味しますFoo3。
3. より適切なタイプを生成するにをリンクすることにより、以前のソリューションを強化しBatchedJoinBlockます。TransformBlockこれは、Tuple<Foo1, Foo3>（アイテムの1つは常にnull）、またはF＃のようなChoice<Foo1, Foo3>もので、2つのうち1つだけが設定されるようにします。
4. 新しいブロックタイプを最初から作成します。これにより、必要な処理が正確に実行されます。組み込みの結合ブロックのように、タイプとタイプのISourceBlock<Foo2>2つのプロパティが必要です。Target1ITarget<Foo1>Target2ITarget<Foo3>

オプション＃1と＃3を使用すると、ブロックを1つのカスタムブロックにカプセル化することもできます。これは、外側から見ると＃4のブロックのように見えるため、より簡単に再利用できます。