playframework - クラスター内の Akka アクターの検出

Question

私は最近、Akka とアクターベースのシステムの概念に頭を悩ませようとしています。Akka の基礎についてはかなりよく理解できましたが、クラスタリングとリモート アクターに関してはまだいくつか苦労しています。

Play Framework 2.0 に付属する WebSocket チャットの例を使用して、この問題を説明しようとします。WebSocket を保持し、現在接続しているユーザーのリストを保持するアクターがあります。アクターは基本的に、技術的にも論理的にもチャット ルームを表します。これは、単一のサーバーで単一のチャット ルームが実行されている限り、問題なく機能します。

ここで、サーバーのクラスター (単一ノードが追加または削除された状態) で実行されている多くの動的チャット ルーム (新しいルームはいつでも開閉できる) について話しているときに、この例をどのように拡張する必要があるかを理解しようとしています。現在の需要による）。このような場合、ユーザー A はサーバー 1 に接続し、ユーザー B はサーバー 2 に接続できます。両方が同じチャット ルームで話している可能性があります。各サーバーには、イベント (メッセージ) を受信して​​適切なユーザーに公開する WebSocket インスタンスを保持するアクター (チャット ルームごと?) が引き続き存在します。しかし、論理的には、サーバー 1 またはサーバー 2 のいずれかに、現在接続しているユーザー (または同様のタスク) のリストを保持するチャット ルーム アクターは 1 つだけ存在する必要があります。

できれば「純粋な akka」で、ZeroMQ や RabbitMQ などの追加のメッセージング システムを追加せずに、これをどのように達成しますか?

これは私がこれまでに思いついたものです。これが意味をなすかどうか教えてください:

1. ユーザー A がサーバー 1 に接続し、彼の WebSocket を保持するアクターが割り当てられます。
2. アクターは、アクティブなチャット ルームの「チャット ルーム アクター」が接続されたクラスター ノードのいずれかに存在するかどうかを (Router? EventBus? 何か他のものを使用して?) チェックします。そうではないので、何らかの方法で新しいチャット ルーム アクターの作成を要求し、このアクターとの間で将来のチャット メッセージを送受信します。
3. ユーザー B はサーバー 2 に接続し、彼の WebSocket にもアクターが割り当てられます。
4. また、要求されたチャット ルームのアクターがどこかに存在するかどうかを確認し、サーバー 1 で見つけます。
5. サーバー 1 のチャット ルーム アクターは、指定されたチャット ルームのハブとして機能し、すべての「接続された」チャット メンバー アクターにメッセージを送信し、受信したアクターを配布します。

サーバー 2 がダウンした場合、チャット ルーム アクターをサーバー 2 で再作成するか、サーバー 2 に移動する必要がありますが、これは現在の私の主な関心事ではありません。アクターのこの動的な発見が、Akka のツールセットを使用して、さまざまな基本的に独立したマシンにどのように広まったかについて、私は最も疑問に思っています。

私はかなり長い間 Akka のドキュメントを見てきました。もしそうなら、私に教えてください:-)

Answer 1

私は、基本的にチャットルームの例の非常に拡張されたバージョンであるプライベートプロジェクトに取り組んでおり、akka と全体の「分散型」思考にもスタートアップの問題がありました。拡張チャットルームをどのように「解決」したかを説明できます。

追加の構成をあまり行わなくても、簡単に複数回展開できるサーバーが必要でした。開いているすべてのユーザー セッション (ActorRef の単純なシリアル化) およびすべてのチャットルームのストレージとして redis を使用しています。

サーバーには次のアクターがあります。

• WebsocketSession: 1 人のユーザーへの接続を保持し、ユーザーからの要求を処理し、システムからメッセージを転送します。
• ChatroomManager: これは、サーバーのすべてのインスタンスにデプロイされる中央ブロードキャスターです。ユーザーがメッセージをチャットルームに送信したい場合、WebSocketSession-Actor はすべての情報を ChatroomManager-Actor に送信し、チャットルームのすべてのメンバーにメッセージをブロードキャストします。

だからここに私の手順があります：

