동시성 이슈 처리 및 테스트 방안 - 2(CountDownLatch 활용)

안녕하십니까!
이번엔 저번 포스팅에 이어서 동시성 요청을 테스트하고 처리하는 방법에 대해 알아보겠습니다.

 

 

일단 원하는 테스트 방식에 대해 정리를 먼저 해볼까요?

저는 많은 수의 요청을 동시에 수용할 수 있는 프로그램을 만들고 싶습니다.

그러기 위해서 많은 수의 요청을 동시에 보내는 테스트 코드를 작성해야합니다.

 

 

위의 그림처럼 테스트 코드 구조를 정리해볼 수 있습니다.

 

 

많은 수의 N번 요청을 보낼 수 있어야 하고, 해당 요청들을 동시에 받은 후 처리해야합니다.

저번시간에 알아봤던 스레드풀 프레임워크를 활용해서 빠르게 테스트를 작성해볼까요?

 

@Test
void 스레드풀_사용한_동시요청() throws InterruptedException {
    int n = 2000;
    ExecutorService executorService = Executors.newFixedThreadPool(n);
    Point point = pointRepository.save(new Point(10000L));

    for (int i = 1; i <= n; i++) {
        executorService.execute(() -> {
            try {
                mockMvc.perform(MockMvcRequestBuilders.post("/points/{id}/{point}", point.id, 1000L)
                                .contentType(MediaType.APPLICATION_JSON))
                        .andExpect(MockMvcResultMatchers.status().isOk())
                        .andDo(MockMvcResultHandlers.print());
            } catch (Exception e) {
                System.out.println(e.getMessage());
                throw new RuntimeException("예외 발생");
            }
        });
    }
    // 작업이 모두 완료될 때까지 대기
    executorService.shutdown();
    while (!executorService.isTerminated()) {
        // 기다림
    }

    System.out.println("모든 작업이 완료되었습니다.");
    System.out.println("====================");
    System.out.println(pointRepository.findById(point.getId()).orElseThrow());
}

 

Executors.newFixedThreadPool() 를 통해 고정크기 쓰레드풀을 생성하고 요청을 1000번 멀티쓰레드로 연속적으로 보내보는 테스트입니다.

 

executorsSerivce.shutdown() 으로 스레드풀의 작업을 종료요청 후 작업이 완료되면 잔여 포인트를 출력하여 결과를 출력합니다.

 

그럼 테스트 코드를 돌려보겠습니다!

요청을 받는 컨트롤러에 로그를 찍어 2000번의 병렬 요청을 모두 받아내었나 확인해보겠습니다.

 

 

이런!

위의 병렬 요청 테스트 코드는 2000번의 요청을 보냈는데도 불구하고 컨트롤러에서는 1078개의 요청 밖에 받지 못하였습니다.

어떻게 된 일 일까요?

 

 

위의 테스트 코드의 문제점을 알아보기 위해 테스트 코드의 구조를 그림으로 그려보겠습니다!

 

 

 

 

스레드풀만을 사용한 위의 테스트 코드는 2가지 문제점을 포함하고 있습니다.

• 첫 번째 문제점 : 2000번 요청을 기대했지만 요청이 도중에 멈출 수 있음
• 두 번째 문제점 : 2000번의 요청을 동시에 보내는 것을 기대했지만 정확한 동시 요청은 아님.

 

위의 스레드풀을 이용한 테스트 코드는 테스트의 결과를 반환하기 위해 스레드풀을 종료시켜야합니다.

 

하지만 스레드풀의 요청 작업이 모두 언제 종료되는지 정확히 판단할 수 없기 때문에, 위의 예시 코드처럼 요청 도중 스레드풀이 종료되버릴 수가 있습니다.

 

또한, 병렬로 요청을 보내긴 하였지만 for문 안에서 병렬 요청을 한 것이기 때문에 완벽한 동시 요청은 아니게 되었습니다.

위 문제를 해결하기 위해서는 스레드풀을 제어할 수 있는 무언가 필요해보입니다.

 

CountDownLatch

 

스레드풀을 제어하기 위해선 CountDownLatch라는 객체의 도움을 받아야합니다!

 

CountDownLatch 클래스는 작명이 참 재밌습니다.

Latch는 걸쇠라는 뜻인데, 걸쇠를 잠궈놓고 여러 처리를 해놓은 다음, 걸쇠를 풀어서 모든 요청을 풀어버리는 그러한 의도로 지어진 클래스명 같습니다.

 

CountDownLatch에 좀 더 자세히 알아보기 위해서 공식 문서를 한 번 참고해 보겠습니다!

 

 

 

도큐멘트의 설명을 보시면, 스레드가 다른 스레드의 작업이 완료될 때까지 대기할 수 있도록 지원한다고 나와있습니다.

