問題タブ [blockingcollection]

数字をハッシュする何かを書き、それらをリストと照合して、一致するハッシュが存在するかどうかを確認してみました。

for ループを使用してこれを正常に動作させた後、Parallel.For を使用してスピードアップを試みることにしました - 残念ながら、これによりデバッグに問題がある ArgumentOutOfRangeException が発生します。

エラーは、encrypter.EncryptCardNumber を呼び出すときにランダムに発生します (returnValue.ToString() のようです)。

2 つの質問があります。

1. 例外が発生するのはなぜですか?
2. 私が達成しようとしていることに BlockingCollection を使用する権利はありますか?

オリジナル

TPL やコンカレント コレクションを利用して、C# でプロデューサー/コンシューマー キューを作成するにはどうすればよいですか? 私は.NET 4.5+を使用しています。

これが私の最初の試みです：

2014 年 9 月 5 日編集: フィードバックに基づくバージョン 2。

編集 2014 年 9 月 7 日: 問題は解決しました。

Microsoft Task Parallel Library Dataflow私が望むものを正確にカプセル化しているクラスを発見しました。NuGet パッケージ:Install-Package Microsoft.Tpl.Dataflow

コミュニティに利益をもたらすために、いくつかのテスト コードを共有します。(こちらもhttps://dotnetfiddle.net/WbwUqz )

アクションA、B、Cと言う1000メッセージのバッチでアクションを実行する必要があります。これらのアクションを並行して実行できます。私は彼らのためにグループを作りました。並列性を高めるために、各グループにサブグループを作成しました。サブグループ内のタスクは、連続して実行する必要があります。ただし、2 つのサブグループを並行して実行できます。1000 のバッチが終了した後、データベースに保存するなどの処理を行う必要があります。しかし、すべてのタスクが完了するのを待つ方法を理解できません (1000 タスクの終わりに途中で待機することに興味はありません)。どんな提案でも大歓迎です。

したがって、基本的にはサブグループごとに OrderlyThreadPool を作成します。I am recv messages from say source, 利用可能なメッセージがない場合はブロックします。だから私のコードは次のようになります

問題: 複数のスレッドがリソースにアクセスしています。それらの数を定数に制限する必要がありMaxThreadsます。スレッド プールに入ることができないスレッドには、エラー メッセージが表示されます。

解決策: 以下のアルゴリズムで a の使用を開始しましたが、それには への呼び出しが必要であるBlockingCollection<string> poolことがわかりました。これは、常に着信スレッドを取得するため (デバッグ目的で以下の例では 10 にハードコードしました)、Web 要求を考えてください。 .BlockingCollectionCompleteAdding

これをよりよく達成する方法についてのアイデアはありますか?

ありがとう！

PS マルチスレッドの初心者です。資料を読むための提案があれば、大いに感謝します。

LE:得られた回答に基づいて、次のアルゴリズムを使用して目的の動作を実現できました。

を使用するクラスがありますConcurrentDictionary。

このクラスには、 this に対していくつかの操作を実行する 3 つの関数がありますConcurrentDictionnary。

各関数は、異なるスレッドによって呼び出されます。

• 最初の関数操作: dictionnary.Where、dictionnary.TryRemove
• 2 番目の関数操作: dictionnary.Where
• 3 番目の関数操作: dictionnary.TryAdd

この最後の関数は、さまざまな時点でブロックされます。ブロックせずに要素を辞書に追加する必要がありますが、どうすればよいですか?

BlockingCollection を使用してパケットのプロデューサーとコンシューマーの動作を実装しようとしていますが、それに関するドキュメントを理解するのに苦労しています。私が知る限り、何かをまったく削除できるという保証は明らかにないため、システムは遅すぎて、パケットの受信や処理などのパフォーマンスが重要な用途には使用できません。これが必要な理由は、パケットを受信するスレッドがそれらを処理するスレッドと同じではないためです。何か不足していますか、それとも別のアプローチを取る必要がありますか?

コード例:

私のアプリには、TCP 接続を処理するためのスレッドがいくつかあります (1 つは読み取り用、もう 1 つは送信用、もう 1 つは新しい着信接続の処理用)。各スレッドは、すべてのクライアントに対して指定されたタイプの操作を処理するためTcpClient、異なる IP で 5 つのインスタンスにデータを送信するとします。BlockingCollection送信スレッドからアクセスしているだけでなく、送信するデータを生成する別のスレッドからもアクセスしているため、バッファとして使用しています。送信スレッドで実行される私の関数は次のようになります。

注:BlockingCollection使用されるのは type<object[]>です。私の問題は、トラフィックのある時点でバッファがいっぱいになり始めることです。バッファに最大 500 メッセージの制限を設定しましたが、これは簡単にオーバーフローします。今、私がそれを正しく理解していれば（よくわかりません）、TryTakeアイテムを削除しようとします。コレクションが現在使用されている場合は、待機して再試行します。(注: タイムアウトを 50 ミリ秒に設定してみました)。これが正しい場合 (そうでない場合は、誰かが私を修正して別の理由を提案してください)、問題はおそらく、TryTakeが呼び出されるほとんどの時間、コレクションがビジーであることです。それはありますか？はいの場合、どのように解決しますか？

コレクションの使用法については、データを生成するスレッドによって、コレクションは、1 ～ 80 アイテムの範囲を反復する foreach で 2 秒に約 1 回アクセスされます。約 20 個以上のアイテムでバッファに問題が発生し始めますが、それまでは問題ありません。送信者スレッドは現在、1 つのクライアントのみに送信しますが、後で最大 15 になります。したがって、ピーク時には、80 アイテム x 15 ユーザー = 約 2 秒あたり約 1200 アクセスになります。アドバイスをいただければ幸いです。

最近実装が必要でしたが、特定の容量に達した場合はブロックし、空の場合はIProducerConsumerCollection<T>ブロックしたかったのです。これは実際には の実装であると確信していましたが、そうではないことに気付きました。何故ですか？TryAddTryTakeBlockingCollectionIProducerConsumerCollection<T>

インターフェイスBlockingCollectionを実装するのに適していないプロパティはどれですか?IProducerConsumerCollection

BlockingCollectionそれがラッパーであることは理解していますIProducerConsumerCollectionが、関係なく、それ自体が同じインターフェイスを実装する必要があると思いました。