RabbitMQを使用する100万人のユーザー向けのアラートサービス

クラウドベースのVLSI Webアプリケーションのプラットフォーム（一部の基盤、すべてのアプリケーションソリューションのベースとなるフレームワーク）の作成のほぼ最初に、サーバーからのイベントをユーザーに通知できるツールなしでは生きていくのはかなり難しいことを認識しました。 私たちは皆、同僚（10メートルを歩くのが面倒）からの新しいメッセージをすぐに見たいと思っています。 しかし、なるまでの道のりは厄介なものだったので、ユーザーからの同時接続が5.0e3から1.0e6に成長するときに直面した困難について少し話しましょう。

開始する

「額の」解決策は、サーバー側のブラウザからの定期的な調査かもしれません。 しかし、すでに（2013年、人々がまだHarlem Shakeの下でクリップを撮影していたとき）私たちはさまざまなサービスを利用していたため、定期的な調査により1秒あたりのリクエストのメトリックが大幅に向上しました。 したがって、質問は、任意のサービスの任意のバックエンドが任意のユーザーにデータを送信できるようにする中間スーパーサービスから生じました（逆の相互作用はHTTP経由で機能します）。

当時、ウェブソケットはすでに十分な人気を得ていたため、全体的な負荷を軽減し、さらに、ユーザーのブラウザへの情報配信の遅延を最小限に抑えることができました。 しかし、残念ながら、お気に入りのユーザーに人気のあるすべてのブラウザーがWebソケットをサポートしているわけではないため（すべてのプロキシが付属物でそれらを見つけることができないわけではありません）、XHRポーリングやXHRストリーミングなどのより一般的なテクノロジーのサポートも必要でした。

このサービスは「昨日」必要だったため、開発戦略が「ひよこと生産」にならないように、既存のソリューションを探すことにしました。 検索の結果、RabbitMQブローカーに移動しました。このブローカーには、Web環境を操作するためのプラグイン（Web STOMPアダプターとSockJS）が既にありました。 1つ目はAMQPバイナリプロトコルをテキストSTOMPに関連付けることができ、2つ目はユーザーとのWebソケットのような接続を確立することを可能にします（サポートされているトランスポートの選択を担当します）。 実り多い協力を期待して、いくつかのノードのクラスターを展開してテストしました。 RabbitMQは動揺せず、それに基づいてユーザー通知システムを構築することにしました。 最初はすべて順調でしたが、Webアプリケーションの機能の開発とアクティブユーザーの数の増加から、朝はコーヒーだけでは始まらないことがわかり始めました。

RabbitMQについて一言

RabbitMQはマルチプロトコルメッセージブローカーであり、各ノードが読み取りおよび書き込み要求を処理できる複数のノードへの完全なデータレプリケーションを備えたフェールセーフクラスターを編成できます。 Erlangで書かれており、 AMQPプロトコルバージョン0.9を主なものとして使用しています。 AMQPのオブジェクトの相互作用の簡単な図を考えてみましょう。

図 1. AMQPのオブジェクトの相互作用。

キューはすべての着信メッセージを保存し、コンシューマーに提供します。 エクスチェンジャー（エクスチェンジ）は、キュー（または他のエクスチェンジャー ）との間に作成されたリンク（バインディング）に基づいてメッセージをルーティングします（保存しません）。 RabbitMQの観点から見ると、キューは状態（メッセージ自体をキャッシュできる）を持つErlangプロセスであり、交換機はルーティングロジックが存在するコードを持つモジュールへの「リンク」です。 つまり、たとえば、10,000個のキューがすでに約800 MBである場合、10,000個の交換機が約12 MBのメモリを消費します。

クラスターモードでは 、メタ情報がすべてのノードにコピーされます。 つまり、各ノードには、交換機、キュー、それらの接続、コンシューマー、およびその他のオブジェクトの完全なリストが含まれています。 キュー自体のプロセスはデフォルトで1つのノードにのみ配置されますが、ポリシーの助けを借りてレプリケーション（ ミラー化されたキュー ）を有効にでき、データは必要な数のノードに自動的にコピーされます。

素朴な実装

最初の実装は非常に簡単でした：

