フォローされたコードを実行すると、キューからメッセージを取得できることがわかりましたが、callbackトリガーできません

走ろうとするpython -m pdb test.py

でループします

Kombu を使用したメッセージ パッシング システムの構築に取り組んでいますが、次の問題が発生しました。ルーティング キー 'x' と 'y' にルーティングされているメッセージがあるとします。これは、「x」と「y」の間に依存関係がない状況でうまく機能します。

ただし、ルーティング キー 'a'、'b'、および 'c' にデータを送信し、単一のキューがそれらのルーティング キーからメッセージを取得している別の状況を考えてみましょう。単一のコールバックを処理するために a、b、および c からのデータが一緒に必要な場合、これらのメッセージを単一のワーカー ドレインに集約する方法はありますか?それとも、これはメッセージ パッシング パラダイムの完全な粗悪品ですか?

メッセージを別の場所 (Redis など) にキャッシュし、必要なデータがある場合にのみ処理できることはわかっていますが、データをキャッシュして毎回ワーカーを起動することなく、Kombu がこれを実行できるかどうか疑問に思っています。ご提案ありがとうございます。役立つ場合は、コード例をいくつか挙げることができます。

私は 2 つのサーバーを持っており、それらを A と B と呼びます。B は RabbitMQ を実行し、A は Kombu 経由で RabbitMQ に接続します。B で RabbitMQ を再起動すると、kombu 接続が切断され、メッセージが配信されなくなります。次に、A のプロセスをリセットして、接続を再確立する必要があります。より良いアプローチはありますか?つまり、RabbitMQ プロセスが再起動された場合でも、Kombu が自動的に再接続する方法はありますか?

私の基本的なコードの実装は以下のとおりです。よろしくお願いします。:)

問題

システムの 1 つのキュー ハブとして機能するRabbitMQ サーバーがあります。先週かそこらで、そのプロデューサーは数時間ごとに完全に停止します.

私は何を試しましたか

強引な

• コンシューマーを停止すると、ロックが数分間解放されますが、その後ブロックが戻ります。
• RabbitMQ を再起動すると、問題は数時間解決しました。
• 醜い再起動を行う自動スクリプトがいくつかありますが、明らかに適切な解決策にはほど遠いものです。

より多くのメモリを割り当てる

cantSleepNow の答えに従って、RabbitMQ に割り当てられたメモリを 90% に増やしました。サーバーにはなんと 16GB のメモリがあり、メッセージ数はそれほど多くない (1 日あたり数百万) ため、問題にはならないようです。

コマンドラインから:

そして/etc/rabbitmq/rabbitmq.config：

コードとデザイン

私はすべてのコンシューマーとプロデューサーに Python を使用しています。

生産者

プロデューサーは、呼び出しを処理する API サーバーです。通話が着信すると、接続が開かれ、メッセージが送信され、接続が閉じられます。

消費者

コンシューマーはそれぞれ若干異なりますが、一般的に次のパターンを使用します - 接続を開き、メッセージが到着するまで待機します。接続は、長期間 (たとえば、数日) 開いたままにすることができます。

設計推論

• コンシューマーは、常にキュー サーバーとの接続を開いたままにしておく必要があります。
• Producer セッションは、API 呼び出しの存続期間中のみ存続する必要があります。

この設計では、約 1 週間前まで問題は発生していませんでした。

Web ビュー ダッシュボード

Web コンソールは、 のコンシューマーが、、およびからのコンシューマー127.0.0.1を172.31.38.50ブロックしていることを示しています。172.31.38.50172.31.39.120172.31.41.38172.31.41.38

システム指標

念のため、サーバーの負荷を確認しました。予想どおり、負荷平均と CPU 使用率の指標は低くなっています。

ウサギの MQ がそれぞれこのようなデッドロックになるのはなぜですか?

Celery-Kombu0-SQS の組み合わせを使用するようにプロジェクトを構成しました。 Celery ワーカーを実行しているノードごとに Celery pidbox キューが作成されます。これらのキューを抑制/回避しようとしています。他のブログを読んで、SQS ではこのオプションがデフォルトで無効になっていることを読みましたが、まだ pidbox が作成されています。

抑制できない場合、どのように作成を抑制するか、pidbox の名前を構成できますか?

メッセージをキューに生成し、メッセージがアプリケーションで正しく消費および処理されるかどうかを確認するいくつかのテストを作成しようとしています。

そのために、kombu ライブラリ、特にメモリ内トランスポートの実装をいじっています。

それでも、生成されたメッセージが消費されるということは機能しません。

したがって、私の質問は、メッセージをメモリ内で生成および消費する簡単な単体テストを誰かが提供できるかどうかです

過去に、送信者と受信者の両方がCelery経由で処理するためにRabbitMQに送信されるタスクを理解するCelery async pythonおよびDjangoアプリケーションを使用しました(クラスター内の同じアプリなど)。

現在、.NET サービスがメッセージを JSON の形式で RabbitMQ に発行しているユース ケースがあります。キューごとに 1 つのメッセージ タイプです。.NET アプリはメッセージを発行するだけで、Rabbit が適切に受信したことを確認してから立ち去ります。次に、メッセージを消費するために Django アプリケーションを実行します。したがって、これらの JSON メッセージを消費する適切な方法がわからないため、この Django アプリはモデルを介してデータを単純に保存し、メッセージが処理されたことを確認できます。

Celery/Kombu を使用する キューにアクセスするための最良の方法がわからないため、キューごとに直接消費者がいます。Celery は内部で Kombu を使用していることを理解しているので、そこでコンシューマーを作成できると思いますが、Celery と Flower のようにプロセスを管理することは不可能であり、アプリの開始時に不正なスレッド コンシューマーを作成するのは不安定に思えます。せいぜい。

新しい SO ドキュメンテーション プロジェクトのために Kombu をドキュメント化していたときに、この問題に遭遇しました。

Consumer Mixinの次の Kombu コードを考えてみます。

コードは次のように失敗します。

print_even_characters()メッセージがとで 2 回 ACK されたためprint_odd_characters()です。

機能する簡単な解決策は、最後のコールバック関数のみに ACK を送信することですが、他のキューまたは接続で同じ関数を使用したい場合、モジュール性が損なわれます。

複数のコールバック関数に送信されるキューに入れられた Kombu メッセージを ACK する方法は?

私はしばらくの間使用してきKombuました-そして今日、展開のために新しいマシンを準備したとき、私はからTypeError乗っています:importkombu