問題タブ [akka-persistence]

Akka アプリケーションを作成しており、Typsafe が提供する Akka Cluster 機能を使用したいと考えています。Akka Cluster、Akka Cluster Singleton、Akka Cluster Sharding の主な違いを理解しようとしています。私には、彼らは同様のものを提供しているように見えます。

私のシナリオでは、次のようなクラスターが必要です。

• どのマシンにも住むことができるステートレスなアクター (私はそれらのプールを用意します)
• 特定のマシンに住んでもらいたいステートレス アクター
• ステートフル アクターは任意のマシンに存在できますが、クラスター全体で 1 つのインスタンスだけが必要であり、アクターが停止した場合に以前の状態でアクターを再起動できる必要があります。

私が読んだ内容に基づいて、第 1 種には Akka Cluster、第 2 種には Akka Cluster Singleton、第 3 種には Akka Cluster Sharding が必要です。私の仮定は正しいですか？いくつかのガイダンスとオンラインの例を探しています。

多数のオリジナルおよび派生データ ポイントを処理する EventSourcing アプリケーションを実装しています。つまり、コマンドを受け入れる集約ルートPersistentActorとして機能しています。

これらのコマンドが検証されると、永続化されて状態を更新するイベントが生成されます。

では、PersistentViewこれらのイベントをリッスンし、派生値を計算します。

しかし、この再計算自体は、他のビューが依存する可能性のある新しい値であるため、新しいコマンドを集約ルートに送り返して新しい値を処理する必要があります。ビュー。

ビューがイベントまたはコマンドを生成することは許容されますか? ビューの責任は、イベントやコマンドを生成するのではなく、イベントに基づいて状態を更新することだと思います。また、イベントが到着する順序は、再生中にブロードキャストされる新しいイベントに影響を与える可能性があります。

イベントの代わりにコマンドを生成する必要がありますか? コマンドは初期値を更新していたので、生成されたすべての派生値は、単にコマンドが処理された結果であるイベントであると主張できますが、それらは分散された方法で生成され、集約ルートによって直接生成されるわけではありません。

私は、これらのアクターをつなぎ合わせるために使用できる Akka の Reactive Streams を見てきました。 -part-i-of-ii/ . その投稿で、ジョナサンは次のように述べています。

Sagas はイベントをリッスンしてコマンドをディスパッチし、集合体はコマンドを受信して​​イベントを発行します。

これらすべてのアクターを FSM として実装することも賢明なアプローチのように思えます: 関連するイベントを 5 秒待ち、すべてを再計算し、コマンドをディスパッチし、イベントを 5 秒待ちます。

物事をもう少し興味深いものにするために、値のストリームは順不同で不完全である可能性がありますが、特定の時点で派生値を生成する必要があります。したがって、値 A と B を受け取った場合:

• A1、B1、B2、A2、B3、A4、B4

派生値を生成する必要があります D:

• D1(A1*B1)、D2(B2*A2)、D3(B3*A2、A3なし)、D4(A4*B4)

これは、順序を追跡しなければならないことを意味し、欠落している値が入ってきた場合は派生値を再発行することがあります.

ありがとう！

次のコードを使用しています。

Akka Persistence を使用してこれらのイベントを永続化し、アクターの状態を再構成します。ただし、呼び出すpersist(ev)と、次の例外が発生します。

Javaシリアライザーは、派生したジェネリックケースクラスを処理できないようですか? 試してみましsealed trait Event extends Serializableたが、あまり役に立ちませんでした。

とにかくこれに対する解決策はありますか？多分他のシリアライザー？または、イベント/コマンド/メソッドの一般的なセットを作成する別のアプローチですか?

UntypedPersistentActor を拡張するオブジェクトに送信された状態変更コマンドの適用が成功したことを呼び出し元に通知したいと思います。

状態の変更を永続化していない場合は、変更に影響を与えるクロージャを含む Future を返し、呼び出し元にそれを Await させます。

Akka の "persist()" 呼び出しは、私の理解が正しければ、非同期で実行されるクロージャーを渡すことを強制します。それ自体が persist() を呼び出す Future を返した場合、呼び出し元に、後のアプリケーションのために変更を正常にキューに入れたことを伝えるだけです...変更が適用されたこと、またはそのアプリケーションが成功することでさえありません。

1 つの呼び出し元に関する保証された順序付けは、変更がいつ適用されたかを知る必要性を少し満たしていると思いますが、変更が失敗した場合はどうなるでしょうか? 失敗のために呼び出し先が再起動された場合、呼び出し元のメッセージはドロップされ、状態の変更は発生せず、呼び出し元は認識されません。エラーを呼び出し元に返し、そこに表示する方がきれいなようです。

これを達成する良い方法はありますか？

実際にはアクターであるSpringサービスがあり、情報を受け取りましたが、インジェクションが失敗したため、別のSpringサービスに渡すことができません。

そしてセカンドサービス