• バックエンドとクライアントの両方が接続された3ノードのRabbitMQクラスター。
• 開始時に、バックエンドは、イベントの名前（「online.sbis.ru:contacts.new-message」などのWebサイトの名前を含む）と直接タイプを使用して交換機を作成しました。
• クライアントはSockJSを介して接続し、ユーザーIDと等しいキーで対応する交換機にサブスクライブすることにより、必要なイベントを「リスニング」しました。
• イベントが発生した時点で、バックエンドはAMQPメッセージを対応する交換機に送信しました。 メッセージには、ルーティングキーとして受信者識別子がありました。

概略的には、すべて次のようになりました。

図 2.接続図。

図 3.ブローカー内のオブジェクトのスキーム。

イベントの成長

当時、ブラウザのクライアントコードは1ページあたり平均で約10個のイベントを「リッスン」していましたが、成長にはそれほど時間はかかりませんでした（たとえば、メインページに約70個しかありません）。 予測不可能な生活に痛みを加えます-いつでも、どんな時間間隔でもイベントが必要になる場合があります。

ただし、STOMPを介して任意の交換機にサブスクライブすると、交換機に既に関連付けられているキューが自動的に作成されるという事実によって、主な問題はもちろん作成されました。 そして、そのようなサブスクリプションが30個ある場合、（1つの接続内に）30個のキューがあり、消費されるリソースとキラーのOOMサービスへの呼び出しの頻度に影響を与える以外にありません。

この問題を解決するために、アプローチを変更し、AMQPメッセージ自体のヘッダー内の交換機の名前からイベントの名前を削除しました。 構造は単純化され、次のようになり始めました。

図 7.最初の改善後のオブジェクトの構造。

したがって、ページ上のイベントの数への依存を削除し、ユーザーからのサブスクリプションが1つだけ必要になりました。 しかし、新しいアプローチでは、イベントがユーザーに配信されて初めてユーザーにイベントが必要かどうかを理解できるため、トラフィックを犠牲にしなければなりませんでした。

出版エリア

私たちのシステムには、「顧客」（当社の製品を使用する会社）と「ユーザー」（顧客の従業員）の概念があります。 企業は完全に異なっており、誰かが数人の従業員を抱えている一方で、誰かが2人または3人の従業員を抱えている場合があります。 クライアントのすべてのユーザーにメッセージを送信したい場合は、リスト全体をサイクルで確認する必要があり、時間がかかります。 したがって、これで何かをすることが急務でした。

解決策は、ユーザー、クライアント、そしてすべての人に向けた3つの出版分野を作成することでした。 システム情報を送信するにはグローバルエリアが必要です。たとえば、バックエンドを更新した後にページをリロードしてください。 各エリアは、直接タイプの交換機としてRabbitMQに実装され、クライアントに複数のサブスクリプションを再度行わせないように、ファンアウトタイプの別の交換機-パーソナルユーザー交換機が追加されました。

図 8.出版物の領域とパーソナルエクスチェンジャーを備えたオブジェクトの構造。

パーソナルエクスチェンジャーは必要な識別子でエリアエクスチェンジャーに連絡し、ユーザーはパラメーターなしでそれに接続する必要がありました。 ただし、ユーザーがサブスクライブする前にパーソナルエクスチェンジャーを作成する必要がありました。そうしないとエラーが発生します。 既存のすべてのユーザーに対してオブジェクトを事前に作成するのは非常に高価です。 そのため、接続時に動的に作成することにしました。 これを行うために、Nginx（RabbitMQの前に立つ）でLuaスクリプトが作成され、リクエストをブローカーにプロキシする前に補助ユーティリティをプルしました。 ユーティリティは、必要なすべてのものの存在をチェックし、不在で作成しました。 その後、彼らはSockJSに必須の/ infoリクエストに対してのみユーティリティ呼び出しを行いました。

キューの自動削除

最初は、切断後に自動的に削除されるキュー（自動削除）を作成しましたが、これにより2つの大きな問題が発生しました。

