Akka Cluster with Spring BOOT

        for(int i=0;i<testCount;i++){
            clusterActor.tell(testMessage + i , ActorRef.noSender());
        }       

version: '3.5'
services:
  light-house:
    image: registry.webnori.com/javalabs-lighthouse:dev        
    ports:
      - "8081:8080"    
    environment:      
      TZ: Asia/Seoul
      akka.role: seed
      akka.seed: akka://ClusterSystem@light-house:12000
      akka.hostname: light-house
      akka.hostport: 12000
      akka.cluster-config: cluster.conf
    networks:
      - mynet      
  work-node1:
    image: registry.webnori.com/javalabs-api:dev        
    ports:
      - "8082:8080"
    depends_on:
      - light-house     
    environment:      
      TZ: Asia/Seoul
      akka.role: work
      akka.seed: akka://ClusterSystem@light-house:12000
      akka.hostname: work-node1
      akka.hostport: 12000
      akka.cluster-config: cluster.conf
    networks:
      - mynet
  work-node2:
    image: registry.webnori.com/javalabs-api:dev        
    ports:
      - "8083:8080"
    depends_on:
      - light-house     
    environment:      
      TZ: Asia/Seoul
      akka.role: work
      akka.seed: akka://ClusterSystem@light-house:12000
      akka.hostname: work-node2
      akka.hostport: 12000
      akka.cluster-config: cluster.conf
    networks:
      - mynet
  manager-node:
    image: registry.webnori.com/javalabs-api:dev        
    ports:
      - "8084:8080"
    depends_on:
      - light-house     
    environment:      
      TZ: Asia/Seoul
      akka.role: manager
      akka.seed: akka://ClusterSystem@light-house:12000
      akka.hostname: manager-node
      akka.hostport: 12000
      akka.cluster-config: cluster.conf
    networks:
      - mynet                     
networks:
  mynet:
    driver: bridge

akka{
  actor {
    provider = cluster
  }

  remote.artery {
    canonical {
      hostname = "127.0.0.1"
      port = 12551
    }
  }

  cluster {
    seed-nodes = [
      "akka://ClusterSystem@127.0.0.1:12551"
    ]
    role{
      seed.min-nr-of-members=1
    }

    # auto downing is NOT safe for production deployments.
    # you may want to use it during development, read more about it in the docs.
    #
    auto-down-unreachable-after = 10s
    downing-provider-class = "akka.cluster.sbr.SplitBrainResolverProvider"
  }

  extensions=["akka.cluster.metrics.ClusterMetricsExtension"]
}

@Component
public class ApplicationStartup implements ApplicationListener<ApplicationReadyEvent> {

    /**
     * This event is executed as late as conceivably possible to indicate that
     * the application is ready to service requests.
     */
    @SneakyThrows
    @Override
    public void onApplicationEvent(final ApplicationReadyEvent event) {
        AkkaManager.getInstance();
    }
}

// 클래스 목적 :
// Actor시스템을 생성하고, 액터관리 Spring 디펜던시 없이 로우코드로 구현
// Spring Bean 활용시 참조 : https://www.baeldung.com/akka-with-spring
public final class AkkaManager {
    private static AkkaManager INSTANCE;
    @Getter
    private final ActorSystem actorSystem;
    private String akkaConfig;
    private String role;

    private String hostname;

    private String hostport;

    private String seed;


    @Getter
    private ActorRef greetActor;

    @Getter
    private ActorRef routerActor;

    @Getter
    private ActorRef clusterActor;

    @Getter
    private ActorRef clusterManagerActor;

    private AkkaManager() {

        akkaConfig = System.getenv("akka.cluster-config");
        role = System.getenv("akka.role");
        hostname = System.getenv("akka.hostname");
        hostport = System.getenv("akka.hostport");
        seed = System.getenv("akka.seed");

        actorSystem = serverStart("ClusterSystem", akkaConfig, role);

        InitActor();
    }

    public static AkkaManager getInstance() {
        if (INSTANCE == null) {
            INSTANCE = new AkkaManager();
        }
        return INSTANCE;
    }

    boolean isEmptyString(String string) {
        return string == null || string.isEmpty();
    }

    private ActorSystem serverStart(String sysName, String clusterConfig, String role) {

        Config regularConfig = ConfigFactory.load();

        Config combined;

        Boolean isCluster = !isEmptyString(clusterConfig) || !isEmptyString(role) || !isEmptyString(hostname)
                || !isEmptyString(hostport) || !isEmptyString(seed);

        if (isCluster) {
            Config newConfig = ConfigFactory.parseString(
                    String.format("akka.cluster.roles = [%s]", role)).withFallback(
                    ConfigFactory.load(clusterConfig));

            newConfig = ConfigFactory.parseString(
                    String.format("akka.cluster.seed-nodes  = [\"%s\"] ", seed)).withFallback(
                    ConfigFactory.load(newConfig));

            newConfig = ConfigFactory.parseString(
                    String.format("akka.remote.artery.canonical.hostname = \"%s\" ", hostname)).withFallback(
                    ConfigFactory.load(newConfig));

            newConfig = ConfigFactory.parseString(
                    String.format("akka.remote.artery.canonical.port = %s ", hostport)).withFallback(
                    ConfigFactory.load(newConfig));

            combined = newConfig
                    .withFallback(regularConfig);
        } else {
            final Config newConfig = ConfigFactory.parseString(
                    String.format("akka.cluster.roles = [%s]", "seed")).withFallback(
                    ConfigFactory.load("cluster"));
            combined = newConfig
                    .withFallback(regularConfig);
        }

        ActorSystem serverSystem = ActorSystem.create(sysName, combined);
        serverSystem.actorOf(Props.create(ClusterListener.class), "clusterListener");
        return serverSystem;
    }

