問題タブ [pika]

Pika で TornadoConnection アダプターを使用していますが、ドキュメントが軽いことがわかりました。私は…したい：

1. Tornado はハンドラーを介して Pika プロデューサーを開始します
2. Pika はコンシューマを実行します
3. Tornado は、消費が完了するとコールバックを介して通知され、Web クライアントを更新します

ドキュメントには Tornado IOLoop の使用方法が示されていますが、そのような例は見たことがありません。とても有難い。

2 つのノードを持つ rabbitmq クラスターがあります。ノード 1 でホストされるいくつかのキューを作成し、ノード 2 でホストされる他のキューを作成したいと考えています。

ConnectionParameters でホストを node2 に設定しても、キューは node1 に作成されます。

プログラム的には、pika を使用してキューを作成するノードを指定する方法がわかりません。にはそのようなパラメータはなくqueue_declare、次のような引数を渡してもうまくいかないようです:

ホスティング ノードを指定するためのインターフェイスはありますか? この場合を処理する別の方法はありますか？

ありがとう！

プロデューサーが複数のタスクを生成し、1 つ以上のコンシューマーが一度にタスクを取得して処理し、メッセージを確認する、基本的なダイレクト キュー システムをセットアップしようとしています。

問題は、処理に 10 ～ 20 分かかる場合があり、その時間にメッセージに応答しないため、サーバーが接続を切断することです。

コンシューマー向けの疑似コードを次に示します。

最初のタスクが完了すると、BlockingConnection の奥深くで例外がスローされ、ソケットがリセットされたことが通知されます。さらに、RabbitMQ ログは、時間内に応答しなかったためにコンシューマーが切断されたことを示しています (FIN を送信するのではなく、接続をリセットする理由は奇妙ですが、心配する必要はありません)。

これがRabbitMQの通常のユースケース（多くのコンシューマー間で分割されるべき多くの長時間実行されるタスクがある）であると信じていたので、私たちは多くのことを調べましたが、実際にこの問題を抱えている人は他にいなかったようです. long_running_task()最後に、ハートビートを使用し、別のスレッドで生成することが推奨されているスレッドに遭遇しました。

したがって、コードは次のようになりました。

これは機能しているように見えますが、非常に面倒です。chオブジェクトがスレッドセーフであると確信していますか? さらに、long_running_task()その接続パラメーターを使用して新しいキューにタスクを追加していると想像してください (つまり、この長いプロセスの最初の部分が完了したので、タスクを 2 番目の部分に送りましょう)。したがって、スレッドはconnectionオブジェクトを使用しています。そのスレッドセーフですか？

さらに言えば、これを行うための好ましい方法は何ですか? これは非常に面倒で、おそらくスレッドセーフではないので、正しく行っていない可能性があります。ありがとう！

私はrabbitmqに最新のpikaライブラリ（0.9.9+）を使用しています。私のrabbitmqとpikaの使用法は次のとおりです。

1. 私は労働者として長時間（約5分）のタスクを実行しています。これらのタスクはrabbitmqからリクエストを受け取ります。リクエストは非常にまれにしか発生しません。つまり、リクエスト間に長いアイドル時間があります。
2. 私が以前直面していた問題は、アイドル接続（アイドル接続による接続の閉鎖）に関連しています。だから、私はピカでハートビートを有効にしました。
3. ここで、ハートビートの選択が問題になります。Pikaは、ハートビートの受信と確認応答がリクエストの時間枠の間に行われるシングルスレッドライブラリのようです。
4. したがって、ハートビート間隔が、コールバック関数が長時間実行される計算を実行するために使用する時間よりも短く設定されている場合、サーバーはハートビート確認応答を受信せず、接続を閉じます。
5. したがって、最小ハートビート間隔は、ブロッキング接続でのコールバック関数の最大計算時間であると想定しています。

アイドル状態の接続を閉じるのを防ぐために、Amazon ec2の適切なハートビート値は何でしょうか？

また、tcp接続を維持するためにrabbitmqキープアライブ（またはlibkeepalive）を使用することを提案する人もいます。アプリケーションがハートビートを管理する必要がないため、tcpレイヤーでハートビートを管理する方がはるかに優れていると思います。これは本当ですか？RMQハートビートと比較した場合、キープアライブは良い方法ですか？

長時間実行されるタスクに複数のスレッドとキューを使用することを提案する人もいます。しかし、これは長時間実行されるタスクの唯一のオプションですか？このシナリオで別のキューを使用する必要があるのは非常に残念です。

前もって感謝します。問題を詳しく説明したと思います。詳細をお知らせいただければお知らせください。

Python で RabbitMQ レシーバー/コンシューマーを作成したいと考えていますが、メッセージを確認する方法がわかりません。pika のコールバックを使用せずに、自分のループでこれを実行しようとしています。

理解できれば、Java クライアントを使用getBasic()して、ブロックせずに利用できるメッセージがあるかどうかを確認できます。メッセージを受信して​​いる間はブロックしてもかまいませんが、メッセージがあるまでブロックしたくありません。