1. コミュニケーションが少し途切れたため、まだ届いていないメッセージを失いました。

   
   
   
   
   
   
   
2. 特にサブスクリプションの数が最適化されるまで、キューおよびリンクを作成/削除するために、システムで多数の操作が実行されました。 クラスターモードでは、このような大規模なワークフローがすべてのノードに広がり、負荷が増加しました。 この現象はバインディングチャーンと呼ばれます。

その結果、切断時に削除されなかった永続キューに切り替えましたが、ユーザーが5分以上留守になった場合、それらの有効期限ポリシーを切断しました。 （このシステムは、ユーザーが永久に受信したものではないデータを保存することを目的としておらず、迅速な配信のためにのみ開発されています。）

RabbitMQクラスターの負荷は減少しましたが、ユーザーの成長は継続しています。 同時ユーザー5万人を超えると、システムはリソース消費の限界に近づきました。 もう一度変更する必要がありました。

アラート2.0

以前に問題が発生した後、この場合、RabbitMQクラスターのノードを増やしてスケールアップすることは、最適な方法ではないことに気付きました.1つのノードの問題が他のノードのパフォーマンスに影響し、データ複製のレベルでの何らかのフォールトトレランスがシステムに必要ではないためです。 しかし、一方で、バックエンドが接続のための単一のポイントを持つことは非常に便利です。 いくつかの議論の後、システムを2つのレイヤーに分割する必要があるという結論に達しました。1）バックエンドを接続するためのRabbitMQクラスターと、2）ユーザーを接続するための独立したRabbitMQノード。 その結果、次のスキームがあります。

図 9. 2層配線図。

ユーザーはWebプロトコルを介して接続するため、独立したサイトには「Web」という名前が付けられました。 これらのノード自体は、 RabbitMQ プラグインFederationを使用してクラスターに接続されます。

図 10. 2層スキームのオブジェクトの構造。

Webサイトでのフェデレーションのセットアップは非常に簡単です。

1. アップストリーム（アップストリーム、ソース）を作成します;単一ノードとフォールトトレランス用のノードのリストの両方で構成できます。
2. このような名前テンプレートを持つ交換機はアップストリームと同期する必要があるというポリシーを作成します。

その後、フェデレーションはアップストリームノードに接続し、それらに独自のキューを作成し、必要な交換機に署名します。その後、メッセージを受信し、ローカル交換機でそれらを複製します。

したがって、クラスターの交換機で公開されたメッセージは、すべてのWebサイトの同じ交換機に送られます。 後者は、ユーザー数に応じてほぼ直線的にスケーリングできます。 すごい！

顧客の分布

最も理想的なシナリオは、すべてのWebサイト間でのクライアント接続の均一な分散と、そのうちの1つのみへの「忠実度」です。これにより、ユーザーのオーバーヘッドを削減できます。 ただし、ノードが失敗することを忘れないでください。ユーザーは、このノードが修復されるまで待つ必要はありません。 N個のWebサイトの前には、M個のディスパッチャがあります（Nginx、必要なバックエンドへのプロキシ要求）。 そして、私たちがしなければならなかったのは、ディスパッチャの設定が同一であることを確認し、ユーザーIDでバックエンドを選択するように設定することです（たとえば、 ハッシュディレクティブを使用して）。

結果のスキームには欠陥がないわけではありませんが、最終的にはより多くのユーザーにサービスを提供し、8つのWebサイトへの約30万の同時接続に耐えることができました。

アラート3.0：新しい希望

フェデレーション、スマートな停止！

フェデレーションはそれ自体ですべてを作成するという点で便利ですが、その目的はノード間のトラフィックを最小限にすることでもあります。 したがって、それによって制御される交換機にあるキーを使用して、アップストリームでのみキューの接続を作成します。 そして、私たちの場合、キーは特定のWebサイトにアクセスしたユーザーの数によるものであることが判明しました。 日中は5万人に達する可能性があります。 また、いずれかのノードで崩壊が発生したか、単に過負荷になった場合、そのユーザーは他のノードで「継承」できます。

そこに何があるのでしょうか？ ただし、2つのポイントがあります。