春の設定、これはakkaの春の例からのものです

では、どうすれば別のSpringサービスを注入できますか?????????

akka-persistence に scala api を使用して、ツリーに編成されたアクター インスタンスのグループを永続化しています。ツリー内の各ノードは永続的なアクターであり、「ルート」ノードからそのノードへのパスに基づいて名前が付けられます。名前には persistenceId が設定されます。たとえば、ルート ノード アクターには persistenceId 'root' があります。次にダウンしたノードには、persistenceId 'root-europe' があります。別のアクターが persistenceId 'root-europe-italy' を持っている可能性があります。

各アクターの状態には、その子の名前のリストが含まれています。たとえば、「ルート」アクターは、その状態の一部として「ヨーロッパ」、「アジア」などのリストを維持します。

このシステムのスナップショットを実装しました。ルートがスナップショットをトリガーされると、スナップショットが実行され、各子に同じことを行うように指示されます。

この問題は、スナップショット リカバリ中に発生します。persistenceId = 'root' でアクターを再作成すると (名前をコンストラクター パラメーターとして渡すことによって)、そのアクターが受け取る SnapshotOffer イベントが正しくありません。たとえば、「root-europe-italy....」です。これは、persistenceId が回復されるアクターの状態を識別するという、永続化の契約の矛盾のようです。ノード アクター (「italy-europe-root」など) の persistenceId を逆にすることでこの問題を回避したので、これは永続モジュールによってファイルが取得される方法に関連しているようです。最初に他のアプローチを試したことに注意してください。たとえば、ノード名の間にさまざまなセパレーターを使用したり、セパレーターをまったく使用したりしませんでした。

他の誰かがこの問題を経験したことがありますか? akka-persistence 開発者は、なぜこれが起こったのかを理解するのに役立ちますか?

ところで: 今のところ、組み込みのファイルベースのスナップショット ストレージを使用しています。

ありがとう。

次のような Akka Persistence システムから始めたとします。

そして、いつか私はこれを次のように変更します：

新しいシステムをデプロイした後、 のインスタンスがMyProcessorその状態を復元しようとすると、ジャーナリングされたメッセージは以前のケース クラスになります。後者のタイプを想定しているため、 がスローされOnReplayFailure、プロセッサが役に立たなくなります。y質問: 欠席が等しい(または何でも)と仮定した場合、0これを克服するためのベスト プラクティスはありますか? たとえば、 を使用して、implicit回復時に前者のメッセージから後者に変換することはありますか?

Akka SyncWriteJournal現在、 API との接続を実装するための独自のプラグインを作成中HSQLDBです。

問題は、メソッドの要件を理解していないことですdoAsyncReplayMessages。future を返す必要があり、すべてのメッセージは によって呼び出される必要があると述べていますreplayCallback。

メッセージのリストを返すクエリがあるとしましょう: List<Message> messages. replayCallbackを使用する方法と、Futureそのリストを使用してメソッドを正しく実装する方法の最小限の例 (説明付き) を誰かが提供できますか? どのようreplayCallbackにFuture連携し、メソッドによって何を返す必要がありdoAsyncReplayMessagesますか?

ありがとう！

-編集-

いくつかのコメントの助けを借りて、完全ではないが提案されたアイデアを組み込んだ実装を作成しました。

お気づきかもしれませんが、私がまだ見逃しているいくつかの重要な概念が見落とされています。PersistentImpl には、静止している 1 つの引数が必要Seq<String> confirmですnull。そしておそらくもっと重要なことはnull、将来は戻り値の型として期待Voidされているため、私は戻ります。それをどのように実装するかわかりません。現在、NPE をスローします。

私は akka の永続化、akka のリモート処理、akka のシャーディングを使用してアプリに取り組んでいます。DDD/CQRS アプローチを採用しています。

以下はアプリの構成です。

アプリ全体に対して 1 つのアクター システムを作成しています。間違っている場合は訂正してください。デフォルトのディスパッチャーは、デフォルトで、少なくとも 8 つのスレッドを持つ fork-join スレッド プールを使用します。

クライアントからの何百万ものリクエストをサポートする必要があります。akka の永続性/シャーディング/クラスタリングを使用する場合にアクター システムのディスパッチャーを調整する方法、推奨されるディスパッチャーのタイプは何ですか? スレッド プールごとのスレッドを計算する方法 コアを考慮する必要がありますか? OS のタイプを考慮する必要がありますか? JVM の構成は問題になりますか? また、app conf でカスタム ディスパッチャ構成を指定する方法。

基本的に、アクター システムごとにディスパッチャーの実行を決定する基準を理解したいと考えています。

良好なパフォーマンスを得るために akka の永続性/クラスター/シャードの種類のアプリケーションを調整するために必要な他の構成はありますか?

また、akka persistence/cluster/shard の種類のアプリケーションをチューニングするための推奨事項は、非常に役立ちます。

-プラキャットMM