    private void InitActor() {
        // Create Some Actor
        greetActor = actorSystem.actorOf(HelloWorld.Props()
                .withDispatcher("my-dispatcher"), "HelloWorld");

        // Create Router Actor
        routerActor = actorSystem.actorOf(new RoundRobinPool(5)
                .props(HelloWorld.Props()), "roundRobinPool");

        actorSystem.actorOf(TimerActor.Props()
                .withDispatcher("my-blocking-dispatcher"), "TimerActor");


        // Cluster Actor
        int totalInstances = 100;
        int maxInstancesPerNode = 3;
        boolean allowLocalRoutees = true;

        Set<String> useRoles = new HashSet<>(Arrays.asList("work"));
        clusterActor =
                actorSystem
                        .actorOf(
                                new ClusterRouterPool(
                                        new RoundRobinPool(0),
                                        new ClusterRouterPoolSettings(
                                                totalInstances, maxInstancesPerNode, allowLocalRoutees, useRoles))
                                        .props(Props.create(ClusterHelloWorld.class)),
                                "workerRouter1");

        Set<String> useManagerRoles = new HashSet<>(Arrays.asList("manager"));
        clusterManagerActor =
                actorSystem
                        .actorOf(
                                new ClusterRouterPool(
                                        new RoundRobinPool(0),
                                        new ClusterRouterPoolSettings(
                                                totalInstances, maxInstancesPerNode, allowLocalRoutees, useManagerRoles))
                                        .props(Props.create(ClusterHelloWorld.class)),
                                "workerRouter2");
    }

}

public class ClusterHelloWorld extends AbstractActor {
    private final LoggingAdapter log = Logging.getLogger(getContext().getSystem(), this);

    Cluster cluster = Cluster.get(getContext().system());

    public static Props Props() {
        return Props.create(ClusterHelloWorld.class);
    }

    //subscribe to cluster changes
    @Override
    public void preStart() {
        cluster.subscribe(self(), (ClusterEvent.SubscriptionInitialStateMode) ClusterEvent.initialStateAsEvents(),
                ClusterEvent.MemberEvent.class, ClusterEvent.UnreachableMember.class);
    }

    @Override
    public void postStop() {
        cluster.unsubscribe(self());
    }

    @Override
    public Receive createReceive() {
        return receiveBuilder().match(ClusterEvent.MemberUp.class, mUp -> {
            log.info("Member is Up: {}", mUp.member());
        }).match(ClusterEvent.UnreachableMember.class, mUnreachable -> {
            log.info("Member detected as unreachable: {}", mUnreachable.member());
        }).match(ClusterEvent.MemberRemoved.class, mRemoved -> {
            log.info("Member is Removed: {}", mRemoved.member());
        }).match(ClusterEvent.MemberEvent.class, message -> {
        // 구현가능 분산처리 메시지 사용자 정의부분
        }).match(String.class, s -> {
            log.info("Received String message: {} {}", s, context().self().path());
        })
        .match(TestClusterMessages.Ping.class, s -> {
            log.info("Received Ping message: {}",  context().self().path());
        })
        .build();
    }
}

@SpringBootTest
public class FactorialTest {

    private static Logger logger = LoggerFactory.getLogger(FactorialTest.class);
    private static ActorSystem clusterSystem1;
    private static ActorSystem clusterSystem2;
    private static ActorSystem clusterSystem3;

    private int maxServerUptime = 20;

    private static ActorSystem serverStart(String sysName, String config, String role) {
        final Config newConfig = ConfigFactory.parseString(
                String.format("akka.cluster.roles = [%s]", role)).withFallback(
                ConfigFactory.load(config));

        ActorSystem serverSystem = ActorSystem.create(sysName, newConfig);
        serverSystem.actorOf(Props.create(ClusterListener.class), "clusterListener");
        return serverSystem;
    }

    @BeforeClass
    public static void setup() {
        // Seed
        clusterSystem1 = serverStart("ClusterSystem", "server", "seed");

        // Works Nodes
        clusterSystem2 = serverStart("ClusterSystem", "factorial", "backend");
        clusterSystem2.actorOf(Props.create(FactorialBackend.class), "factorialBackend");

        clusterSystem3 = serverStart("ClusterSystem", "factorial", "backend");
        clusterSystem3.actorOf(Props.create(FactorialBackend.class), "factorialBackend");

        logger.info("========= sever loaded =========");
    }

    @AfterClass
    public static void gracefulDown() {
        clusterSystem3.terminate();
        clusterSystem2.terminate();
        clusterSystem1.terminate();
        logger.info("========= sever down =========");
    }

    @Test
    public void clusterTest() {
        logger.info("========= client start =========");
        final int upToN = 200;

        final Config config = ConfigFactory.parseString(
                "akka.cluster.roles = [client]").withFallback(
                ConfigFactory.load("factorial"));

        final ActorSystem system = ActorSystem.create("ClusterSystem", config);
        system.log().info("Factorials will start when 2 backend members in the cluster.");

        new TestKit(system) {
            {
                ActorRef probe = getRef();
                Cluster.get(system).registerOnMemberUp(new Runnable() {
                    @Override
                    public void run() {
                        ActorRef frontActor = system.actorOf(Props.create(FactorialClient.class, upToN, false),
                                "factorialClient");
                        frontActor.tell(new FactorialRequest(upToN), probe);
                    }
                });
                expectMsgClass(Duration.ofSeconds(maxServerUptime), FactorialResult.class);
            }
        };
    }
}