1. アップストリームノード（クラスターノード）には、アクティブユーザーの合計数に等しい非常に多くの接続がありました。送信されたメッセージについては、最初にWebサイトの関心のあるキューを検索する必要がありました。 ユーザーの増加に伴い、これはパフォーマンスに影響するだけでした。
2. そして、ケーキのチェリーは同期です。 Webサイトが再起動から戻った、または失われたアップストリームへの接続を復元した後、フェデレーションプラグインは同期プロセスを開始しました。これには通常数分かかりました。 この間、クラスターノードのプロセッサ負荷が高く、メッセージ配信に遅延が発生しました。

代わりに、別の同様のプラグインであるShovelを詳しく調べることにしました。 その目的は、不必要なトラブルなしに（あるノードのキュー間で）、あるノードから別のノードにデータを転送することです。

交換時には、ネットワーク負荷が増加することが予想されていましたが、CPUコストの削減とレイテンシーの削減には価値がありました。 その結果、稼働中のシステムへの移行中に、数ギガバイトのメモリによるクラスターノードの急激な「重量損失」に気付き、Webサイトをリロードしても追加の負荷は発生しませんでした。

レイヤー間の多数のリンク

ユーザーとサービスの数の増加に加えて、アラートシステムが提供するはずのWebサイトの数も増加しています。 最初にそれらが3つあった場合、その後この数は10に増加し、誰も停止しませんでした。 オブジェクトの構造の最後の図からわかるように、現在の実装では、接続の数は提供されるサイトの数に依存し、さらに3倍（公開領域の数）に依存します。 さらに、Shovelプラグインは（フェデレーションができるように）ポリシーを介して機能しないため、パブリケーションエリアの各エクスチェンジャーを作成した後、すべての接続を手動で作成する必要があります（これはユーザーが接続したときに実行された同じスクリプトによって実行されました）。 この依存関係を取り除き、すべてのサイトに1つの接続を使用することは可能ですか？ -必要です！

私たちはすぐに別のタイプの「箱から出してすぐに」交換機を思い出しました-ヘッダー。 なぜ彼はここのアシスタントではないのですか？ その動作の原則は、接続を作成すると、特定のパラメーターセットとその値が設定され、メッセージヘッダーにも示され、一致する場合は接続がトリガーされるというものです。 この例では、サイト名と公開エリアという2つのパラメーターがありました。 たとえば、 site = "online.sbis.ru"およびscope = "user"です。 このペアは、通信用と通信用に設定されています。 停止したのは、そのような交換機の速度が他の交換機と比べて遅いという文書の追記だけでした。 しかし、彼とのつながりはほとんどなく、メッセージの数はそれほど多くありませんでした。 テストは、彼が負荷に非常に精力的に対処していることを示しました。 私たちはそれに切り替えることにしました。

テストの期間が不十分だったため、これがこの規模の最初の失敗でした。 交換機の動作は非常に奇妙で、最初の数時間はすべて正常に機能し、処理速度はゆっくりとゼロに低下しましたが、接続の数はまったく変化しませんでした。 この交換機の作業を実行する外部スクリプトをすばやく作成する必要がありました。 悲しみ。

それともあなたの交換機？

RabbitMQを使用してさまざまな問題を解決する過程で、Erlang言語に出くわすことがよくありました。今度は、Erlang言語をよりよく勉強するときです。 判明したように、RabbitMQの交換機の実装はそれほど難しくありません。 そしてその瞬間から、私たちはこのブローカーと仕事をする新しい時代を始めました。

標準交換機（ fanout 、 direct ）のコードを調べた後、ルート関数RabbitMQがメッセージのコピーが送信されるリソースのリストを待っていることが明らかになりました。 標準交換機は、リンクの内部データベースからこのリソースのリストを抽出します。 ただし、この場合、次の交換機の名前は通常、着信メッセージに基づいて計算されます。

次に、すべてをRabbitMQのプラグインとして配置する必要がありますが、これも非常に単純です（RabbitMQ 3.5のバージョンがあり、もう少し複雑でした）。 その結果、ベースにサンプルが不足しているため、すべてがファンアウトよりも速く動作し、さらには高速に動作しました。 かっこいい！ ところで、交換機のタイプはsbis-ep（エントリポイント）と呼ばれていました。

大きなメッセージの問題

