CountDownLatch는 특정 작업을 수행하는 데 있어서 특정 쓰레드로 하여금 대기할 수 있게 해주는 역할을 한다.
Latch의 사전적 의미는 자물쇠이다. 여러 쓰레드를 자물쇠로 걸어놓고 특정 시점에 자물쇠를 한번에 푼다는 의미인 듯 하다.
CountDownLatch는 counter 필드를 가지고 있다. 이 필드를 필요할 때 감소시킬 수가 있다. 이 값이 0이 될 때까지 쓰레드의 실행을 대기할 수 있다는 것이다.
만약 특정 병렬 프로세싱 작업을 하고 있다면, CountDownLatch를 생성하고 필요한 만큼의 쓰레드 수의 값으로 초기화 할 수 있다. 그리고 나서 각 쓰레드 작업이 완료되면 countDown()을 호출할 수 있다. await는 쓰레드가 작업이 완료될 때까지 대기하게 만들 수가 있다.
쓰레드 풀이 완료될 때까지 대기
Worker 와 CountDownLatch 필드를 생성하여 완료될 때 특정 작업을 수행해보도록 하자.
우선 특정 작업을 수행하는 Worker 쓰레드를 생성하자. 내부 클래스로 생성하였다.
@AllArgsConstructor
private static class Worker implements Runnable {
CountDownLatch countDownLatch;
@Override
public void run() {
log.info("execute task"); // 여기에 작업 수행 로직이 들어간다.
countDownLatch.countDown(); // countDownLatch의 counter 필드가 -1씩 감소
}
}그리고 나서 Worker 인스턴스가 완료하기를 대기하는 CountDownLatch를 만들어보자.
@Test
public void threadPoolTest() throws InterruptedException {
int threadCount = 5;
CountDownLatch countDownLatch = new CountDownLatch(threadCount); // countdown을 5개로 설정
List<Thread> workers = Stream.generate(() -> new Thread(new Worker(countDownLatch)))
.limit(threadCount) // 쓰레드를 5개 생성
.collect(Collectors.toList());
log.info("Start multi threads");
workers.forEach(Thread::start); // 모든(5개) 쓰레드 시작
countDownLatch.await(); // countdown이 0이 될때까지 대기한다는 의미
log.info("Waiting for some work to be finished");
log.info("Finished");
}실행을 해보면 아래 로그가 찍힌다.
19:26:56.584 [main] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - Start multi threads
19:26:56.587 [Thread-0] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - execute task
19:26:56.587 [Thread-1] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - execute task
19:26:56.587 [Thread-2] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - execute task
19:26:56.587 [Thread-3] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - execute task
19:26:56.587 [Thread-4] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - execute task
19:26:56.587 [main] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - Waiting for some work to be finished
19:26:56.587 [main] INFO com.example.countdownlatch.CountDownLatchThreadPoolTest - Finished- [main] Start multi thread
- [Thread-0] ~ [Thread-4]까지 5개의 쓰레드가 실행
- [main] Waiting - 작업이 완료될 때까지 대기
- [main] Finished 모든 쓰레드 종료
countDownLatch.await()는 countDown이 0이 될 때까지 다음 로직을 수행하지 않는다. 0이 되는 순간 이후 로직이 실행이 된다.
FixedThreadPool을 사용하기
@Test
public void fixedThreadPool() throws InterruptedException {
int threadCount = 5;
CountDownLatch countDownLatch = new CountDownLatch(threadCount);
ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
log.info("Start multi threads");
for (int i = 0; i < threadCount; i++) {
executorService.execute(() -> new Worker(countDownLatch).run());
}
countDownLatch.await();
log.info("Waiting for some work to be finished");
log.info("Finished");
}모든 쓰레드 풀이 준비될 때 한번에 실행하기
이전 예제에서는 5개의 쓰레드로 실행을 했지만, 만일 수 천개의 쓰레드로 작업을 한다면 가장 마지막에 실행된 쓰레드는 첫번째 실행된 쓰레드가 완료가 된 다음 실행을 할 수도 있다. 이런 경우에는 동시성 문제를 재현하기가 어려울 수도 있다. 병렬로 모든 쓰레드를 동시에 실행하기가 어려운 상황이 생길 수도 있다는 것이다.
이런 문제를 해결하기 위해서 CountdownLatch를 좀 다른 방식으로 작동하게 해보자.
자식 쓰레드가 완료되기 전에 부모 쓰레드를 대기하지 말고 모든 쓰레드가 동시에 시작할 때까지 자식 쓰레드를 대기시킬 수 있다.
@AllArgsConstructor
private static class WaitingWorker implements Runnable {
private CountDownLatch readyLatch;
private CountDownLatch startLatch;
private CountDownLatch finishLatch;
@Override
public void run() {
readyLatch.countDown(); // readyLatch 수를 -1 한다.
try {
startLatch.await(); // startLatch가 0이 될 때까지 대기
log.info("execute task"); // 여기에 작업 수행 로직이 들어간다.
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
finishLatch.countDown(); // finishLatch 수를 -1 한다.
}
}
}여기서 readyLatch.countDown()는 readyLatch의 카운트를 -1을 하고 0이 될때까지 준비를 하자.
그리고 해당 작업이 startLatch가 0이 될때까지 다음 로직은 대기를 한다. 외부에서 startLatch가 countDown()을 하면 모든 쓰레드에서 다음 로직이 수행이 된다. 그리고 작업 수행이 완료되면 finishLatch.countDown()를 통해 finishLatch의 수를 -1을 한다.
이제 모든 Worker가 시작될 때까지 대기하고, Worker를 시작하고 완료할 때까지 대기하는 테스트를 만들어보자.
@Test
public void threadPoolTest() throws InterruptedException {
int threadCount = 5;
CountDownLatch readyLatch = new CountDownLatch(threadCount); // readyLatch을 5개로 설정
CountDownLatch startLatch = new CountDownLatch(1); // startLatch를 1개로 설정
CountDownLatch finishLatch = new CountDownLatch(threadCount); // finishLatch를 5개로 설정
List<Thread> workers = Stream.generate(() -> new Thread(new WaitingWorker(readyLatch, startLatch, finishLatch)))
.limit(threadCount) // 쓰레드를 5개 생성
.collect(Collectors.toList());
workers.forEach(Thread::start); // 모든 쓰레드 시작
readyLatch.await(); // readyLatch가 0일 될때까지 대기
log.info("start all thread");
startLatch.countDown(); // startLatch가 1이니깐 countDown()하는 순간 모든 쓰레드가 대기상태에서 동시에 실행된다.
finishLatch.await(); // finishLatch가 0이 될 때까지 대기
log.info("Waiting for all thread finished");
log.info("Finished");
}테스트에서 쓰레드 수는 5개이다. 즉 readyLatch와 finishLatch는 5개이다. 그리고 startLatch는 1개이며 모든 쓰레드가 startLatch가 1인 상태에서 대기를 한다. 그리고 main 쓰레드에서 startLatch.countDown()을 함으로써 모든 쓰레드의 작업을 실행하게 한다. 그리고 나서 finishLatch가 0이 될 때까지 대기시킨다.
쓰레드 수를 100개 이상으로 바꾸어 테스트해보면 결과를 바로 확인할 수 있다.
첫 번째 만든 방식 (쓰레드 풀이 완료될 때까지 대기)은 쓰레드 시작 시간 간격이 생각보다 길어서 특정 작업이 종료된 후에 다음 쓰레드가 실행되는 케이스가 발생할 것이고 위의 방식은 거의 대부분 비슷하게 시작을 한다.
작업의 로그를 아래와 같이 start와 end로 2개를 찍어서 실행하면 금방 알 수 있다.
@Override
public void run() {
readyLatch.countDown(); // readyLatch 수를 -1 한다.
try {
startLatch.await(); // startLatch가 0이 될 때까지 대기
// log.info("execute task"); // 여기에 작업 수행 로직이 들어간다.
log.info("start task");
// some logic
// Thread.sleep(10);
log.info("end task");
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
finishLatch.countDown(); // finishLatch 수를 -1 한다.
}
}
