とスケーリングのキューは、私たちは、使用する複数のキューに交換機からのルーティングキーをシャードに便利であるのRabbitMQのconsistent hash exchange。
私たちはキューを使用して作業者を作業に仕上げます。理想的には、これらの労働者を動的に拡大したいが、これは問題を提示する。
スケールアップするには、新しいキューとバインディングを追加する必要があります。それ自体では、これは大したことではありません。シャードされたキーが別のキューに移動し始める可能性はありますが、
したがって、作業者AがThing1を管理していた可能性がありますが、新しいキューを追加すると、作業者BがThing1のメッセージを取得する可能性があります。作業者BがThing1メッセージを取得する前に、作業者Aがすべての処理を完了していることが重要です。
この問題を緩和する方法やプラグインはありますか？
この場合、ハッシュ交換の代わりにメッセージを送信するワーカーを使用します。
producer1 ... producern =>トピック交換=>キュー=>ディスパッチャのワーカー=> QUEUE1 ... queuen => worker1 ... workern
ディスパッチャ労働者はすべてのメッセージを追跡することができますこの方法。このため、キュー内に残っているメッセージの数を確認するか、ワーカー確認メッセージを確認するか、rabbitmqs RPC機能を使用することができます。
分散ハッシュ交換（といくつかの機能）を別のワーカーに効果的に実装しますか？ –
この特殊なケースではい。それはより多くの作業ですが、あなたが望むきめ細かなコントロールを提供します。私が考えることができるもう一つの事は、実際の労働者の中から現時点でmsgを処理することが有効な場合、rpcの実装として "尋ねる"ことができる一種の制御作業者です。 – slowjack2k
私が尋ねなかったことは、おそらく、このプロデューサーからの最後のメッセージが処理されるまで、プロデューサーの中を待つ可能性があります。その後、プロデューサーを変更するだけです。 – slowjack2k
私の最初の質問は、同じキューに対して複数のワーカーを使用できない理由です。特定の順序を保証する必要がある場合に限り、1つのワーカーを1つのキューに制限する必要があります。 – slowjack2k
私はそれを何度も読まなければならなかった。メッセージの順序はあなたにとって重要です。したがって、新しいワーカーを追加する前に、すべてのキューが空であるかどうかを確認する必要があります。プラグインは、新しいキューを追加するときに競合状態が存在することを示します。まれにしか複製されないメッセージで終わることがあります。私はこのプラグインをこのような特殊なケースですべて追加するつもりはない。私は1つの入力キューを作成し、1つのディスパッチャワーカがリッスンし、このプロセスはすべてのメッシュを右側のメッセージキューに配信する役割を担います。 – slowjack2k
@ slowjack2kワーカーはメモリ内の状態を管理しているため、1人の作業者はThing1の状態のみを管理する必要があります。 –