ある時点で、ウェブサイトは記憶不足のために「ハエのように死ぬ」ようになりました。 rabbitmq_tracingプラグインを使用して、親切な人々が通知システムから何らかのCDNを整理し、1つのメッセージで20 MBのデータを送信したことを理解することができました。 しかし、彼らの断絶は喜びを追加しませんでした-落下は続いた（より低い頻度で）。 最大のメッセージは400 Kbのままでした。 単純なイベントの場合、これも非常に多くなりますが、システムがこれで死ぬことはありません。 Erlangのデバッグツールを知った最初の夜は生産的であり、朝までにはすでにヒーリングパッチが公開されていました。 SockJSはxmerl_ucsモジュールを使用してUTF8を操作し、すべてのバイナリをリストに変換して処理したことが判明しました。 その結果、処理中に400キロバイトのメッセージが16 MBを超えるメモリを消費したという事実につながりました。 そして、そのようなメッセージは一度に複数のユーザーに送信されました。 パッチでは、 UTF8 での作業はバイナリ文字列のみを使用して完全にやり直され、その後、ドロップは認識されませんでした。 眠りにつくことができます。

行方不明者を祝う

RabbitMQにはエクスチェンジャーのExpireパラメーターがない（キューの場合のみ）という事実により、別の不快な問題が形成されました。 ユーザーが来たとき、彼のためにすべてが作成されましたが、彼が去ったとき、リソースは削除されませんでした。 予防目的で、私は定期的にウェブサイトを過負荷にしなければなりませんでした。 私たち自身の交換機の最初の経験はプラスの波を残しました-TTLサポートを持つ別のものはこれをすることができますか？ わあ！ しかし、それが個人ユーザーエクスチェンジャー（まだファンアウトがある）であり、作成されると、それが依存する他のすべてのオブジェクトも作成しますか？ 今、これはかなり大人です！ sbis-user exchangeと呼びます 。

上流の交換機と接続パラメーターの名前は、ユーザーとクライアントの識別子、およびユーザーが作業しているサイトの名前に依存します。 新しいエクスチェンジャーの作成中に、それらをパラメーター（引数）に渡すことができます。 さて、パラメータを介して孤立交換器の寿命を制御する機能を追加しましょう。

交換機と接続を作成するには-行う作業はあまりありません。既存のRabbitMQ関数を呼び出すだけで幸福が得られます。 しかし、TTLメカニズムを編成するには、依然として一生懸命働く必要があります。 実際には、交換機には独自のErlangプロセスがないため、定期的に現在のユーザー数を確認して自分自身を削除することができます。 したがって、交換モジュールの実装に加えて、全員を監視する独自のプロセスも必要です。 これを行うには、Erlangアプリケーション（アプリケーション）を作成する必要があります-ブローカー用の新しいプラグインの開始から開始し、実際には1つのワークフローが含まれます。

孤立した交換機を削除するプロセスは、単純に実装できます。定期的にすべての交換機のリストを取得します。 それぞれについて、接続の数を取得します。 存在しない場合は、リストに追加して削除し、適切な間隔で削除します。 しかし、特に数万または数十万の交換機がある場合、このアプローチは大きな負荷を生み出します。 エクスチェンジャーのAPIには、リンクの作成または削除時にブローカーが呼び出す関数が含まれています。 それらは、私たちが最良の決定を下すのに役立つものです。 新しい接続を作成するときは、常に孤児のリストから交換機を削除し、接続を削除するときに、他の接続がなければそこに置きます。

最後の質問：リスト内の交換機の存在を確認し、そこからそれを削除し、最低コストですでに期限切れの交換機を探すために、どの構造を使用する必要がありますか？ いくつかのテストの後、2つの標準構造の組み合わせに決めました。

• gb_treeは、最小値（現在の時間と比較される有効期限が最も短い交換機）、キーとしての{ExpireTime、Exchange}構造、および値が0（使用されていない）を見つけるのに適しています。
• マップ （Erlang / OTP 18で既に機能している）は、 Exchangeキーとして、またExpireTimeの値として、存在を確認し、時間を見つけるために使用されます（最初の構造から削除するため）。

