問題タブ [pika]

いくつかのスレッドでメッセージを処理したいのですが、このコードの実行中にエラーが発生します:

エラーの説明:

バージョン: pika 0.9.5、rabbitMQ 2.6.1

RabbitMQ からのデータを処理するために Pika を使用しています。さまざまな種類の問題に遭遇したように見えたので、切断をどのように処理できるかを確認するために、小さなテスト アプリケーションを作成することにしました。

次のことを行うこのテストアプリを作成しました。

1. ブローカーに接続し、成功するまで再試行します
2. 接続したら、キューを作成します。
3. このキューを消費し、結果を python Queue.Queue(0) に入れます
4. Queue.Queue(0) からアイテムを取得し、ブローカ キューに戻します。

私が気づいたのは2つの問題でした：

1. 別のホスト (VM 内) の rabbitmq に接続している 1 つのホストからスクリプトを実行すると、このスクリプトはエラーを生成せずにランダムな瞬間に終了します。
2. RabbitMQ がインストールされているのと同じホストでスクリプトを実行すると、スクリプトは正常に実行され、実行され続けます。

これは、ネットワークの問題が原因である可能性があります。接続はあまり堅牢ではありませんが、パケットがドロップされました。

スクリプトが RabbitMQ サーバーでローカルに実行され、RabbitMQ を強制終了すると、スクリプトは次のエラーで終了します: "ERROR pika SelectConnection: Socket Error on 3: 104"

そのため、再接続戦略を正常に機能させることができないようです。誰かがコードを見て、私が間違っていることを確認できますか?

ありがとう、

ジェイ

Pika timed received exampleに従って、クライアントがより多くの同時リクエストを処理できるようにしたいと考えています。私の質問は、新しいメッセージが受信されるたびにhandle_deliveryを何らかの方法で呼び出すことができ、以前の handle_delivery の戻りを待たない場合は?

これを実行して、rabbitmq でキューを一覧表示できることはわかっています。

しかし、どうすればピカ経由でこれを行うことができますか?

ピカにユニコード文字列を投稿しようとしています。送信すると、次のメッセージが表示されます。

調べてみると、pika.BasicProperties を使用して文字セットまたはコンテンツ タイプを UTF-8 に設定する必要があるようですが、そのクラスに関するドキュメントが見つかりません。

Unicode データを投稿するには何を設定すればよいですか?

私は websockets から始めていますが、数のライブラリと構成オプションに非常に混乱しています。node.js サーバーが python/django のメソッドを呼び出すプロジェクトをセットアップしたいだけで、最後のメソッドが終了すると、結果が node.js サーバーに転送されます。これは私がこれまでに持っているものです:

このチュートリアルの Nodes.js AMQP :

このチュートリアルの Django Celery :

これが進むべき道かどうかはわかりません。誰かがこの問題に光を当てることができれば幸いです。

---編集

AMQP を使用して単純な文字列転送を行うことに成功しました。

test.py

app.js

});

それでも、これが最適な実装かどうかはわかりません。ここでは queue.bind() メソッドを使用していません。複雑なオブジェクト (json または単純な配列) を渡そうとすると、問題が発生します。この行を変更する

次のエラーが発生します。

複雑なオブジェクトをシリアル化するための解決策はありますか? 何か不足していますか？

タイムアウトを設定したいrabbitmqサーバーから受信するクライアントプログラムを作成しようとしています。指定された時間が経過すると、クライアントはキューからのメッセージの消費を停止して終了する必要があります。

Blocking connection でピカを使用しています。ブロッキング接続で add_timeout(delay,callback) が定義されているのを見ましたが、それを使用しようとすると、次のようなエラーが発生します。

これは私のクライアント プログラムです。タイムアウト条件をシミュレートするためにサーバーを実行していません。

指定されたタイムアウト期間内にメッセージを取得すると、クライアントはメッセージを出力することになり、それ以上の消費を停止する必要があります。

クライアントでタイムアウトを設定する方法を教えてもらえますか?

クライアントが '/test' に対して get 要求を行うと、node.js と python の間で AMQP を介して単純な文字列が交換されますが、応答をクライアントに送信する方法がわかりません (プロセスが非同期であるため)。

test.py

app.js

ユーザーリクエストはpythonに公開されますが、終了したらユーザーに返信するにはどうすればよいですか?

(注: これが最適な実装かどうかはわかりません。ヒント、提案を歓迎します!)

簡単な例として、新しい RabbitMQ(v 2.6.1) キューに 5 つのアイテムを追加します。

キューを消去してから、上記のコードを実行して 5 つのアイテムを追加します。

今、失敗した処理をシミュレートしようとしています。キューから消費する次のコードが与えられます。basic_ack への呼び出しがコメントアウトされていることに注意してください。

受信コードを実行して、キューからアイテムを取得します。予想どおり、アイテム #1 を取得します。

channel.basic_ack() への呼び出しはコメントアウトされているため、未確認のメッセージがキューに配置され、次のコンシューマーがそれを取得できるようになると思います。Redelivered プロパティが True に設定された状態で、メッセージ #1 が (再び) キューから出された最初のメッセージであることを願っています。代わりに、メッセージ #2​​ が受信されます。

また、#1 が再配信フラグを True に設定して戻ってくる前に、キュー内の他のすべてのメッセージが受信されます。

...

確認されるまで #1 を配信し続けるように設定できるプロパティまたはオプションはありますか?

私のユースケースは、順次生成されたファイルをデータ ウェアハウスにロードすることです。メッセージベースの処理を使用して、いくつかの新しいファイルの準備ができており、DW にロードされることをプログラムに知らせています。ファイルは、生成された順に処理する必要があります。

キューに送信するメッセージはそれぞれ個人のスケジュールごとに異なる必要があります。だから私はメッセージリストとそれぞれを再送するための関連する間隔を持っています。私はrabbitMQ/pikaとapschedulerを使用しています。多くの例によると、私は最も単純なBlockingConnection / channel/queueを作成しました。その直後にメッセージをプッシュしようとすると、すべてが正常に機能します。rabbitmqWebインターフェイスで、すべてのメッセージがキューに入れられていることがわかります。動作するコードは次のとおりです。

しかし、apschedulerコールバック関数を介して（まったく同じ方法で）メッセージをプッシュしようとすると、キューに表示されるメッセージはごくわずか（約1〜10）です（ただし、コールバックは常に発生し、メッセージの公開時に例外はありません！） 。

最後に、私はそのような警告を受け取り始めます：

/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/site-packages/pika/connection.py:642：UserWarning：Pika：書き込みバッファーが1125バイトで警告しきい値を超え、推定43フレーム遅れて警告（メッセージ％（self.outbound_buffer.size、est_frames_behind））

まだキューに新しいメッセージはありません。

私はPythonを初めて使用しますが、どんな助けでも大歓迎です。