問題タブ [py-amqplib]

Python で RabbitMQ にアクセスするためにpy-amqplibを使用しています。アプリケーションは、特定の MQ トピックをリッスンする要求を時々受け取ります。

このようなリクエストを初めて受信すると、AMQP 接続とチャネルを作成し、新しいスレッドを開始してメッセージをリッスンします。

AMQPListenerは非常に単純です。

接続を作成した後、次のように、関心のあるトピックにサブスクライブします。

これで初めて、すべて正常に動作します。ただし、後続の別のトピックへのサブスクライブ要求では失敗します。後続のリクエストでは、AMQP 接続と AMQPListener スレッドを再利用し (トピックごとに新しいスレッドを開始したくないため)、channel.queue_declare()メソッド呼び出しの上のコード ブロックを呼び出しても戻りません。また、その時点で新しいチャネルを作成しようとしましたが、connection.channel()呼び出しも返されません。

私がそれを機能させることができた唯一の方法は、トピックごとに新しい接続、チャネル、およびリスナー スレッド (つまり、routing_key) を作成することですが、これは実際には理想的ではありません。どういうわけか接続全体をブロックしているのは wait() メソッドだと思いますが、どうすればよいかわかりません。単一のリスナー スレッドを使用して、複数のルーティング キー (または複数のチャネル) を持つメッセージを受信できるはずですか?

関連する質問:トピックが不要になった場合、リスナー スレッドを停止するにはどうすればよいですか? メッセージがない場合、 channel.wait() 呼び出しは永久にブロックされているように見えます。私が考えることができる唯一の方法は、ダミーメッセージをキューに送信して「毒」することです。リスナーによって停止の合図として解釈されます。

テキストをOCRできるプロセスのファームを作成したいと思います。複数のOCRプロセスによって読み取られるメッセージの単一のキューを使用することを考えました。

私はそれを確実にしたいと思います：

• キュー内の各メッセージは最終的に処理されます
• 仕事は多かれ少なかれ均等に分配されます
• 画像は1つのOCRプロセスによってのみ解析されます
• OCRプロセスは一度に複数のメッセージを取得しません（他の無料のOCRプロセスがメッセージを処理できるようにするため）。

AMQPを使用してそれを行うことは可能ですか？

pythonとrabbitmqを使用する予定です

メッセージがキューに配信されたことを確認したいのですが。

そのために、basic_publishに必須のパラメータを追加しています。basic.returnメッセージが正常に配信されなかった場合、メッセージを受信するには他に何をすべきですか？

メッセージが正常に配信されると関数が永久にハングchannel.wait()するため、リッスンに使用できません。（タイムアウトはありません）一方。電話をかけないと、メッセージが配信されなくても、は空のままになります。basic.returnwait()channel.wait()channel.returned_messages

py-amqplibバージョン0.6を使用しています。

どんな解決策も歓迎します。

py-amqpとFlopsyモジュールを使用すると、この奇妙な問題が発生します。メッセージを RabbitMQ サーバーに送信するパブリッシャーを作成しましたが、それを指定されたキューに送信できるようにしたいと考えていました。Flopsy モジュールではそれができないので、Publisher オブジェクトの _init__ メソッドでキューを宣言するためのパラメーターと行を追加して微調整しました

チャネル オブジェクトは py-amqplib ライブラリの一部です

私が抱えている問題は、メッセージを指定されたキューに送信している場合でも、メッセージをデフォルトのキューに送信しているということです。このシステムでは、非常に多くのメッセージを送信することが予想されるため、無駄な重複を作成してシステムにストレスを与えたくありません... コードをデバッグして py-amqplib ライブラリ内に移動しようとしましたが、エラーや不足しているステップを把握できません。また、コードの外側に py-amqplib 形式のドキュメントを見つけることができません。

なぜこれが起こっているのか、それを修正する方法についてのアイデアはありますか?

RabbitMQ サーバーにメッセージを送信してから、(「返信先」キューで) 返信メッセージを待ちたいと思います。もちろん、これらのメッセージを処理するアプリケーションがダウンした場合に備えて、永遠に待ちたくはありません。タイムアウトが必要です。非常に基本的な作業のように思えますが、これを行う方法が見つかりません。py-amqplibとRabbitMQ .NET clientの両方でこの問題に遭遇しました。

私がこれまでに得た最善の解決策は、中間で使用basic_getしてポーリングするsleepことですが、これはかなり醜いです:

確かに何か良い方法がありますか？

これと同様の質問があります。py-amqp/flopsyでは、「永遠に待つつもりです。メッセージが届くたびにこのコールバックを呼び出したい」と言うのは簡単ですが、見つかりません。 「OK、今すぐ待ちたいメッセージが届きました」という言い方をします。（多分GOTO？冗談です...）これを行うためのエレガントな方法はありますか？

RabbitMQを使用して、トピックごとに関心のあるサブスクライバーにメッセージをルーティングしています。各サブスクライバーにはキューがあり、関心のあるトピックにキューをバインドします。ユーザーがトピックリストからアイテムを削除できるようにします。

私の設定では、そのユーザーのキューからバインドされたトピックを「バインド解除」する必要があります。

pyamqplibを使用していますが、チャネルオブジェクトを介してこれを行う方法がわかりません。以前にバインドされたルーティングキーをキューから削除する方法はありますか？

昆布APIを介したAMQPトランザクションの使用に関するドキュメントが見つからないようです。このページでは、トランザクション状態にメッセージを追加する方法について説明していますが、関連性はないようです。

私はpikaバックエンドがそれらをサポートしていることを知っており、amqplibバックエンド（Imが現在使用している）もサポートしていると確信していますが、これが昆布でどのように公開されているかはまだわかりません。

編集：明確にするために、Imはchannel.commit（）、channel.select（）、...タイプのメソッドを探しています

AMQP にメッセージを送信するときに、交換が存在しないかどうかを検出できるようにしたいと考えています。

次の例を検討してください。

このスクリプトは取引所への公開を続けますが、取引所が存在しない場合でもエラーは発生しません。交換が存在する場合、メッセージが到着します。

私が Outing.wait() を追加すると、 amqp.exceptions.NotFound が発生します。これは私が望むものです。ただし、問題は、この場合に交換が存在する場合、メッセージは到着しますが、outinging.wait() がループをブロックすることです。(別のスレッドで outside.wait() を実行することもできますが、実行したくありません。)

これに対処する方法は？

アドバイスのヒントポインタは大歓迎です

ありがとう、

ジェイ

メッセージのパブリッシュ/コンシュームにamqplibを使用しています。キューにあるメッセージの数を読み取れるようにしたい (理想的には、確認済みと未確認の両方)。これにより、管理者ユーザーに適切なステータス ダイアグラムを表示し、特定のコンポーネントが負荷に追いついていないかどうかを検出できます。

amqplib のドキュメントに、読み取りキューのステータスに関する情報が見つかりません。

誰かが私を正しい方向に向けることができますか?