さらに、同時に削除される交換機の数は、ピーク負荷が発生しないように制限されていました。 これで、2番目のプラグインはユーザーの利益のために機能する準備ができました。

希望は叶いましたか？

はい、そうです。 さらにシンプルで美しく、信頼性の高い構造が得られます。

図 11.独自の交換機sbis-epおよびsbis-userを使用したスキーム内のオブジェクトの構造。

クラスターノードには、ファンアウトタイプのエクスチェンジャーが1つだけあり、Webノードとクラスターの間には接続も1つしかありません。スキームはサイトの数に依存せず、ユーザーがサイトに不在になると、ユーザーに割り当てられたリソースはクリアされます。 ただのおとぎ話、あなたは他に何を夢見ることができますか？ この形式では、システムはすでに15のWebサイトで最大60万の同時接続を処理し、1秒あたり30万の送信メッセージのピークがありました。 しかし、それは彼女の限界ではありませんでした。

アラートX：無限大

ユーザーの成長が止まることはなかったので、新しい冒険の時間です。

内部のすべて

ほとんどすべてのサービス操作はすでにRabbitMQ内で動作します（遅延が最小限に抑えられます）。 外に残されている唯一のものは、個人ユーザーエクスチェンジャーの作成です。 しかし、結局のところ、RabbitMQのプラグインとして私たちを悩ませたり、実装したりするものはありませんか？ つまり、すでに実行中のカウボーイWebサーバーを使用して、次の新しいプラグインからエクスチェンジャーの作成を開始できます。

/ infoへのリクエストからのブラウザバージョンのSockJSはwebsocketフラグのみを必要とし、サーバー側はこのリクエストなしで実行できるため、このリクエストをインターセプトするだけでなく、完全に処理することもできます。 つまり、このリソースのリクエストはプラグインに送信され、他のすべてのリクエストは引き続きWeb STOMPアダプターRabbitMQに送信されます。

全員に十分な64Kポート

多数のWebサイトを作成しなければならなかった問題の1つは、同じホスト上のNginxとRabbitMQの間のオープン接続の数の制限でした。 Nginxは発信接続アドレスを指定できませんでした。プラグインのWeb STOMP側では、複数のポートを指定することはできませんでした。 さらに、現在Webソケットに住んでいるユーザーの数と、XHRポーリングとストリーミングのユーザー数を確認する必要がありました。 Nginxはそのようなことを伝えることができず、事態を悪化させる必要がありました。

次に、HAProxyが助けになりました。

• URL（Nginxなど）に応じて、目的のポートにリクエストを送信します。
• 1つのポートへの多数の接続を作成します（ソースオプションを使用すると、発信アドレスを指定できます。ローカルには多くの接続アドレス127.0.0.0/8があります）。
• 接続の最大数を制限して、大量のリクエストの流入からサービスを保護し、より安定させます。
• サービスに関する多くの統計情報（特定のトランスポートの現在の接続数、新しい接続の速度、応答時間、応答コード、キュー内の要求の待機時間など）を受け取ります。
• リソースを節約します。 HAProxyは、同じ接続数で3倍少ないメモリを消費し、プロセッサの負荷は3〜5％でした。

キャッシュして圧縮してください！

標準パッケージには、RabbitMQには便利なWebベースの管理コンソール-管理プラグインが含まれています。 保守スタッフは、ブローカーで何が起こっているかを迅速に評価できます。 また、そのAPIを通じて、ブローカーの作業に関する統計を収集します。 問題は、キャッシュメカニズムがなく、リクエストごとにすべての情報を収集することです。 RabbitMQ 3.5にはページネーションがなく、リソースの完全なリストが常に提供されます。 また、デフォルトでは、5秒後に自動更新データがオンになります。 40,000のキューを使用すると、それらに関するすべての情報が10秒間収集され、多くのプロセッサ時間が消費され、重量が15〜20 MB（ネットワーク負荷）になります。 つまり、デフォルトでは、ユーザーは常にリクエストを行います。 そして、複数のユーザーがいる場合はどうなりますか？ 覚えておくのは怖いです。

