問題タブ [task-parallel-library]

BlockingCollectionC# 4.0 でプロデューサー/コンシューマー パターンを実装するために a を使用します。

はBlockingCollection、かなり多くのメモリを消費するアイテムを保持します。プロデューサーが BlockingCollection から一度に 1 つのアイテムを取り出して処理できるようにしたいと考えています。

foreach on を使用するBlockingCollection.GetConsumingEnumerable()たびにBlockingCollection、基礎となるキューからアイテムが削除されるので（つまり、参照と一緒にすべてを意味します）、アイテムを処理する Process() メソッドの最後に、アイテムがガベージになる可能性があると考えていました集めました。

しかし、これは真実ではありません。foreach ループはBlockingCollection.GetConsumingEnumerable()、キューに入力されたアイテムのすべての参照を保持しているようです。foreach ループから出るまで、すべての項目が保持されます (したがって、ガベージ コレクションが防止されます)。

で単純な foreach ループを使用する代わりに、BlockingCollection.GetConsumingEnumerable()BlockingCollection.IsComplete フラグをテストする while ループを使用し、ループ内BlockingCollection.Take()で消耗品を取得します。BlockingCollection からアイテムの参照を削除する とBlockingCollection.Take()同様の効果があると思います。List.Remove()しかし、これもまた間違っています。すべてのアイテムは、while ループの外側でのみガベージ コレクションされます。

私の質問は、BlockingCollection が潜在的にメモリを消費するアイテムを保持し、各アイテムが消費者によって消費されるとガベージ コレクションできるように、要件を簡単に実装するにはどうすればよいかということです。助けてくれてありがとう。

編集: 要求に応じて、簡単なデモ コードが追加されます。

出力は、すべての作成および処理メッセージの後に「処理の終了」が続きます。エンティティからのすべての破壊メッセージが続きます。また、0 から 9 までの Entity.ID を示す作成エンティティ メッセージと、9 から 0 までの Entity.ID を示す破棄メッセージ。

編集：

BlockingCollection のバインドされた容量を設定しても、これまでに入力されたすべてのアイテムは、ループが終了したときにのみ確定されます。これは奇妙です。

次のコードを実行しようとしていますが、配列の値をリストに割り当てようとすると、Index out of range 例外が発生し続けます:-

ここで何か間違っていますか？プロセスが順序付けられていない/非同期であることは理解していますが、「i」が「array.Length」の値よりも大きい値を取得するのはなぜですか?

次のようなタスクを返すメソッドがあります。

タスクへの参照を保持し、後で実行したいと思います。ただし、FromAsyncメソッドによって作成されたタスクはすぐに実行されるようです。どうすれば実行を延期できますか？

タスクがあり、実行に 1 秒もかからないと予想していますが、数秒以上かかる場合は、タスクをキャンセルしたいと考えています。

例えば：

3000 ミリ秒後に待機が期限切れになることに注意してください。タイムアウトの期限が切れたときにタスクがキャンセルされたか、それともタスクがまだ実行中か?

TPLを把握しようとしています。

楽しみのために、ランダムなスリープでいくつかのタスクを作成して、それがどのように処理されるかを確認してみました。私は火を狙っていて、パターンを忘れていました..

しかし、スリープ（ランダム）を無視してすべての行をコンソールに出力する理由がわかりません

誰かが私にそれを説明できますか？

C# (および VB) の新しい非同期機能と .NET 4.0 のTask Parallel Libraryの違いはわかりません。たとえば、ここからEric Lippert のコードを見てみましょう:

このawaitキーワードは 2 つの異なる目的を果たしているようです。最初の出現 ( FetchAsync) は、「この値が後でメソッドで使用され、そのタスクが終了していない場合は、完了するまで待ってから続行する」という意味のようです。2 番目のインスタンス ( ) は、 「このタスクがまだ終了していない場合は、完了するまで今すぐarchive待ってください」という意味のようです。私が間違っている場合は、私を修正してください。

こんな風に簡単に書けませんか？

最初の値を実際に値が必要な場所に置き換え、2 つ目を待機が実際に発生awaitしている場所に置き換えました。機能は既に実装されており、意味的にはコードで実際に起こっていることにより近くなっています。Task.ResultawaitTask.Wait()(1)(2)

メソッドがイテレータと同様にステートマシンとして書き直されることは理解していasyncますが、それがもたらす利点もわかりません。動作に別のスレッドを必要とするコード (ダウンロードなど) は別のスレッドを必要とし、別のスレッドを必要としないコード (ファイルからの読み取りなど) は、TPL を利用して 1 つのスレッドのみで動作する可能性があります。

ここには明らかに何か大きなものが欠けています。誰かがこれをもう少しよく理解するのを手伝ってもらえますか?

私には、かなり高速でデータを生成する大量のセンサーと、それを消費する必要のある消費者がいるプログラムがあります。消費者は非常に異なる割合で消費します。

IObserver / IObservableを使用しているので、簡単な解決策は、イベントごとにタスクを作成し、OnNext（）呼び出しとデータをlamdaでラップすることでした。これは非常にうまく機能しました。生の呼び出しよりもオーバーヘッドが少ないことに驚きました。

問題は、これらのコンシューマーの一部は、厳密に適用されるイベントの順序を必要とし、イベントを見逃すことはできないということです。「PerferFairness」は十分ではありません。

私が思いついた最善の解決策は、event / OnNext（）ペアをラップする代わりに、InsertをParallelQueueにラップし、コンシューマーごとに1つのキューを作成し、キューのもう一方の端にスレッドを作成してOnNext（）を作成することです。呼び出します。

このアプローチに関する3つの差し迫った問題。Task / OnNext（）ラッピングソリューションよりもはるかに低速です。ParallelQueueにはブロッキングデキューがない（またはありますか？）ため、実装には少し注意が必要です。3つ目は、これは非常に一般的な問題のように思われるため、見逃した順序を強制する方法がないことを想像できません。たとえば、複数のタスクファクトリが基になるプールを共有し、各ファクトリに厳密に強制する設定があります。注文。

誰かが私がやろうとしていることを達成するための適切な方法を知っていますか？

編集：コンシューマーまたはプロデューサーごとのスレッドを含むソリューションは機能しません。生産者/消費者は長いチェーンを形成し、それぞれが数百あります。

私の質問は少しナイーブに聞こえるかもしれませんが、私はマルチスレッド プログラミングの初心者です。

着信外部データを処理するアプリケーションを作成しています。到着するデータごとに、次の方法で新しいタスクが作成されます。

データの項目は非常に速く (毎秒、0.5 秒など) 到着するため、多くのタスクが作成されます。各タスクの処理には約 1 分かかる場合があります。テストすると、スレッドの数が常に増加していることがわかりました。実際に動作するスレッドの数が安定して効率的になるように、作成されるタスクの数を制限するにはどうすればよいですか。私のコンピューターはデュアルコアのみです。

ありがとう！

画像圧縮サービスにタスクライブラリを使用しています。多くのファイルの同時実行性を圧縮します。しかし、私はユーザーがアイドル状態のとき（またはプログラムでこれ以上重要なタスクがないとき）にのみサービスを実行したいと思っています。

threadPoolは「スレッド優先度の変更」機能をサポートしていないことを知っているので、タスクもこの機能をサポートしています。

その機能をより高いレベルの制御で開発できますか？（たとえば、TaskSchedulerの優先順位）

以下のコードのセクションで異なる結果が得られるのはなぜですか

コード例 1

サンプル 2