明確な例が見つからず、対応する pika の呼び出しをまだ把握していません。

CentOS 6.3 の VMWare の新規インストールをセットアップしました。インターネットは機能しており、すべてが機能しているようです。

RabbitMQ を試してみようとしていますが、チュートリアルのステップ 1 で行き詰まっています。

http://www.rabbitmq.com/tutorials/tutorial-one-python.html

基本的に、私は：

1. Linux インスタンスをセットアップする
2. erlang/esel など、RabbitMQ のすべての依存関係をインストールしました
3. Hello World チュートリアルを試す

この行で実際に失敗しています：

次のエラーが表示されます。

私はすべてのトラブルシューティングの試みを試してみようとしています.他の誰かもこの同じ問題を抱えていて、それについて投稿したと思っていました. まあ、私が最初だと思います！

とにかく、現時点では、RabbitMQ ライブラリには触れていないと思うので、これはピカの問題かもしれません。

以下は、127.0.0.1 に焦点を当てた Wireshark からの情報です。

Wireshark から詳細情報を提供できます。お知らせください

私が使用する Java クライアントと Python クライアントの両方がありますchannel.basicConsume()。ある時点で、プログラム全体を停止することなく、それらの消費者を停止したいと考えています。

Pika を使用した Python では、channel.stop_consuming()呼び出しを配置し​​ましたが、それらは無視しているエラーを生成します。動作するようです

Java では、stop_consume() が利用できないように見えるため、これを行う方法がわかりません。

私が見るすべてのドキュメントは、コンシューマを作成するすべての方法について述べていますが、コンシューマを停止する方法を示すものを見つけることができないようです.

これについて最善の方法は何ですか？

pika と tornado を使用する python コードを実行しようとしています。両方をインストールしましたが、エラーが発生しました

両方のパッケージがインストールされます/usr/local/lib/python2.7/dist-packages/

Pythonで遊ぶのはこれが初めてなので、なぜそれが見つからないのか本当にわかりません。コードではすべてが正しいようです。

Python、rabbitmq、pika、rabbit.js、node.js を使用しています。

アイデアは、クライアントにメッセージを送信し、それに応じて行動することです。送信されるメッセージには複数の種類があります。

したがって、サーバー側には、メッセージを受信するメソッドと、メッセージを送信する交換があります。

まだサーバー側では、node.js を使用して、複数のコンテキストがあり、交換は名前に従って呼び出すと思います (注:「クライアント」は、サーバーに接続されているクライアントのリストを指します)。

ここで起こっていることは、consumer_1 がすべてのクライアントにメッセージを送信し、consumer_2 が特定のクライアントに送信し、その部分がうまく機能していることです。

問題は、すべての消費者がすべての取引所からメッセージを受信して​​いることです。consumer_1 にメッセージを送信しようとすると、consumer_2 もメッセージを受信します。

実用的な面では、私が呼び出すと：

結果は次のようになります。

• ノードは「client_method_1」に出力します。
• また、「client_method_2」に発行しようとします。

他のコンシューマーを呼び出さずに特定のコンシューマーを呼び出すにはどうすればよいですか?

または、コンシューマを 1 つだけ宣言し、データをフィルタリングして別のメソッドを呼び出す必要がありますか?

編集：

TokenMacGuy のアドバイスに従って、交換タイプを「トピック」または「直接」に変更しようとしました。それはまさに私が望むことをするはずです。

しかし、常にありますが、どういうわけか交換を再宣言できません。ノードを再起動すると、次のエラーが発生し続けます。

わかりましたので、rabbitmq で list_exchanged を確認しましたが、コンシューマーはまだ「ファンアウト」のままです。

次の動きは、rabbitmq を停止、リセット (および後で強制リセット)、および開始することによってリセットすることでした。

consumer_1 にメッセージを送信しました

これで、consumer_1 は「direct」としてリストされます。

わかりました、consumer_1 だけを呼び出し、他のすべてがリストに表示されるので、その情報を含むファイルが必要です。どこ？何も思いつきません！！

とにかく、consumer_1 で同じ PRECONDITION_FAILED エラーが発生します。

何か案は？

Python-Pikaを使用してメッセージをフェッチするように、RabbitMQ サーバーをセットアップしました。問題は、永続配信モードを有効にすると、ワーカーがメッセージの処理に失敗することです。メッセージを解放する代わりに、RabbitMQ 接続がリセットされるまでメッセージを保持します。

処理に失敗したメッセージが、同じワーカーを含む利用可能なワーカーから合理的な時間枠内で再度取得されるようにする方法はありますか?

これは私の現在のコードです

アイデアは、メッセージを失いたくないということです。代わりに、メッセージを再発行するか、失敗した場合は再度取得することを望んでいます。メッセージがワーカーによって正常に処理されるまで、これは常に当てはまります。これは通常、ワーカーの 1 つが HTTP サーバーへの接続を開くことができない場合に発生します。