ForkJoinPool의 Thread Size 고정이 안되는 경우

회사에서 Spring Boot 2.x로 버전업 하던 중 ForkJoinPool의 Thread 수가 고정이 안되는 문제가 생겼다... 재연해보면서 내용을 좀 정리해 봐야겠다. 

 

가. ForkJoinPool

1. 무엇일까?

 -   Fork는 쪼개고 Join은 합친다로 작업을 쪼개서 나누고 합치고 일을 뺏어서 Thread 효율을 높인다.

 -   자세한 내용은 이 블로그에 정말 잘 정리 되어있다. !! hamait.tistory.com/612

쓰레드풀 과 ForkJoinPool

2. 예제 만들기

 - 숫자 더하는 로직을 멀티스레드로 처리해보자

  IntStream.range(1,10).parallel().peek(i->{
            System.out.println(Thread.currentThread().getName());
        }).sum();

- 실행을 해보면..

: Started RedistestApplicationTests in 2.01 seconds
Test worker
ForkJoinPool.commonPool-worker-8
ForkJoinPool.commonPool-worker-4
ForkJoinPool.commonPool-worker-6
ForkJoinPool.commonPool-worker-2
ForkJoinPool.commonPool-worker-15
ForkJoinPool.commonPool-worker-13
ForkJoinPool.commonPool-worker-9
ForkJoinPool.commonPool-worker-11

- 단순히 parallel() 만 사용할 경우 ForkJoinPool.commonPool에서 Thread를 가져다 쓰고 있다. (PC의 Core는 6)

    -> 디폴트 Pool Size 값을 설정하는 부분은 코드에서 못찾았다...( 디폴트는 Core *2 -1 이다.) 

- Thread가 많으면 좋은거 아닐까라? 는 생각이 들수도 있다

    1. Thread도 리소스이므로 불필요하게 Thread를 생성할경우 아무리 ForkJoinPool라도 낭비가 발생한다.

        -> Thread가 변경될때 컨텍스트 스위칭이 발생한다. (리소스 소모)

    2. 다른 API를 콜할때 TPS 이상으로 처리될 수가 있다. (다른 회사의 API 서버가 뻗으면...) 

- 이번엔 Pool Size를 고정해보자

   int parallelism = 2; // Pool의 Thread 개수 지정
   ForkJoinPool forkJoinPool = new ForkJoinPool(parallelism);
   int sum = forkJoinPool.submit(() -> {
   	return IntStream.range(1,10).parallel().peek(i->{
   			System.out.println(Thread.currentThread().getName()+
  		 				", Pool 사이즈 : "+forkJoinPool.getPoolSize());
  	 }).sum();
   }).join();

   System.out.println("## 결과 : "+sum);

- 출력 내용을 보면...

: Started RedistestApplicationTests in 2.117 seconds
ForkJoinPool-1-worker-1, Pool 사이즈 : 2
ForkJoinPool-1-worker-0, Pool 사이즈 : 2
ForkJoinPool-1-worker-0, Pool 사이즈 : 2
ForkJoinPool-1-worker-1, Pool 사이즈 : 2
ForkJoinPool-1-worker-1, Pool 사이즈 : 2
ForkJoinPool-1-worker-0, Pool 사이즈 : 2
ForkJoinPool-1-worker-1, Pool 사이즈 : 2
ForkJoinPool-1-worker-0, Pool 사이즈 : 2
ForkJoinPool-1-worker-1, Pool 사이즈 : 2
## 결과 : 45

- Thread Pool 사이즈가 원하는데로 고정되어있다.

 

3. 이슈 재연

- 이제 실제이슈를 재연해보자

      -> 여기서 의미는 없지만 Redis(lettuce)에 데이터를 불러오는 메서드를 추가해보자

   ValueOperations<String, Object> operations = redisTemplate.opsForValue();
   int parallelism = 2; // Pool의 Thread 개수 지정
   ForkJoinPool forkJoinPool = new ForkJoinPool(parallelism);
   int sum = forkJoinPool.submit(() -> {
   	return IntStream.range(1,10).parallel().peek(i->{
    		operations.get("TEST");// Redis
   			System.out.println(Thread.currentThread().getName()+
  		 				", Pool 사이즈 : "+forkJoinPool.getPoolSize());
  	 }).sum();
   }).join();

   System.out.println("## 결과 : "+sum);

- 실행을 해보면..

