問題タブ [blockingcollection]

BlockingCollectionとプロデューサー/コンシューマーの問題に頭を悩ませようとしています。

私が達成したいのは、次のとおりです。

• オブジェクト（「ジョブ」）のリストをFIFO方式で保持するためのスレッドセーフなキュー。
• これらのジョブの結果のリストをFIFO方式で保持する2番目のスレッドセーフキュー。

言い換えると：

これまでの私の謙虚な試みは次のとおりです。

BlockingCollectionを選択した理由の1つは、負荷を最小限に抑えたいためです。つまり、アイテムが実際にコレクション内にある場合にのみ機能します（待機/スリープは処理しません）。foreachがそのための正しいアプローチであるかどうかはわかりません。

これが正しいかどうか、またはより良いアプローチがあるかどうかを教えてください。ありがとう！

編集 ユニットテストから、タスク内の作業は実際には同期的であることがわかりました。新しいバージョンは次のとおりです。

入力は大歓迎です！

IIS 7、.NET 4 で実行されている ASP.NET アプリケーションがあり、別の Web サービス (スタイル通知を起動して忘れる) と通信する必要がある場合があります。これまで、私はThreadfor each single 通知を起動していましたが、これはまったく効果がないと思います (新しいスレッドを作成するコストが高いため)。

そのため、BlockingCollection<>処理が必要な通知を追跡するためにコンテナーを使用し始めApplication_Start、このコードを使用して 3 つのコンシューマーを開始し、このキューを処理します。

sessionBuilder は、データベース セッションを提供する単なるオブジェクトです。

コンシューマーは次のようになります。

これは問題なく動作しているように見えますが、新しいバージョンのアプリケーションを IIS にデプロイするとすぐに、サーバーの CPU がw3wpプロセスで 99% 使用され始めます。これは間違いなく BlockingQueue の使用に伴うもので、変更前はこのようにはなっていませんでした。

CPU 時間がどこで費やされているかを正確に把握しようとして、WMemoryProfilerと WinDbg を使用して、すべてのマネージド スレッドのスタック トレースのキャプチャをいつでも確認できるようにしました。

このようにして、プロセスが CPU を 100% 使用している瞬間に、17 個のコンシューマ スレッドが実行されており、それらすべてが次の処理に時間を費やしていることがわかりました。

だから、明らかに、私はここで問題を抱えています。コンシューマーは、AppDomain のアンロード時に終了されません (デプロイ後に最初の HTTP 要求が到着するとすぐに発生します)。

1) 消費者が終了しないのはなぜですか? そのためだと思いAppDomainUnloadExceptionました。

2) 通知するグローバル キャンセル トークンを用意する必要があります (ただし、いつですか?) 現在の消費者に時間が終わったことを知らせますか? または、ここで推奨されるパターンは何ですか?

3) 上記の呼び出しスタックで 17 のスレッドが時間を費やしているのは正常ですか? Take()というか、 onを呼び出すとBlockingCollection<>、このような状況に同時に多くのスレッドが配置されるのは正常なことでしょうか?

複数のプロデューサーと複数のコンシューマーがいます。私の共有リソースは BlockingCollection です。ただし、私のコードは、コンシューマが 1 つしかない場合にのみ機能します。コードを実行するたびに出力が異なるため、競合状態であることはわかっています。

BlockingCollection がすべての同期などを処理すると思っていましたが、そうではありません。

すべてのプロデューサーとコンシューマー間で共有リソースを同期するにはどうすればよいですか?

これが私のコードです：

デリゲートを非同期的に呼び出すことができるように、ブロッキング コレクションを使用する必要があります。残念ながら、デリゲートには、構造体と追加の文字列で構成される 2 つのパラメーターがあります。構造体が使用されるのは、Interop を介した外部 C 関数への呼び出しの結果であるためです。
現在、ブロッキング コレクションを使用するときに構造体をコピーしないようにする方法を検討しています。
現在、私のコードは次のようになっています。

Consumer は StructTransfer をアンパックします。

StructTransfer は現在、次のようになっています。

ブロッキングコレクションを簡単に使用できるように、ポインターを使用してコンストラクターのコピーステートメントを回避する簡単な方法はありますか?

