問題タブ [node-amqp]

node-amqpライブラリを使用して RabbitMQ の複数のコンシューマーを作成したいのですが、prefetch-countオプションがコンシューマーごとのカウントを参照しているかどうかがわかりません。各コンシューマーに、他のコンシューマーに依存しない独自のプリフェッチ カウントを持たせたいと考えています。

ありがとう。

同時に 100 万人のユーザーがオンラインになっている分析システムを構築しています。メッセージブローカーなどのRabbitMQを使ってサーバーの容量を削減しています

ここに私の図があります

私のシステムには 3 つのコンポーネントが含まれています。

パブリッシャー サーバー : (プロデューサー) このシステムは nodejs 上に構築されました。メッセージを発行するこのシステムの目的queue

RabbitMQ キュー: このシステムは、送信されたメッセージを格納しますpublisher server。その後、 のキューからメッセージを送信するために 1 つの接続が開かれますsubscriber server。

サブスクライバー サーバー (コンシューマー) : このシステムはメッセージを受信します。queue

パブリッシャー サーバーのソース コード

すべてのコンシューマにブロードキャストを送信する排他的交換"message_queues"を作成します"fanout"

サブスクライバー サーバーのソース コード

"message_queues"取引所からメッセージを受け取る

メッセージを送信するときに実装します。システムは正常に動作しますが、このシステムのベンチマーク テスト パフォーマンスを試してみたところ (1 秒あたり最大 1000 ユーザーがリクエストを送信)、システムに問題がありました。システムが過負荷/バッファ オーバーフロー (または何かがうまく機能しない) のように見えます。

2日前にrabbitmqについて読んだだけです。私はそのチュートリアルが基本的な例であることを知っているので、現実の世界でシステムを構築するための助けが必要です.. 解決策と提案

私の質問が理にかなっていることを願っています

私は、RabbitMQ からのメッセージを処理できるワーカーに取り組んでいます。

ただし、これを達成する方法がわかりません。

メッセージを受信し、処理中にエラーが発生した場合、どうすればそのメッセージをキューの最後に入れることができますか?

nack または reject を使用しようとしていますが、メッセージは常に最初の位置に再配置され、他のメッセージはフリーズしたままです!

メッセージを最初の位置に配置する必要がある理由がわかりません。requeue や AllupTo などの他のオプションで「再生」しようとしていますが、どれも機能していないようです。

前もって感謝します！

簡単にできるはずのことをやろうとしていますが、驚くほど難しいことがわかっています。

RabbitMQ キューにサブスクライブする機能があります。具体的には、これは Channel.consume 関数です: http://www.squaremobius.net/amqp.node/channel_api.html#channel_consume

サブスクリプション ID で解決される promise を返します。これは、後でサブスクリプションを解除するために必要です。また、メッセージがキューから取り出されたときに呼び出すコールバック引数もあります。

キューから登録解除したい場合は、次の Channel.cancel 関数を使用してコンシューマーをキャンセルする必要があります: http://www.squaremobius.net/amqp.node/channel_api.html#channel_cancel。これは、以前に返されたサブスクリプション ID を取ります。

オブザーバブルがサブスクライブされたときにキューにサブスクライブし、オブザーバブルがサブスクライブ解除されたときにサブスクリプションをキャンセルするオブザーバブルにこれらすべてのものをラップしたいと思います。ただし、呼び出しの「二重非同期」の性質のために、これはやや難しいことが証明されています (つまり、コールバックとプロミスの両方があるということです)。

理想的には、私が書きたいコードは次のとおりです。

ただし、このコンストラクターは非同期サブスクライバー関数または破棄ロジックをサポートしていないため、これは不可能です。

私はこれを理解することができませんでした。ここで何か不足していますか？なぜこれが難しいのですか？

乾杯、アレックス