問題タブ [node-amqplib]

私はNodeJSとRabbitMQに取り組んでいます-

• マスター NodeJS サービス (1 つのインスタンス) データをキューにプッシュします。
• スレーブ NodeJS サービス (複数のインスタンス) は、データを消費して処理します。
• デフォルトの交換が使用されています。

スレーブ サービスは PM2 クラスター モードで実行されています。つまり、実行中のスレーブ サービスのインスタンスが 8 つあるということです。

私の予想では、マスター サービスがキューを介してデータをプッシュし始めると、スレーブ サービスはそれらを非同期的に消費し始めるはずです。

例として、マスター サービスがキューを介して 10 個のジョブをプッシュし、各ジョブが完了するまでに 5 秒かかる場合、スレーブはジョブを完了するのに 50 秒かかります。

これは、複数のスレーブを使用する目的を完全に無効にします。通常、スレーブは一度に 8 つのジョブを取得する必要があるためです。

RabbitMQ ダッシュボードによると、上記のセットアップでは -

• 9 接続 (1 マスター + 8 スレーブ)
• 9 チャネル (1 マスター + 8 スレーブ)
• 1 キュー

セットアップ全体でデフォルトの交換が使用されます。

私の質問は -

スレーブがキューからデータを非同期的に読み取ることができないのはなぜですか?

キュー内の次のアイテムに設定noAckしてもtrue、現在のアイテムが処理されるまでピックアップされません

私の意図は、ユーザーの複数のスレーブ インスタンスによるキューの消費率をスケールアップすることですが、ここで何かが欠けていると思います。

これがコードベースです-

スレーブ出力 -

お気づきのように、さまざまなスレーブがラウンド ロビン方式でメッセージを取得していますが、それらは同期的に動作しています。

ありがとう！

を使用してメッセージをキューに直接送信できることを理解しています。channel.sendToQueueこれにより、1 つのコンシューマーのみが各タスクを処理する、タスクとワーカーの状況が作成されます。

また、トピックベースの交換で使用できることも理解していますchannel.publish。メッセージは、ルーティング キーに基づいてキューにルーティングされます。ただし、私の理解では、これは一致するキューのすべてのサブスクライバーに常にブロードキャストされます。

基本的にトピックベースの交換を使用したいのですが、各タスクを処理するコンシューマは 1 つだけです。ドキュメントを確認しましたが、これを行う方法がわかりません。

私のユースケース:

複数の場所にマイクロサービスのインスタンスをセットアップしています。カリフォルニアに 2 つ、ロンドンに 3 つ、シンガポールに 1 つなどです。タスクが作成されるとき、重要なのは、特定の場所にあるインスタンスの 1 つによって処理されることだけです。

もちろん、「usa-90210」、「uk-ec1a」などの名前の何百ものキューを作成できます。トピックを使用すると、はるかにクリーンになるように思えます。また、ワイルドカードを使用できることを考えると、より柔軟になります。

これが RabbitMQ の機能でない場合は、他の考えやアイデアも受け入れます。

アップデート

istepaniuk の提案に従って、2 つのワーカーを作成してみました。それぞれが独自のキューを交換にバインドします。

残念ながら、両方のコンシューマーはまだメッセージを受信して​​います。