• CountDownLatch.await()를 호출한 쓰레드는 선언된 countDown이 0가 될 때까지 대기합니다.
• 다른 쓰레드에서는 CountDownLatch.countDown()을 호출하며 countDown을 차감하고 마지막 작업까지 처리한 후 countDown()을 호출하여 countDown을 0으로 만들면 처음 await()를 호출했던 쓰레드의 대기상태를 풀고 로직을 진행시킵니다.

그럼 CountDownLatch를 사용한 테스트 코드를 한 번 보시겠습니다!

 

@Test
public void 포인트_차감_동시_요청()_1 throws Exception {
    Point point = pointRepository.save(new Point(10000L));

    int n = 2000;
    ExecutorService executorService = Executors.newFixedThreadPool(n); // 고정된 크기의 스레드풀 생성
    CountDownLatch countDownLatch = new CountDownLatch(n); //동시에 여러 스레드가 작업을 완료할 때까지 기다릴 수 있는 수
    for (int i = 1; i <= n; i++) {
        executorService.execute(() -> { // 스레드를 하나씩 할당해서 비동기적으로 실행하도록 한다
            try {
                mockMvc.perform(MockMvcRequestBuilders.post("/points/{id}/{point}", point.id, 1000L)
                                .contentType(MediaType.APPLICATION_JSON))
                        .andExpect(MockMvcResultMatchers.status().isOk());
            } catch (Exception e) {
                System.out.println(e.getMessage());
                throw new RuntimeException("예외 발생");
            } finally {
                countDownLatch.countDown();
            }
        });
    }
//      스레드가 중단되지 않는 한 래치가 0으로 카운트다운될 때까지 현재 스레드가 대기
//      현재 래치의 개수가 0이면 메서드 즉시 종료
    countDownLatch.await(); // countDownLatch가 0이 될 때까지 기다림

    System.out.println("====================");
    System.out.println(pointRepository.findById(point.getId()).orElseThrow());
}

 

 

위 테스트 코드는 병렬 스레드로 요청을 보낼 뿐만 아니라, CountDownLatch를 사용하여 요청을 보내고 있습니다.

 

ExecutorService와 CountDownLatch를 선언한 후, 작업을 execute()하면서 countDown()를 호출하는데요.

 

countDown이 0이 될때까지 결과를 반환하지 않고 대기하고 있다가 countDown이 0이 되는 시점에서 결과를 호출하고 있습니다.

 

 

한 번 테스트 코드의 실행 결과를 보실까요?

 

 

컨트롤러가 2000번의 요청을 모두 받아낸 것을 볼 수 있습니다!

 

 

흠.. 하지만 위 코드에서도 아쉬운 점이 몇가지 남아있습니다..

 

일단, 저번 게시물에서 기존 Thread 객체에선 작업(task)과 실행 (execution)이 분리되어 있지 않았지만

Executor를 사용하여 작업과 실행을 분리시킬 수 있게 되었다고 알아보았었는데요.

지금의 코드는 그 장점을 살리진 못한 것 같습니다.

 

한 번 작업과 실행을 분리한다는 것이 어떤 것인지 코드에 적용해보실까요?

 

@Test
public void 포인트_차감_동시_요청_2() throws Exception { // Executor를 통한 작업과 실행 분리 적용
    Point point = pointRepository.save(new Point(10000L));

    int n = 2000; // 스레드 풀의 크기
    ExecutorService executorService = Executors.newFixedThreadPool(n); // 고정된 크기의 스레드 풀 생성
    CountDownLatch countDownLatch = new CountDownLatch(n); // CountDownLatch 생성

    // 동시에 실행될 작업 정의
    Runnable task = () -> {
        try {
            mockMvc.perform(MockMvcRequestBuilders.post("/points/{id}/{point}", point.id, 1000L)
                            .contentType(MediaType.APPLICATION_JSON))
                    .andExpect(MockMvcResultMatchers.status().isOk());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            countDownLatch.countDown(); // 작업이 완료되면 countDownLatch의 카운트 감소
        }
    };

    // n개의 작업을 스레드 풀에 제출  --> 작업과 실행의 분리!
    System.out.println("n개의 작업을 스레드 풀에 제출");
    for (int i = 0; i < n; i++) {
        executorService.submit(task); // 실행
    }

    // 모든 작업이 완료될 때까지 기다림
    System.out.println("현재 스레드 대기...");
    countDownLatch.await(); // countDownLatch가 0이 될 때까지 기다림
    System.out.println("모든 작업 완료");

    System.out.println("====================");
    System.out.println(pointRepository.findById(point.getId()).orElseThrow());
}

 

위의 코드는 작업과 실행을 분리시킨 코드입니다..!

 

Runnable task 변수에 작업의 내용을 선언해두고 executorService.submit(task); 를 통해서 별개의 시점에 실행시키고 있습니다.

 

 

하지만 아직도 아쉬운 점은 남아있습니다…

위의 코드는 웬만한 상황에서의 동시성 요청에 대한 테스트를 받아낼 수는 있긴 하지만, 사실 100% 완벽한 동시 요청은 아닙니다..

 

 

