問題タブ [producer-consumer]

Webサービス用の比較的単純な「プロキシ」アプリを作成しています。一般的な考え方は、TCPサーバー（非同期接続付き）がクライアントからデータを読み取り（文字列）、そのデータを（コールバック読み取り関数の一部として）2つのキュー（Q1とQ2）のいずれかに配置することです。別のスレッドがこれらのキューのデータを読み取り、それをWebサービスに渡します。第1四半期のデータは、第2四半期にある可能性のあるすべてのデータよりも優先される必要があります。

私は生産者/消費者パターンについて読んでいますが、それは私がキューに関して実装しようとしていることとほぼ同じようです。エンキューとデキューの操作は異なるスレッドで行われるため、キューがスレッドセーフであり、何らかのロックメカニズムをサポートしている必要があることは明らかですか？これは.NET4.0アプリケーションであり、新しいBlockingCollectionクラスとConcurrentQueueクラスのドキュメントを見ましたが、違いが何であるか、またはこのシナリオでそれらをどのように実装するかは正確にはわかりません。誰かがそれについてもう少し光を当てることができますか？ありがとうございました！

「高優先度」と「低優先度」のキューが次のように実装されたC＃4.0アプリがあります。

BlockingCollection highPriority = new BlockingCollection（1000）; BlockingCollection lowPriority = new BlockingCollection（1000）;

highPriorityで生成されたデータは、lowPriorityで生成されたデータよりも先に消費する必要があります。ここでのひねりは、データがいつでも2つのキューのいずれかに生成される可能性があることです。したがって、highPriorityですべてのデータを消費した後、lowPriorityである可能性のあるすべてのデータを消費します。lowPriorityでデータを消費しているときに、highPriorityで新しいデータが生成された場合、lowPriorityで現在のアイテムの消費を終了してから、切り替えてhighPriorityでデータを処理したいと思います。

誰かがこれを助けるためのアルゴリズムを提案できますか？擬似コードは問題ありません。どうもありがとう。

私は現在、AmazonSQSを利用するPythonで記述された分散処理アプリケーションに取り組んでいます。

Pythonでキューコンシューマーを適切に作成してホストする最もPython的な方法は次のとおりです。

• 信頼性が高い：問題が発生してコンシューマーが終了した場合、いくつかのリアクティブコードを実行し、処理を再開する必要があります
• コードを再利用します：これを支援するためにそこにいくつかのパッケージがなければなりません、そしてそれは車輪を再発明する必要がないのは素晴らしいことです:)

前もって感謝します！

アイテムをコンシューマー（C）にプッシュするプロデューサー（P）がいます。Cは、到着後すぐにすべてのアイテムを消費しないことを決定する場合があります。代わりに、キューにn個のアイテムがあるまで、または特定の時間間隔tが経過するまで非同期で待機してから、キュー内のすべてのアイテムを消費してタイマーを再起動する場合があります。

Pは、Cにプッシュしたアイテムが実際に消費されたときに通知を受ける必要があります。これは、障害が発生した場合の回復をPが担当するためです（Pは現在の状態をDBに保持します）。したがって、再起動の場合、Pは、消費者によって承認されていないため、どのアイテムを再度プッシュする必要があるかを知る必要があります。

私の最初のアイデアは、Cにコールバック関数（デリゲート）を介してPに通知させることでした。アイテムが消費されるたびに、消費されたアイテムのリストを使用してPのコールバック関数が呼び出されます。

しかし、プロデューサーに通知する他の（より良い）方法があるのではないかと思います。あなたの考えは何ですか？

ステファノ

更新：これまでの回答ありがとうございます。私は現在、非同期メソッドを使用することが、発行されているアイテムのステータスについて発行者と消費者を同期させるための洗練された方法であるかどうかを調査しています。

JavaDoc for ThreadPoolExecutorBlockingQueueは、エグゼキュータのバッキングにタスクを直接追加できるかどうかについては不明です。ドキュメントによると、呼び出しexecutor.getQueue()は「主にデバッグと監視を目的としています」。

私はThreadPoolExecutor自分でを構築していBlockingQueueます。キューへの参照を保持しているので、タスクを直接キューに追加できます。同じキューが返されるgetQueue()ので、の警告はgetQueue()、私の手段で取得したバッキングキューへの参照に適用されると思います。

例

コードの一般的なパターンは次のとおりです。

queue.offer()vsexecutor.execute()

