블락킹모드에서 병렬처리 TPS 보장하기

    private static final Executor executor = Executors.newFixedThreadPool(450);

    private static CompletionStage<String> callApiAsync(Integer param) {
        // CompletableFuture를 사용하여 비동기 처리 구현
        return CompletableFuture.supplyAsync(() -> {
            try {

                double dValue = Math.random();
                int iValue = (int)(dValue * 1000);
                Thread.sleep(iValue); // API 응답 시간을 시뮬레이션하기 위한 지연
                tpsActor.tell("CompletedEvent", ActorRef.noSender()); // TPS측정을 위
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
            return "Response for " + param;
        }, executor);
    }

    public void ThrottleTest(int tps, int testCount ) {
        new TestKit(actorSystem) {
            {
                final ActorMaterializerSettings settings = ActorMaterializerSettings.create(actorSystem)
                        .withDispatcher("my-dispatcher-streamtest");

                final Materializer materializer = ActorMaterializer.create(settings, actorSystem);
                final TestKit probe = new TestKit(actorSystem);

                tpsActor = actorSystem.actorOf(TpsMeasurementActor.Props(), "TpsActor");
                tpsActor.tell(probe.getRef(), getRef());
                expectMsg(Duration.ofSeconds(1), "done");

                // Source 생성
                Source<Integer, NotUsed> source = Source.range(1, testCount);

                // 병렬 처리를 위한 Flow 정의
                final int parallelism = 450;
                Flow<Integer, String, NotUsed> parallelFlow = Flow.<Integer>create()
                        .mapAsync(parallelism, BackPressureTest::callApiAsync);

                // Buffer 설정 및 OverflowStrategy.backpressure 적용
                int bufferSize = 100000;
                Flow<Integer, Integer, NotUsed> backpressureFlow = Flow.<Integer>create()
                        .buffer(bufferSize, OverflowStrategy.backpressure());

                AtomicInteger processedCount = new AtomicInteger();

                // Sink 정의
                Sink<String, CompletionStage<Done>> sink = Sink.foreach(s -> {
                    //처리완료
                    processedCount.getAndIncrement();
                    if(processedCount.getAcquire() % 10 == 0) {
                        //System.out.println("Processed 10");
                    }
                });

                System.out.println("Run backpressureFlow bufferSize:"+bufferSize);

                // RunnableGraph 생성 및 실행
                source.via(backpressureFlow)
                        .throttle(tps, FiniteDuration.create(1, TimeUnit.SECONDS), tps, (ThrottleMode) ThrottleMode.shaping())
                        .via(parallelFlow)
                        .to(sink)
                        .run(materializer);

                within(
                        Duration.ofSeconds(15),
                        () -> {
                            // Will wait for the rest of the 10 seconds
                            expectNoMessage(Duration.ofSeconds(10));
                            return null;
                        });
            }
        };
    }

my-dispatcher-streamtest {
  type = Dispatcher
  executor = "thread-pool-executor"
  thread-pool-executor {
    fixed-pool-size = 50
  }
  throughput = 1
}