한 번 위의 테스트 코드의 구조를 그림으로 그려보겠습니다.

 

 

위의 코드는 모든 요청을 병렬로 받아내어 처리하고 있습니다.

 

await()를 통해 모든 요청이 받아질 때까지 대기하기 때문에 많은 수의 요청도 놓치지 않고 받아내어 안전하게 결과를 받아낼 수도 있습니다.

또한 for문 마다 실행되는 submit()의 딜레이가 매우 미세한 정도기에, 이 정도의 테스트 코드로도 동시 요청에 대한 테스트가 충분히 가능합니다만..

 

 

그래도 100% 동시 요청이 아니라는 것이 영~ 찜찜한 구석이 있긴 한 듯합니다.

맨 처음 가정한 완벽한 동시 요청에 대한 방법이 있긴 한 것일까요?

• https://www.baeldung.com/java-countdown-latch
• 위의 페이지에 이에 대한 적절한 예시 코드가 소개되어 있었습니다! 한 번 현재 코드에 적용시켜보겠습니다.

@Test
public void 포인트_차감_동시_요청_3() throws Exception { // 준비, 요청, 완료 카운트다운래치로 동시 요청
    class WaitingWorker implements Runnable {

        private Point point;
        private CountDownLatch readyThreadCounter;
        private CountDownLatch callingThreadBlocker;
        private CountDownLatch completedThreadCounter;

        public WaitingWorker(
                Point point,
                CountDownLatch readyThreadCounter,
                CountDownLatch callingThreadBlocker,
                CountDownLatch completedThreadCounter) {
            this.point = point;
            this.readyThreadCounter = readyThreadCounter;
            this.callingThreadBlocker = callingThreadBlocker;
            this.completedThreadCounter = completedThreadCounter;
        }

        @Override
        public void run() {
            readyThreadCounter.countDown();
            try {
                callingThreadBlocker.await(); //요청 대기
                mockMvc.perform(MockMvcRequestBuilders.post("/points/{id}/{point}", point.id, 1000L)
                                .contentType(MediaType.APPLICATION_JSON))
                        .andExpect(MockMvcResultMatchers.status().isOk());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                throw new RuntimeException(e);
            } finally {
                completedThreadCounter.countDown();
            }
        }
    }

    Point point = pointRepository.save(new Point(10000L));

    int n = 2000; // 스레드 풀의 크기
    ExecutorService executorService = Executors.newFixedThreadPool(n); // 고정된 크기의 스레드 풀 생성
    CountDownLatch readyThreadCounter = new CountDownLatch(n); //준비 카운트
    CountDownLatch callingThreadBlocker = new CountDownLatch(1); //요청 카운트
    CountDownLatch completedThreadCounter = new CountDownLatch(n); //완료 카운트
    // 동시에 실행될 작업 정의
    Runnable task = new WaitingWorker(
            point, readyThreadCounter, callingThreadBlocker, completedThreadCounter);

    // n개의 작업을 스레드 풀에 제출  --> 작업과 실행의 분리!
    System.out.println("n개의 작업을 스레드 풀에 제출");
    for (int i = 0; i < n; i++) {
        executorService.submit(task); // 실행
    }

    readyThreadCounter.await(); // 모든 작업 준비될 때까지 대기
    callingThreadBlocker.countDown(); // 준비 완료 시 요청 count 0으로 호출 -> 요청 동시에 n개 호출
    completedThreadCounter.await(); // 모든 작업 완료될 때까지 대기
    System.out.println("모든 작업 완료");
    System.out.println("====================");
    System.out.println(pointRepository.findById(point.getId()).orElseThrow());
}

 

좀 복잡한 코드가 되었는데요, 개념은 간단합니다.

위에서 알아보았던 CountDownLatch를 준비, 요청, 완료로 나누어 각각의 단계에서 await()-countDown()을 적용했습니다.

 

요청 래치를 먼저 await() 한 후, 준비 래치를 차례대로 countDown() 시킵니다.

 

준비 래치의 count가 0이되면 요청 래치를 countDown() 하여 요청을 실행시킵니다. 그리고 바로 완료 래치를 await()시켜 대기시킵니다.

 

요청을 n번 동시 호출하며 완료 래치를 countDown()합니다.

 

모든 요청이 완료되어 완료 래치의 count가 0가 결과를 호출합니다.

 

그림으로 나타내보면 아래처럼 나타내볼 수 있습니다.

 

 

 

준비, 요청, 완료의 단계로 나누어졌기 때문이 조금은 복잡한 모습입니다.

위 같은 상황은 동시 요청에 대한 테스트가 민감한 부분까지 이루어져야하는 경우에 도움이 될 수 있을 것 같습니다.

 

 

이번 게시물에선 이렇게 동시성 요청에 대한 테스트 코드를 작성해보는 시간을 가져보았습니다.

다음 게시물부턴 그렇다면 어떻게 동시성 요청을 처리해야하는지 본격적으로 알아보겠습니다!!