複数のコンシューマーを持つことができる BlockingCollection に似たコレクション オブジェクトを探しています。ただし、すべてのアイテムを消費するには、これらの複数の消費者が必要です。そこからアイテムを削除することはできません。基本的に、別のスレッドから項目を追加できるようにする必要があり、Enumerator を取得した人は、これまでに追加されたすべての項目をスクロールする必要があります。閉じるまでブロックします。

そのようなコレクションタイプはどこにありますか?

[編集] いいですね、明確にすると思います... BlockingCollection.GetConsumingEnumerable() は、アイテムが利用できない場合、または CompleteAdding が呼び出されるまで、列挙をブロックする列挙型を返します。コレクションからアイテムを削除しないことを除いて、同じものが欲しいです。

[編集] わかりました。なんでもいい。私はこれを解決しました：私自身のブロックリストを書くことで。

WebベースのAPIにリクエストを送信するコンポーネントがありますが、APIのデータ制限に違反しないように、これらのリクエストを調整する必要があります。つまり、すべての要求は、送信される速度を制御するためにキューを通過する必要がありますが、最大のスループットを達成するために同時に実行できます（実行する必要があります）。各リクエストは、完了した将来のある時点で、呼び出し元のコードにデータを返す必要があります。

データの返送を処理するための優れたモデルを作成するのに苦労しています。

エンキュープロセスとデキュープロセスはバッファの両端にあるため、Iを使用してメソッドからBlockingCollectionaを返すことはできません。その代わりに、フォームのコールバックを含む型を作成します。Task<TResult>ScheduleRequestItem<TResult>Action<Task<TResult>>

アイテムがキューからプルされると、開始されたタスクでコールバックを呼び出すことができますが、その時点でジェネリック型パラメーターを失い、リフレクションとあらゆる種類の厄介さを使用したままになります（可能です）。

例えば：

Task<TResult>リクエストをバッファリングし続けても、リクエストがバッファからプルされてAPIに送信されたときにクライアントに返すにはどうすればよいですか？

これが私の問題です。
BlockingCollection にいくつかの BitmapImages をロードしています

読み込みはバックグラウンドワーク スレッドで行われます。

コードのこの部分をデバッグするとすべて問題ないように見えますが、問題は今発生しています...

画像の読み込みが完了したら、UI に 1 つずつ表示したいと思います。そのためにディスパッチャーを使用していますが、呼び出されたスレッドは別のスレッドに属しているため、オブジェクトにアクセスできないというメッセージが常に表示されます。

デバッグは、エラーがここにあると非難します

私はすべてを試していますが、何が間違っているのかわかりません。誰でも何か考えがありますか？

C# でコンシューマーを実装しようとしています。同時に実行できるパブリッシャーは多数あります。3 つの例を作成しました。1 つは Rx と subject を使用し、もう 1 つは BlockingCollection を使用し、もう 1 つは BlockingCollection の ToObservable を使用しています。この単純な例では、それらはすべて同じことを行っており、複数のプロデューサーで動作するようにしたいと考えています。

それぞれのアプローチの違いは何ですか?

私はすでに Rx を使用しているので、このアプローチをお勧めします。しかし、OnNext にはスレッド セーフの保証がなく、Subject と既定のスケジューラのキューイング セマンティクスがどのようなものかわかりません。

スレッドセーフな件名はありますか?

すべてのメッセージが処理されますか?

これが機能しない他のシナリオはありますか? 同時に処理されていますか？

Rx ではありませんが、簡単に使用/サブスクライブできます。アイテムを受け取り、それを処理します。これは連続して発生するはずです。

サブジェクトに少し似たブロッキング コレクションを使用することは、良い妥協点のように思えます。async/await を使用できるように、暗黙的にタスクにスケジュールを設定すると思いますが、それは正しいですか?

BlockingCollection を使用して、1 つのプロデューサーと複数のコンシューマーを持つことを達成しようとしています。このようなコードを作成しました

しかし、要素を追加するたびに、その値を消費するタスクは 1 つだけで、他のすべてのタスクにはそれがありません。複数の消費者をシミュレートするにはどうすればよいですか。

Rx または TPL ライブラリを使用できません。

どうも