HAProxyに切り替える前に、Nginxでキャッシュを作成できましたが、少しデータがあり、メモリキャッシュがより効率的であるため、Varnishを試すことにしました。 これらには、すべてのコンテンツのキャッシュ（ beresp.ttl ）が15秒間（静的な場合は数日間）、キャッシュの有効性（ beresp.grace ）およびデータ圧縮（ beresp.do_gzip ）が含まれていました。

その結果、以下を受け取りました 。

• リソースへの要求は、1分間に4回まで実行されます。
• 圧縮されたデータの重量は、元のデータの約50倍（270 KB対13 MB）です。
• データは数ミリ秒で与えられました（猶予期間を超えない場合、Varnishは期限切れのデータを出力し、バックグラウンドでバックエンドにそれらを更新するよう要求します-見た目は素晴らしいです）;
• ブローカーの負荷はユーザー数に依存せず、計算できます。

私の光、鏡、言って

Webサイトを最適化し、それらを適切にスケーリングする方法を学習した後、クラスターノードの下位層で既に問題が発生し始めました。 1秒あたり数十万から数十万の送信メッセージは、主にクライアント領域での発行によるものでした。 たとえば、当社では5,000人がオンラインにいる場合があります。 誰かが1秒間に60件のメッセージを会社全体に送信する場合、最終的には60x5000 = 30万/秒になります。 クラスターノードの場合、最初の要因は重要です-RabbitMQのバックエンドを持つ公開されたメッセージの数。 そして、ある時点で、それも成長し始めました。 18のWebサイトにサービスを提供する完全なレプリケーション（ミラーリング）を備えた2つのノードは、1秒あたり2000件を超えるパブリケーションを処理できず、適切にスケーリングする方法がないことが判明しました。 バックエンドの数も増加し、それらからの接続の数はマークの2500を超えました。スキームには変更が必要でした。

RabbitMQは非常に安定していることが証明されており、通知システムは「小数点以下9桁」を必要としなかったため、クラスターを放棄することにしました。 さらに、バックエンドはノードのリストを操作し、別のノードの1つが失敗した場合に別のノードへの要求を繰り返す方法をすでに知っていました。 既存のすべてのダウンストリームブローカーにすぐに接続するようWebサイトに教えることは残っていますが、これはまったく問題ではありません。 新しいタイプのノードにはルートという名前が付けられ（将来、ルーティングメカニズムもそれに割り当てられる）、それに切り替えた後、スキームは次のようになり始めました。

図 12.非クラスターノードの2層スキーム。

これで、必要に応じて両方のレイヤーをスケーリングできます。 クラスタノードと同じ構成の2つのルートノードを持つオプションは、既に1秒あたり約1万3千のパブリケーションを処理しましたが、これはそれらの制限ではありません！

新しいチャットメッセージが1つあります

クラウドアプリケーションでのチャットの開発と普及に伴い、別の問題が発生しました。ユーザーのリストに同一のメッセージを大量に送信することです。 チャットに数十人がいる場合、すべてが十分に迅速に行われます。 ただし、数百人のユーザーがいる場合、RabbitMQへの投稿には数秒かかる可能性があります（また、多くの負荷が発生します）。 受信者のリストにメッセージを効率的に送信できるメカニズムが必要でした。

AMQPメッセージのヘッダー（サイトのリストを送信する方法）を介してこのリストを送信できることがすぐに明らかになりました。 問題は、Webサイトで処理するタスクでした。 最初のソリューションは、個人ユーザーエクスチェンジャー（sbis-user）の名前がサイト名とユーザーIDで構成されるという事実によって使用されました。このようなケースを処理するためにエクスチェンジャーsbis-epを変更できます。 つまり、メッセージはサイトとユーザーのリストとともに到着し、交換コードはこれらのリストを乗算して結果をRabbitMQに渡します。RabbitMQはデータベース内の既存のリストを検索し、メッセージのコピーを送信します。 このオプションは簡単かつ迅速に実装できましたが、いくつかの欠点がありました。

• 1つの公開領域（ユーザー）のみで動作します。
• 1つのサイトのみが全ユーザーの約5％しかサービスを提供しないため、非常に冗長なリストが作成されます。 また、リストにある3つのサイトのうち、1つのサイトにしか住んでいません。