: Started RedistestApplicationTests in 2.117 seconds
ForkJoinPool-1-worker-1, Pool 사이즈 : 9
ForkJoinPool-1-worker-2, Pool 사이즈 : 10
ForkJoinPool-1-worker-0, Pool 사이즈 : 8
ForkJoinPool-1-worker-3, Pool 사이즈 : 8
ForkJoinPool-1-worker-4, Pool 사이즈 : 6
ForkJoinPool-1-worker-6, Pool 사이즈 : 6
ForkJoinPool-1-worker-7, Pool 사이즈 : 6
ForkJoinPool-1-worker-5, Pool 사이즈 : 4
ForkJoinPool-1-worker-11, Pool 사이즈 : 4
## 결과 : 45

- 위와는 다르게 고정이 되지않고 default 만큼 사용하는 모습이다.. 

 

나. Lettuce

1. 무엇일까?

-  Lettuce는 Redis에 접속하기 위한 Client 이다. (Non-blocking Reactive 서비스를 위한 확장가능한 Client)

-  특징

      1. Netty 기반으로 리액티브, 비동기, 동기 방식을 지원한다.

           =>Netty는 Non Blocking I/O 기반의 Network Client Framework 이다. (Low level)

      2. 기능들은 Netty EventLoop 기반으로 동작한다.

      3. 하나의 네트워크 커넥션을 공유하기 때문에 Thread-Safe 하다.

 

2. lettuce vs Jedis

- Jedis는 lettuce와 마찬가지로 Redis의 Client 양대산맥 중 하나다.  => lettuce와는 다르게 동기방식으로 통신한다.

- Jedis는 Spring boot 2부터 data-redis의 lettuce에게 자리를 내어주었다..

- 아무래도 비동기로 동작하는 lettuce에 비해 성능이 떨어질 수 밖에 없다. 자세한 성능테스트는 이 블로그에 잘나와있었다. !!  jojoldu.tistory.com/418

Jedis 보다 Lettuce 를 쓰자

 

다. 문제 해결

- 거창하게 썼지만 사실 문제 해결보다는 회피 방법을 사용할 수 밖에 없었다 .. ㅜㅜ 결국 Lettuce를 대신 Jedis로 변경하는 방법을 택했다.

 

1. Jedis로 변경방법

- 간단하게 gradle.build에서 lettuce를 빼고 Jedis를 따로 추가하면 된다.

implementation ('org.springframework.boot:spring-boot-starter-data-redis'){
	exclude group: 'io.lettuce', module: 'lettuce-core'
}

implementation group: 'redis.clients', name: 'jedis'

- 그리고 lettuceConnectionFactory를 JedisConnectionFactory로 수정한 후에 사용하면 된다.

 

2. 다른 방법은 없었을까..

- Lettuce는 그대로 쓰면서 Thread Pool을 고정해보려고 찾아봤는데 어디 나와있는데가 없었다..

