1

次の構成を使用して、ワークロードに基づいてスケールアップ/ダウンできるアクター システムを構築しようとしています。

  akka {
        actor {
          serializers {
            wire = "Akka.Serialization.WireSerializer, Akka.Serialization.Wire"
          }
          serialization-bindings {
            "System.Object" = wire
          }              
        deployment {
          /analysis {
            router = round-robin-pool
            routees.paths = ["/user/analysis"]
            resizer {
                enabled = on
                lower-bound = 1
                upper-bound = 20
            }
          }
        }
        }
      }

私のメイン ループでは、3000 のメッセージを作成し、処理のためにアクターにプッシュしています。

        var runAnalysisProps = Props.Create<RunAnalysisActor>().WithRouter(FromConfig.Instance);
        //var runAnalysisProps = Props.Create<RunAnalysisActor>().WithRouter(new RoundRobinPool(0, new DefaultResizer(lower:1, upper: 10 ))); **This do not help either **
        var analysisRef = RunAnalysisActorSystem.ActorOf(runAnalysisProps, "analysis");

        Logger.LogMessage("Started Posting all messages", ConsoleColor.Blue);
        for (int i = 0; i < 3000; i++)
        {
            analysisRef.Tell(new AnalysisMessage(100 + i, $"FIC{100+ i}", $"c:\\temp\\fic{100 + i}.dat"));

        }
        Logger.LogMessage("Completed posting all messages", ConsoleColor.Blue);

上記の for ループに何か問題があると思いますか?私には、ルーターではなく 1 つのアクター ref にすべてのメッセージを送信しているように見えます。

私のReceiveActor非常に単純なロギングアプリケーションです

public RunAnalysisActor()
        {
            Receive<AnalysisMessage>((message)=> {
                Logger.LogMessage($"Analysis Started => Id: {message.AnalysisId}, Loop: {message.LoopName}, File:{message.FilePath}");
                var waitTime = rnd.Next(5) * 100;
                //Logger.LogMessage($"Waiting => {waitTime}");
                Thread.Sleep(10);
                Logger.LogMessage($"Analysis Completed=> AId: {message.AnalysisId}, Loop: {message.LoopName}, File:{message.FilePath}");

            });
        }

また、開始および停止されたアクターの数を確認するためにAroundPreStart、およびをオーバーライドしました。PostStopしかし、アプリケーションが起動すると、アクターの数が 2 つしか作成されないことがわかります。lower-boundこれにより、デフォルトの 1 に基づいてより多くのアクターが作成されることを期待していましたが、pressure-threshold発生していないことがわかります。処理するメッセージはありません。

誰かが私がここで間違っていることを理解するのを手伝ってくれますか? - ありがとう

更新 1:

私は実際の問題を理解したと思います。メッセージを投稿している間、ロジックを変更して展開構成にパスを追加し、関数をroutees.paths = ["/user/analysis"]使用して、このようなものAskを使用して正しい参照を取得する必要があります analysisRef.Ask<Routees>。

        RunAnalysisActorSystem.Scheduler.Advanced.ScheduleRepeatedly(
            TimeSpan.FromMilliseconds(500),
            TimeSpan.FromMilliseconds(500),
            () => {
                if (analysisRef.Ask<Routees>(new GetRoutees()).Result.Members.Any())
                {
                    var i = rand.Next();
                    analysisRef.Tell(new AnalysisMessage(100 + i, $"FIC{100 + i}", $"c:\\temp\\fic{100 + i}.dat"));
                }
            }
        );
4

0 に答える 0