すべての欠点にもかかわらず、ソリューションは職場に来て、大きなチャットルームで通信するときの負荷を減らすことができました。 追加された唯一のものは、50ユニットのユーザーリストの長さに対する人為的な制限でした。 そうしないと、バックエンドが数ミリ秒でメッセージを公開したときに大きな負荷係数が判明し、その後、すべてのWebサイトが数秒間動作しました（大きな責任が生じますが、すべての開発者が2番目のことを知っているわけではありません）

正しくやろう

最後の問題の程度がわずかに減少したので、適切な方法でメカニズムを実装できます。 大きなリストを生成するのではなく、既存のユーザーのデータベースを作成して検索するのが最も正しいでしょう。 RabbitMQには、ビルトインMnesiaデータベースにすべての交換機のリストを持つテーブルが既にあります。 残念ながら、検索のために最適な方法で編成されていないため、独自の処理を行う必要があります。 新しいテーブルはタイプセットになり、プライマリキーとして交換機を含み、検索に必要な他のすべての識別子が続きます（必要に応じてリストを展開できます）。

メッセージを処理するとき、必要な値のリストを含むQLCリクエストを作成し、実行のためにデータベースに渡します。 結果は、すべての条件を満たす既存の交換機のリストになります。 すべてが正常に機能しますが、加速が予想されるものの、すぐにパフォーマンスが低下しました。 何らかの形で状況を修正する必要があります。

主キーを検索すると、ほぼ瞬時に実行されます。テーブルの他の列を検索するには、そのすべてのレコードを検索する必要があります。 しかし、Mnesiaにはセカンダリインデックスがあります。それらをオンにして、速度を楽しみましょう。 そして、別の失望が私たちを待っています、何も変わっていません。 50ユーザーと3サイトでの検索は30ミリ秒で完了します。 ここでの問題は、2つのフィールド（識別子とサイト）を同時に表示していることであり、インデックスはあまり役に立ちません（ソースデータは約10 MBしかないため）。 救助のために、内部データベースにはデータ型に制限がなく、フィールド値として{Site、Ident}タプルを挿入できるという事実があります。 その後、検索は1つのフィールドのみを通過します。 セカンダリインデックスは完全に機能していますが、60ミリ秒でさらに悪化しました。 実際、2つのリストを1つの大きなリストに展開する必要があり、検索クエリが巨大になりました。

さて、反対側に行きましょう。 Mnesiaには、インデックスで値を検索する機能があり、最短時間で動作しますが、単一の値で動作するため、ループを追加する必要があります。 さらに、私たちはかなりの時間をかけて、Mnesiaのダーティオペレーションを使用して生産性を高めます。

各インデックス検索には約6マイクロ秒かかります。つまり、同じリストサイズでは、1ミリ秒未満で回答が得られます（サイズが小さいほど、ゲインはさらに大きくなります）。 この結果はすでに私たちに合っています。

最近のフィールドサマリー

すべての変更の後、次のオブジェクトの構造を取得します。

図 13.オブジェクトの最終構造。

最後のノード接続スキームとブローカー内のオブジェクトの構造により、サービスを提供できました

21のWebサイトで100万の同時Webソケット（110万ありましたが、監視の履歴の集計では、この図からの痕跡は残りませんでした）：

図 14.稼働週あたりの接続の総数。

また、合計トラフィックが4.2ギガビット/秒を超えるユーザーに、1秒あたりほぼ130万のメッセージを送信することに貢献しました。

図 15.ピーク負荷の処理：1秒あたりのメッセージ数とメガバイトのトラフィック。

おわりに

RabbitMQを使用してユーザー通知システムの構築を開始した時点では、システムが最終的にどれだけ大きくなるか、および処理できるメッセージの量がどれほどかはわかりませんでした。 RabbitMQはときどき気まぐれでしたが、（特にプラグインを書くためにカンフーを習得した場合）開発コストを抑えてすべてを実装し、実装時間を短縮することができました。

私たちは経験をhabrasocietyと共有できることを嬉しく思います。すべての質問に喜んでお答えします。

記事の著者セルゲイ・ヤーキン