- Lettuce의 API async, sync로 모두 테스트 해보았지만 결과는 똑같았다.. :(

- 만약 불가능하다면 이유라도 정확하게 알고 싶었지만 이부분도 나와 있는 곳이 없었다.. :(

- 내 예상으로는 Netty NIO 방식 때문에 Lettuce가 high level 에서는 동기식으로 동작하더라도

   Low level에서 Non-Block 으로 동작해서 Thread Pool Size 설정을 무시하는게 아닌가하는 느낌이 든다.

- 나중에 Netty에 대해서 정리하면서 이부분을 다시 테스트 해봐야겠다.

 

- 혹시 제 글을 보는 분은 없겠지만 혹시 정답을 아신다면 댓글 부탁드려요!!

---

P.S) 이것 저것 찾아보던 중 원인을 발견했다.. 원인은 잘 못 알고 있던 부분에 있었다..

 

라. 원인 파악

a. ForkJoinPool의 parallelism

  -   위의 있는 관련 코드를 다시한번 살펴보면

   int parallelism = 2; // Pool의 Thread 개수 지정
   ForkJoinPool forkJoinPool = new ForkJoinPool(parallelism);
   int sum = forkJoinPool.submit(() -> {
   	return IntStream.range(1,10).parallel().peek(i->{
   			System.out.println(Thread.currentThread().getName()+
  		 				", Pool 사이즈 : "+forkJoinPool.getPoolSize());
  	 }).sum();
   }).join();

   System.out.println("## 결과 : "+sum);

  -  ForkJoinPool 생성할때 parallelism 값을 받는다. Pool의 Thread 개수를 지정한다고 썼는데 잘 못 알고 있었다. ForkJoinPool의 toString을 찍어보자

// 레디스 X
int sum = forkJoinPool.submit(() -> {
            return IntStream.range(1,5).parallel().peek(i->{
                System.out.println(forkJoinPool.toString());
          }).sum();
 }).get();
 

// 레디스 O
ValueOperations<String, Object> operations = redisTemplate.opsForValue();
int sum = forkJoinPool.submit(() -> {
            return IntStream.range(1,5).parallel().peek(i->{
               operations.get("Test");
               System.out.println(forkJoinPool.toString());
        }).sum();
 }).get();

  -  결과 로그

ForkJoinPool@[Running, parallelism = 2, size = 2, active = 2, running = 2, 
ForkJoinPool@[Running, parallelism = 2, size = 2, active = 2, running = 2, 
ForkJoinPool@[Running, parallelism = 2, size = 2, active = 2, running = 2,


// 레디스 있는 로그
ForkJoinPool@[Running, parallelism = 2, size = 9, active = 7, running = 6, 
ForkJoinPool@[Running, parallelism = 2, size = 9, active = 7, running = 8, 
ForkJoinPool@[Running, parallelism = 2, size = 9, active = 7, running = 8, 

  -  parallelism은 설정대로 둘다 2로 고정되어 있는걸 볼 수 있다. 둘의 차이가 무엇일까?

  -  답은 추측한대로 lettuce의 비동기 방식이었다. 오해했던 부분은 병렬(parallelism)과 동시성(concurrency)에 있었다. 

  -  <모던 자바 인 액션> 책에 둘의 차이를 비교한 그림이 있는데 그 그림에 맞춰 그려보았다.

  -  parallelism 을 고정한거는 저 화살표를 고정한 것과 같다. 즉 몇개의 병렬로 동시에 처리할 것 인지!!

  -  그리고 하나의 화살표에는 여러 개의 Thread가 동작할 수도 있는데 이부분은 비동기와 관련이 있다.

  -  결국 Jedis는 동기방식이어서 parallelism 고정하면 마치 Thread도 고정한것 처럼 보였던 것이었다.

 

b. 구현

  -  위대로라면 내부에 비동기로직이 있을경우 똑같은 현상이 발생할 것 이다. 테스트 해보자

public class AsyncTest {
    public Future<Object> async(){
        CompletableFuture<Object> completableFuture = new CompletableFuture<>();
        new Thread(()->{
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            completableFuture.complete(0);
        }).start();
        return completableFuture;
    }
}

   - 1초뒤에 0을 반환해주는 비동기코드를 만들었고 이거를 위의 코드에 추가해보았다.

int parallelism = 2; // 병렬을 지정
ForkJoinPool forkJoinPool = new ForkJoinPool(parallelism);
AsyncTest asyncTest = new AsyncTest();
int sum = forkJoinPool.submit(() -> {
       return IntStream.range(1,5).parallel().peek(i->{
           try {
                 Future<Object> async = asyncTest.async();
                 System.out.println(forkJoinPool.toString());
                  sync.get();
              } catch (InterruptedException e) {
                    e.printStackTrace();
              } catch (ExecutionException e) {
                    e.printStackTrace();
              }
        }).sum();
 }).get();

  -  결과 로그를 보면

ForkJoinPool@[Running, parallelism = 2, size = 2, active = 2, running = 2, 
ForkJoinPool@[Running, parallelism = 2, size = 2, active = 1, running = 1, 
ForkJoinPool@[Running, parallelism = 2, size = 4, active = 1, running = 1, 

  -  lettuce를 사용했을때와 동일한 결과가 나오는 것을 알 수 있다.

 

원인을 찾고 보니 의문이 생겼다. 하나의 parallel에 비동기 처리하는 Thread를 고정할 필요가 있을까? 

물론 동시성으로 처리하게 되면 기존보다 빠를 순 있지만 Concurent 하기 때문에 오히려 더 나은 성능을 기대할 수 있을 것 같다.

 

----- 추가

여러번 테스트 하고있는데 비동기가 원인이 아닌것 같다. 이거는 좀 더 실험을 해봐야겠다. 원인을 발견했다. 원인은 CompletableFuture 여기에 있었다.

 

마. 원인 파악(2) 

 a. 원인은 비동기가 아닌 것 같다?

  -  내부적으로 비동기만 처리 했을경우에 ForkJoinPool의 Thread 개수는 늘어나지 않았다.

 

 b. 진짜 원인

  -  여러번의 테스트와 인내 끝에 원인을 찾아냈다!!  CompletableFuture의 get 내부 코드를 보자

 -  waitingGet 으로 들어가보자

   -  맨아래에 ForkJoinPool의 managedBlock을 호출하는게 보인다. 뭔가 느낌이 좋다.!! 들어가보자

   -  p.tryCompensate 저부분을 지나치면 Size가 늘어나고 있었다. 거의 다온것 같다.

  -  뭔가 복잡해 보이는 곳에 눈에 띄는 createWorker가 보인다.. 정상이 눈앞이다..

  -  드디어 newThread를 발견했다...  

 

 c. 검증

  -  저 가설대로면 lettuce는 CompletableFuture 를 사용해야한다. 검증해보자!

int parallelism = 1;
ForkJoinPool forkJoinPool = new ForkJoinPool(parallelism);
ValueOperations<String, Object> operations = redisTemplate.opsForValue();
int sum = forkJoinPool.submit(() -> {
	return IntStream.rangeClosed(1,10).parallel().peek(i->{
		operations.get("TEST");
    }).sum();
  }).get();

  -  확인을 위해 break point를 CompletableFuture에 걸고 디버그 해보자

 -  이렇게 lettuce는 내부적으로 CompletableFuture를 사용하는 것을 확인했다.

 -  결국 CompletableFuture 사용할 경우 값을 불러올때 일정시간이 되면 ForkJoinPool에서 새로운 Thread를 생성하면서 계속해서 늘어나는 것이었다

 

위에 쓴 내용들은 반정도 맞았던것 같다. 역시 확실하게 확인하니까 마음이 놓이는것 같다 ㅎㅎ 

 

---- 추가 마음을 너무 놓았나 보다..

 

바. 재도전 Lettuce 

 a. 결과

  -  정리를 하면서 lettuce 써도 크게 문제 없을 것으로 생각되어 반영해서 테스트를 해봤는데

        CPU 사용률이 95 퍼센트를 넘어가버리는 대참사를 만들었다... 

  -  내부적인 문제 없을꺼라 생각했는데 이런 결과가 만들어질 줄 몰랐다 .. 

 

 b. 놓친 부분..

  -  최대치에 대한 생각을 하지 못했던게 문제였다. 전체적으로 최대치를 가져갈꺼라 생각했는데 아니었다.

  -  parallelism 이 1일때와 10일때를 비교해 보자

1일때

10일때

-  느낌이 온다.. parallelism 만큼 최대로 가질수 있는 Thread를 생성한다... 간단한 예제인데 이정도면 서버의

     CPU가 화난 이유를 알것같다.. 여기에 ForkJoinPool 이 여러 개면 ... 쉽지않다. 

 

사. Thread 고정하기

 a. CompletableFuture

  -  위에 쓴거처럼 parallel을 사용하면 ForkJoinPool을 사용하고 lettuce를 쓰면 내부적으로

        CompletableFuture가 동작해서 블로킹이 발생하면 새로운 Thread를 최대치까지 만들어버린다.

  -  Thread를 고정하려면 parallel 과 ForkJoinPool 대신 CompletableFuture의 메서드를 이용하자

 

 b. 수정

  -  Executors.newFixedThreadPool 를 이용하여 Thread를 고정하는 ExecutorService를 활용하자

  -  그리고 paralell 대신 CompletableFuture.runAsync 를 사용하여 처리해보자

 @Test
 void 테스트() throws ExecutionException, InterruptedException {
   int nThreads = 3;
   ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
   
     for (int i = 0; i < 10; i++) {
          final int num = i;
          CompletableFuture.runAsync(()->{
               AsyncTest asyncTest = new AsyncTest();
               System.out.println(num+"번째 "
               				+Thread.currentThread().getName());
               
               CompletableFuture<Object> async = asyncTest.async();
               try {
                      async.get();
               } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
               }
             },executorService).get();
       }
       
     executorService.shutdown();
 }

c. 테스트

  -  결과를 살펴보자

  -  원하는대로 Thread가 고정되었고 ForkJoinPool은 사용하지 않는 걸 볼 수 있다. 그렇기 때문에 성능은 기존보다 떨어질 수도 있다.