들어가며
우테코, 레벨2 방탈출 예약 미션을 진행하던 중, 코드 리뷰에서 다음과 같은 질문을 받았다.
여기서 동시 요청이 발생하면 한명의 클라이언트는 500 에러를 보게 되지 않을까요?
이는 중복 예약 방지 로직과 관련된 리뷰였다.
당시 같은 시간대에 하나의 예약만 가능해야 한다는 비즈니스 규칙이 있었고,
이를 막기 위해 두 가지 방법을 사용하고 있었다.
- 애플리케이션 코드에서의 중복 예약 여부 확인
- DB 유니크 제약 조건
그런데 동시 요청이 발생하면 아래와 같은 문제가 생긴다.
Thread A와 Thread B의 요청이 동시에 도착해서 수행되지만 Thread A가 먼저 DB에 저장하게 되면 Thread B 에서는 DB의 유니크 제약 조건에 걸려서 DataIntegrityViolationException이 발생하며 사용자에게 500 에러와 함께 해당 에러가 노출된다는 내용이었다.
문제의 해결 자체는 어렵지 않았다.
내가 해당 예외를 잡아서 내부 비즈니스 예외로 변환 후 Http 상태코드로 409 Conflict를 반환하게 했다.
이렇게 코드를 개선한 뒤엔 자연스럽게 이 고민이 생겼다.
동시 요청이 실제로 제대로 처리되는지 테스트로 검증할 수 있을까?
그러다 알게 된 것이 ExecutorService와 CountDownLatch다.
여러 스레드를 준비시켰다가 동시에 출발시키는 방식으로, 실제 동시 요청 상황을 테스트 코드로 재현할 수 있었다.
비관적 락이나 낙관적 락으로 해결할 수도 있으나 이는 여기서 다루지 않고, 동시성 테스트하는 방법에 집중했다.
동시성 테스트 코드
int numberOfThreads = 5;
ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
CountDownLatch readyThreadCounter = new CountDownLatch(numberOfThreads);
CountDownLatch callingThreadBlocker = new CountDownLatch(1);
CountDownLatch completedThreadCounter = new CountDownLatch(numberOfThreads);
for (int i = 0; i < numberOfThreads; i++) {
executor.execute(() -> {
readyThreadCounter.countDown();
callingThreadBlocker.await();
// 테스트할 작업 코드 추가
});
}
readyThreadCounter.await();
callingThreadBlocker.countDown();
completedThreadCounter.await();
executor.shutdown();테스트 코드 전체가 아닌 동시성 테스트를 위한 뼈대만 가져왔다.
코드의 흐름을 전반적으로 살펴보면 다음과 같다.
ExecutorService로 스레드풀을 만들고 작업을 제출한다.- 각 스레드는 준비가 되면 신호를 보내고, 출발 신호가 올 때까지 대기한다.
- 모든 스레드가 준비되면 동시에 출발 신호를 보낸다.
- 모든 스레드가 완료될 때까지 기다린 뒤 결과를 검증한다.
여기서 2~4번 과정을 담당하는 것이 CountDownLatch다.
그럼 이제 ExecutorService와 CountDownLatch가 각각 무엇인지 알아보자.
ExecutorService
ExecutorService는 Java에서 스레드풀을 관리하는 인터페이스로 Executor를 상속받고 있다.
public interface ExecutorService extends Executor, AutoCloseable {}스레드를 직접 생성하고 관리하는 것은 번거롭고 비용이 크다.
ExecutorService는 미리 스레드를 만들어두고 재사용하는 스레드풀 방식으로 이를 해결한다.
public class Executors {
public static ExecutorService newFixedThreadPool(int nThreads) {
return new ThreadPoolExecutor(nThreads, nThreads,
0L, TimeUnit.MILLISECONDS,
new LinkedBlockingQueue<Runnable>());
}
}ExecutorService executor = Executors.newFixedThreadPool(5);newFixedThreadPool(5)를 통해 5개의 고정 스레드를 가진 스레드풀을 생성한다.
execute()
executor.execute(() -> {
// 작업 내용
});execute()는 ExecutorService의 부모 클래스인 Executor에 정의되어있다.
작업을 스레드풀에 던지고 즉시 반환하는 메서드이다.
작업을 실행하는 건 스레드풀의 스레드가 담당하고, execute()를 호출한 메인 스레드는 바로 다음 줄로 넘어간다.
스레드 풀에 순서대로 작업을 요청하는 형식이다.
shutdown()
executor.shutdown();모든 작업이 완료된 후 스레드풀을 종료한다.
호출하지 않으면 스레드풀이 계속 살아있어 프로그램이 정상 종료되지 않기 때문이다.
CountDownLatch
CountDownLatch는 하나 이상의 스레드가 다른 스레드들의 작업이 끝날 때까지
기다릴 수 있게 해주는 동기화 도구다.
CountDownLatch latch = new CountDownLatch(5);생성 시 지정한 숫자가 카운트할 개수가 된다.
핵심 메서드는 다음과 같다.
countDown()
latch.countDown();카운트를 1 감소시킨다. 카운트를 감소시킨 스레드는 그대로 계속 실행된다.
await()
latch.await();카운트가 0이 될 때까지 호출한 스레드를 블로킹한다.
카운트가 0이 되는 순간 대기 중이던 스레드가 모두 풀린다.
사실 이렇게 봤을 때는 잘 이해가 되지 않는다. 코드 상에서의 사용 방식을 보자.
테스트 코드에서의 CountDownLatch의 역할
앞서 본 테스트 코드에서 CountDownLatch는 3개가 사용됐다.
각각의 역할은 다음과 같다.
readyThreadCounter — 모든 스레드가 준비됐는지 확인
CountDownLatch readyThreadCounter = new CountDownLatch(numberOfThreads);
// 각 스레드: 준비 완료 신호
readyThreadCounter.countDown();
// 메인 스레드: 모두 준비될 때까지 대기
readyThreadCounter.await();스레드에서 countdown()을 호출해서 CountDownLatch의 카운트 수를 감소 시켰다는 것을 통해 해당 스레드가 준비완료 되었다는 신호를 보낸다.
그리고 await()를 통해 모든 스레드가 준비될 때까지 대기(블로킹)한다.
await()는 카운트가 0이 되면 블로킹을 해제하는데, 카운트가 0이 되었다는 것은 모든 스레드들이 전부 countdown()을 호출하고 이는 곧 준비가 완료되었음을 의미하는 것이기 때문이다.
- 각 스레드가
countDown()으로 신호 전송 - 메인 스레드가
await()으로 대기
이게 없으면 일부 스레드가 준비되기도 전에 출발 신호가 발사되어
진짜 동시 실행이 보장되지 않는다.
callingThreadBlocker — 동시 출발 신호총
CountDownLatch callingThreadBlocker = new CountDownLatch(1);
// 각 스레드: 출발 신호 올 때까지 대기
callingThreadBlocker.await();
// 메인 스레드: 모두 준비됐으니 동시 출발
callingThreadBlocker.countDown();다른 CountDownLatch들과 다르게 카운트를 1로 설정하고, await()와 countDown()의 호출 주체가 반대다. (메인 스레드와 각 스레드의 역할이 다르다)
- 각 스레드가
await()으로 대기 - 메인 스레드가
countDown()으로 신호 전송
자바 공식 문서에서는 카운트가 1인 CountDownLatch를 다음과 같이 설명한다.
A CountDownLatch initialized with a count of one serves as a simple on/off latch, or gate: all threads invoking await wait at the gate until it is opened by a thread invoking countDown().
카운트를 1로 설정하면 countDown() 한 번으로 대기 중인 모든 스레드를 동시에 출발시킬 수 있다는 내용이다.
일종의 실행 스위치처럼 동작하는 셈이다.
아래의 다이어그램으로 동작을 살펴보자.
메인 스레드가 스레드 풀에 작업들을 요청하면 각 스레드 풀들이 작업을 실행한다.
하지만 이때는 아직 CountDownLatch의 카운트가 1이기 때문에 전부 블로킹된다.
메인 스레드에서 스레드 풀의 각 스레드들이 준비가 완료됐다는 신호를 받으면(readyThreadCounter.await() 통과 후) callingThreadBlocker.countDown()을 호출해서 카운트 값을 1 → 0 으로 바꾼다.
그러면 await()의 블로킹이 해제되면서 각 스레드들이 전부 동시에 실행될 수 있다.
completedThreadCounter — 모든 스레드가 완료됐는지 확인
CountDownLatch completedThreadCounter = new CountDownLatch(numberOfThreads);
// 각 스레드: 작업 완료 신호 (예외가 발생해도 반드시 실행)
finally {
completedThreadCounter.countDown();
}
// 메인 스레드: 모두 완료될 때까지 대기 후 검증
completedThreadCounter.await();- 각 스레드가
countDown()으로 완료 신호 전송 - 메인 스레드가
await()으로 대기
finally 블록에서 호출하기 때문에 예외가 발생하더라도 반드시 카운트가 감소한다.
이게 없으면 아직 작업 중인 스레드가 있는데 메인 스레드가 먼저 결과를 검증해버리는 상황이 생긴다. (테스트 코드에서의 검증 의미)
completedThreadCounter.await()가 모든 스레드의 완료를 보장한 뒤에 블로킹을 해제해주기 때문에 각 스레드들이 만들어낸 결과값을 신뢰할 수 있다.
전체 실행 흐름 정리
지금까지 설명한 내용을 전체 타임라인으로 정리하면 다음과 같다.
1단계 - 작업 제출
메인 스레드가 executor.execute()로 작업을 스레드풀에 제출한다.
스레드풀의 각 스레드는 작업을 받아 즉시 실행을 시작한다.
2단계 - 준비 완료 신호
각 스레드가 readyThreadCounter.countDown()을 호출해 준비 완료 신호를 보낸다.
메인 스레드는 readyThreadCounter.await()을 통해 모든 스레드가 준비될 때까지 대기한다.
3단계 - 출발 대기
각 스레드는 callingThreadBlocker.await()를 통해 출발 신호가 올 때까지 전부 블로킹된다.
4단계 - 동시 출발
2단계의 readyThreadCounter.await()가 실행되면 모든 스레드가 준비가 완료되었다는 의미이다.
그러면 메인 스레드가 callingThreadBlocker.countDown()으로 스위치를 켠다.
블로킹되어 있던 모든 스레드가 동시에 출발한다.
5단계 - 작업 수행
각 스레드가 테스트할 작업을 동시에 실행한다.
6단계 - 완료 신호
각 스레드는 자신의 작업이 끝나면 completedThreadCounter.countDown()으로 완료 신호를 보낸다.
메인 스레드는 completedThreadCounter.await()으로 모든 스레드가 완료될 때까지 대기한다.
7단계 - 검증
모든 스레드의 완료가 보장된 뒤 결과를 검증한다.
중요 단계
핵심은 2단계와 3단계의 순서다.
모든 스레드를 callingThreadBlocker.await()에서 블로킹시키며 대기하다가 동시에 출발 신호를 보내기 때문에, 모든 스레드들의 진짜 동시 실행이 보장된다.
만약 이 순서가 보장되지 않으면 일부 스레드는 이미 출발하고, 일부 스레드는 아직 준비 중인 상황이 생겨 반쪽짜리 동시성 테스트가 되어버린다.
동시성 테스트를 적용한 실제 테스트 코드
위에서 살펴본 동시성 테스트 방식을 사용해서 실제로 동시성 테스트를 수행한 코드는 다음과 같다.
@DisplayName("동시에 예약 생성 시 하나는 성공하고 나머지는 예외 발생을 테스트합니다.")
@Test
void save_concurrent_duplicate_exception() throws InterruptedException {
Long themeId = testHelper.insertTheme(ThemeFixture.horrorThemeCreateCommand());
Long timeId = testHelper.insertReservationTime(LocalTime.of(10, 0));
ReservationCreateCommand command = ReservationFixture.futureStarkCreateCommand(themeId, timeId, NOW);
int numberOfThreads = 5;
ExecutorService executor = Executors.newFixedThreadPool(numberOfThreads);
CountDownLatch readyThreadCounter = new CountDownLatch(numberOfThreads);
CountDownLatch callingThreadBlocker = new CountDownLatch(1);
CountDownLatch completedThreadCounter = new CountDownLatch(numberOfThreads);
List<Exception> exceptions = new CopyOnWriteArrayList<>();
AtomicInteger successCount = new AtomicInteger();
AtomicInteger exceptionCount = new AtomicInteger();
for (int i = 0; i < numberOfThreads; i++) {
executor.execute(() -> {
try {
readyThreadCounter.countDown();
callingThreadBlocker.await();
reservationCommandService.save(command);
successCount.incrementAndGet();
} catch (Exception e) {
exceptions.add(e);
exceptionCount.incrementAndGet();
} finally {
completedThreadCounter.countDown();
}
});
}
readyThreadCounter.await();
callingThreadBlocker.countDown();
completedThreadCounter.await();
executor.shutdown();
SoftAssertions.assertSoftly(softly -> {
softly.assertThat(successCount.get()).isEqualTo(1);
softly.assertThat(exceptionCount.get()).isEqualTo(numberOfThreads - 1);
softly.assertThat(exceptions).hasOnlyElementsOfType(ConflictException.class);
});
}로직은 자세히 살펴보지 않아도 된다.
마지막 테스트의 검증 과정에서 검증하고자 하는 것은 다음과 같다.
- 정상적으로 성공한 스레드의 개수가 1개인지
- 나머지 스레드는 전부 실패했는지.
- 실패한 스레드에서 발생한 예외는
DataIntegrityViolationException이 아닌, 내가 변환한 비즈니스 예외가 맞는지
실제로 테스트를 돌려보면 성공하는 모습을 볼 수 있었다.
마무리
동시성 테스트는 스레드를 다뤄야 한다는 막연함 때문에 쉽게 시도해보지 못했던 영역이었다.
하지만 미션에서의 코드 리뷰 한 줄이 스스로 도전해보는 계기가 되었고,
내부 동작 방식을 깊이 파고든 것은 아니지만 동시성 테스트의 작동 원리는 이해할 수 있었다.
ExecutorService로 스레드풀을 만들고, CountDownLatch 3개를 조합해서
진정한 동시 요청 상황을 테스트 코드로 재현할 수 있다는 걸 알게 되었다.
참고