1. ユーザー A は、新しい WebsocketSession を割り当てるサーバー 1 に接続します。このアクターは、このアクターへの絶対パスを redis に挿入します。
2. ユーザー A はチャットルーム X に参加し、絶対パス (私はこれをユーザー セッションの一意の ID として使用します) を redis に挿入します (各チャットルームには「接続」セットがあります)。
3. ユーザー B がサーバー 2 に接続 -> redis
4. ユーザー B がチャットルーム X に参加 -> redis
5. ユーザー B は次のようにチャットルーム X にメッセージを送信します。ユーザー B は Websocket を介してセッション アクターにメッセージを送信し、セッション アクターは (いくつかのチェックの後) アクター メッセージを ChatroomManager に送信します。このアクターは実際に redis (akka のactorFor-method で使用される絶対パス) からチャットルームのユーザー リストを取得し、各セッション アクターにメッセージを送信します。これらのセッション アクターは、WebSocket に書き込みます。

各 ChatroomManager-actorActorRefで、追加の速度を実現するキャッシングを行います。これは、特にこれらの ChatroomManagers がすべてのチャットルームのリクエストを処理するという点で、あなたのアプローチとは異なると思います。しかし、1 つのチャットルームに 1 人のアクターを配置することは、避けたかった単一障害点です。さらに、これによりさらに多くのメッセージが表示されます。たとえば、次のようになります。

• ユーザー A とユーザー B はサーバー 1 にいます。
• チャットルーム X はサーバー 2 にあります。

ユーザー A がユーザー B と話したい場合、両者はサーバー 1 のチャットルーム アクターを介して通信する必要があります。

さらに、(ラウンド ロビン) ルーターなどの akka の機能を使用して、各システムに ChatroomManager アクターの複数のインスタンスを作成し、多くの要求を処理しました。

シリアライゼーションと redis を組み合わせて akka リモート インフラストラクチャ全体をセットアップするのに数日を費やしています。しかし今では、redis を使用してそこを共有するサーバー アプリケーションの任意の数のインスタンスを作成できますActorRef(ip+port で絶対パスとしてシリアル化されます)。

これはあなたをもう少し助けるかもしれません、そして私は新しい質問を受け付けています（私の英語についてではありません;）。

Answer 2

複数のマシンにまたがってスケールアウトするための鍵は、変更可能な状態をできるだけ分離しておくことです。分散キャッシュを使用してすべてのノードの状態を調整できますが、これにより同期が発生し、多数のノードにスケールアウトするときにボトルネックの問題が発生します。理想的には、チャット ルームのメッセージと参加者について知っている 1 人のアクターが存在する必要があります。

問題の核心は、チャット ルームが単一のマシンで実行されている単一のアクターによって表されている場合、または実際にそのようなルームが存在する場合です。その秘訣は、チャット ルームの名前などの識別子を使用して、特定のチャット ルームに関連するリクエストをルーティングすることです。名前のハッシュを計算し、数に応じて、n 個のボックスから 1 つを選択します。ノードは現在のチャット ルームを認識し、適切なチャット ルーム アクターを安全に検索または作成できます。

Akka でのクラスタリングとスケールアウトについて説明している次のブログ記事をご覧ください。

http://blog.softmemes.com/2012/06/16/clustered-akka-building-akka-2-2-today-part-1/

http://blog.softmemes.com/2012/06/16/clustered-akka-building-akka-2-2-today-part-2/

Answer 3

Zookeeper+Norbert を使用して、稼働しているホストと停止しているホストを確認します。

http://www.ibm.com/developerworks/library/j-zookeeper/

これで、チャットルーム サーバー ファーム内のすべてのノードが、論理クラスター内のすべてのホストを認識できるようになりました。ノードがオフラインになる (またはオンラインになる) と、コールバックを受け取ります。どのノードも、現在のクラスター メンバーのソートされたリストを保持し、チャットルーム ID をハッシュし、リスト サイズで変更して、特定のチャットルームをホストするノードであるリスト内のインデックスを取得できるようになりました。1 を追加して再ハッシュして 2 番目のインデックスを選択し (新しいインデックスを取得するまでループが必要)、2 番目のホストを計算して、冗長性のためにチャットルームの 2 番目のコピーを保持できます。2 つのチャットルーム ホストのそれぞれにチャットルーム アクターがあり、チャットルーム メンバーである各 Websocket アクターにすべてのチャット メッセージを転送します。

これで、アクティブなチャットルーム アクターとカスタム Akka ルーターの両方を介してチャット メッセージを送信できるようになりました。クライアントはメッセージを 1 回送信するだけで、ルーターがハッシュ変更を行い、2 つのリモート チャットルーム アクターに送信します。Twitter スノーフレーク アルゴリズムを使用して、送信されるメッセージの一意の 64 ビット ID を生成します。次のリンクで、コードの nextId() メソッドのアルゴリズムを参照してください。異なるサーバーで衝突する ID が生成されないように、norbert プロパティを使用して datacenterId と workerId を設定できます。