私が理解しているように、一般的な使用法は、を介してタスクを追加することですexecutor.execute()。上記の私の例のアプローチには、キューをブロックするという利点がありexecute()ますが、キューがいっぱいになり、タスクが拒否されるとすぐに失敗します。また、ジョブの送信がブロッキングキューと相互作用することも気に入っています。これは私にとってより「純粋な」生産者/消費者のように感じます。

キューにタスクを直接追加することの意味：呼び出す必要がありますprestartAllCoreThreads()。そうしないと、ワーカースレッドが実行されません。エグゼキュータと他の相互作用がないと仮定すると、キューを監視するものはありません（ThreadPoolExecutorソースの調査によりこれが確認されます）。これは、直接エンキューのThreadPoolExecutor場合、0を超えるコアスレッド用に追加で構成する必要があり、コアスレッドがタイムアウトできるように構成してはならないことも意味します。

tl; dr

次のようにThreadPoolExecutor構成されている場合：

• コアスレッド>0
• コアスレッドはタイムアウトできません
• コアスレッドは事前に開始されています
• BlockingQueue遺言執行者の支援への参照を保持する

呼び出す代わりに、タスクをキューに直接追加することはできますexecutor.execute()か？

関連している

この質問（プロデューサー/コンシューマーワークキュー）も同様ですが、キューに直接追加することについては特に説明していません。

私はプロデュースによってファイルを読み取り、消費者によってそれらを他のファイルに書き込んでいます。入力ファイルがプロデューサーによって読み取られると、セマフォの1つを閉じる必要があり、消費者はプロデューサーが新しい入力をバッファーに挿入するのを待ちません。それを行う方法はありますか？

よろしく...

mmapされたファイルに書き込むプロデューサープロセスと、そこから読み取るコンシューマープロセスがあります。これはLinux上にあります。

プロデューサーがmmapに変更を加え、すぐにフラッシュされない場合、コンシューマーがそれにアクセスするとどうなりますか？ディスクから古いバージョンを取得しますか、それともフラッシュされていないページを取得するのに十分賢いですか？

私は単純な生産者/消費者プログラムを作成しようとしています。私はこのコードを持っています：

残念ながら、このコードはデッドロックで終了します。私はこの出力を持っています：

「バイ！」に注意してください。書かれていません。一方、余分な「消費」はです。このソリューションの何が問題になっていますか？終了を検出するためにグローバル変数を使用することは大丈夫ではありませんか？理解できない...

アイデアありがとうございます。

編集： あなたのアドバイスによると、ローカル変数の割り当てをvolatileに変更し、「\ n」を追加しましたが、問題は解決しません。

コルーチン、特に python について少し読んだことがありますが、完全には明らかではありません。

私は生産者/消費者モデルを実装しました。その基本的なバージョンは次のとおりです。

明らかに、MyConsumer にはプロデュース用のルーチンもある可能性があるため、データ パイプラインを簡単に構築できます。これを実際に実装したので、消費者/生産者モデルのロジックを実装する基本クラスが定義され、子クラスで上書きされる単一の処理関数が実装されます。これにより、簡単に定義され分離された処理要素を使用して、データ パイプラインを簡単に作成できます。

これは、よく引用されるチュートリアルhttp://www.dabeaz.com/coroutines/index.htmlなど、コルーチン用に提示される種類のアプリケーションの典型であるように私には思えます。残念ながら、上記の実装に対するコルーチンの利点は明らかではありません。呼び出し可能なオブジェクトの扱いが難しい言語では、得られるものがあることがわかりますが、python の場合、これは問題にならないようです。

誰かが私のためにこれに光を当てることができますか? ありがとう。

編集: 申し訳ありませんが、上記のコードのプロデューサーは 0 から 9 までカウントし、コンシューマーに通知します。コンシューマーは名前とカウント値を出力します。

次のコード フラグメントに問題があります。これは、イベント キュー (ConcurrentLinkedQueue) に追加されるイベント (processEvent メソッドの呼び出しによって提供される) を処理することを目的としています。イベントはイベント キューに追加され、run メソッドで定期的に処理されます。

ほとんどの場合、すべて問題ありません。しかし、processEvent メソッドの呼び出し後、イベントがキューに追加されたときに、実行部分が新しいイベントの存在を認識できないことがあります。

何が間違っているかについて何か考えはありますか？String 定数をロックとして使用する明らかな間違い以外に?