https://github.com/twitter/snowflake/blob/master/src/main/scala/com/twitter/service/snowflake/IdWorker.scala

これで、2 つのアクティブなチャットルーム アクターのそれぞれを介して、すべてのメッセージの 2 つのコピーが各クライアント エンドポイントに送信されます。各 Websocket クライアント アクターで、スノーフレーク ID のビットマスクを解除して、メッセージを送信している datacenterId+workerId 番号を学習し、クラスター内の各ホストから見られる最大のチャット メッセージ番号を追跡します。次に、特定の送信者ホストの特定のクライアントで既に見られたものよりも高くないメッセージを無視します。これにより、2 人のアクティブなチャットルーム アクターを介して着信するメッセージのペアが重複排除されます。

ここまでは順調ですね; いずれかのノードが停止した場合でも、生き残ったチャットルームのコピーを 1 つ失わないという点で、回復力のあるメッセージが得られます。メッセージは、2 番目のチャットルームを介して自動的に途切れることなく流れます。

次に、ノードがクラスターから脱落したり、クラスターに追加されたりすることに対処する必要があります。各ノード内で nobert コールバックを取得して、クラスター メンバーシップの変更について通知します。このコールバックで、新しいメンバーシップ リストと現在のホスト名を示すカスタム ルーター経由で akka メッセージを送信できます。現在のホストのカスタム ルーターはそのメッセージを確認し、その状態を更新して新しいクラスター メンバーシップを認識し、特定のチャットルーム トラフィックを送信するための新しいノードのペアを計算します。この新しいクラスタ メンバシップの確認は、ルータによってすべてのノードに送信されるため、すべてのサーバがいつメンバシップの変更に追いつき、メッセージを正しく送信できるようになったかを追跡できます。

生き残ったチャットルームは、メンバーシップの変更後もアクティブである可能性があります。その場合、すべてのノードのすべてのルーターは通常どおり送信を続けますが、新しい 2 番目のチャットルーム ホストにも投機的にメッセージを送信します。その 2 番目のチャットルームはまだ稼働していない可能性がありますが、メッセージはサバイバー経由で流れるので問題ありません。メンバーシップの変更後に生き残ったチャットルームがアクティブでなくなった場合、すべてのホストのすべてのルーターが最初に 3 つのホストに送信します。生存者と2つの新しいノード。akka デス ウォッチ メカニズムを使用すると、すべてのノードが最終的に生き残ったチャットルームのシャットダウンを確認して、2 つのホストを介したチャット トラフィックのルーティングに戻ることができます。

次に、状況に応じて、生き残ったサーバーから 1 つまたは 2 つの新しいホストにチャットルームを移行する必要があります。生き残ったチャットルームのアクターは、ある時点で、新しいクラスター メンバーシップについて知らせるメッセージを受け取ります。チャットルームのメンバーシップのコピーを新しいノードに送信することから始めます。このメッセージは、新しいノードに正しいメンバーシップを持つチャットルーム アクターの新しいコピーを作成します。生存者が、チャットルームを保持する必要がある 2 つのノードのいずれかでなくなった場合、廃止モードになります。廃止モードでは、新しいプライマリ ノードとセカンダリ ノードへのメッセージのみを転送し、チャットルーム メンバーには転送しません。Akka メッセージ転送はこれに最適です。

廃止チャットルームは、各ノードからの norbert クラスター メンバーシップ確認メッセージをリッスンします。最終的に、クラスター内のすべてのノードが新しいクラスター メンバーシップを認識したことがわかります。その後、転送するメッセージをこれ以上受信しないことがわかります。その後、それは自分自身を殺すことができます。Akka ホットスワップは、デコミッション動作を実装するのに最適です。

ここまでは順調ですね; ノードがクラッシュしてもメッセージが失われない、回復力のあるメッセージング設定があります。クラスター メンバーシップが変更された時点で、チャットルームを新しいノードにコピーするためのノード内トラフィックのスパイクが発生します。また、どのチャットルームが 2 つのどのサーバーを移動したかにすべてのサーバーが追いつくまで、ノードへのメッセージのノード内転送の混乱が残ります。システムをスケールアップしたい場合は、ユーザー トラフィックが低下するまで待ってから、新しいノードをオンにするだけです。チャットルームは、新しいノード間で自動的に再分配されます。

上記の説明は、次の論文を読み、それを akka の概念に翻訳したことに基づいています。

https://www.dropbox.com/s/iihpq9bjcfver07/VLDB-Paper.pdf