병행성(Concurrency)을 위한 CountDownLatch

CountDownLatch

우리가 사용하는 고급 언어는 대부분 Concurrency(병행성)에 관련된 API를 제공한다. 자바도 마찬가지로 atomic, volitile, semaphore등 기본적인 병행성 관련 메커니즘에서 CyclicBarrier, CountDownLatch등의 고급기능까지 제공한다. 그 중 이 포스트에서는 CountDownLatch에 대해 알아보도록 한다.

CountDownLatch는 언제 쓸까? 

쓰레드를 N개 실행했을 때, 일정 개수의 쓰레드가 모두 끝날 때 까지 기다려야지만 다음으로 진행할 수 있거나 다른 쓰레드를 실행시킬 수 있는 경우 사용한다. 예를들어서 리스트에 어떤 자료구조가 있고, 각 자료구조를 병렬로 처리한 후 배치(batch)로 데이터베이스를 업데이트 한다거나 다른 시스템으로 push하는 경우가 있다.

CountDownLatch의 어떤점이 이를 가능하게 하는가?

CountDownLatch를 초기화 할 때 정수값 count를 넣어준다. 쓰레드는 마지막에서 countDown() 메서드를 불러준다. 그러면 초기화 때 넣어준 정수값이 하나 내려간다. 즉 각 쓰레드는 마지막에서 자신이 실행완료했음을 countDown 메서드로 알려준다. 이 쓰레드들이 끝나기를 기다리는 쪽 입장에서는 await()메서드를 불러준다. 그러면 현재 메서드가 실행중이 메인쓰레드는 더이상 진행하지않고 CountDownLatch의 count가 0이 될 때까지 기다린다. 0이라는 정수값이 게이트(Latch)의 역할을 한다. 카운트다운이 되면 게이트(latch)가 열리는 것이다.

예제를 통해 확인하도록 하자. 

예제

예제에서는 Worker라는 쓰레드 Runnable을 구현하고 CountDownLatch를 사용할 때와 사용하지 않을때 결과가 어떻게 다른지 실험해본다.

import java.util.concurrent.CountDownLatch;
        import java.util.stream.IntStream;

public class App {

    public static void main(String[] args) throws InterruptedException {
        App app = new App();
        app.runWorkers();
    }

    private void runWorkers() throws InterruptedException {
        CountDownLatch countDownLatch = new CountDownLatch(5); // 최대 5개의 쓰레드 동시 실행
        IntStream.range(0, 5)
                .mapToObj(i -> new Worker(i, countDownLatch))
                .map(Thread::new)
                .forEach(Thread::start);

        System.out.println("Done awaiting..");

    }

    public class Worker implements Runnable { //쓰레드
        private CountDownLatch countDownLatch;
        private int index;

        public Worker(final int index, final CountDownLatch countDownLatch) {
            this.index = index;
            this.countDownLatch = countDownLatch;
        }

        @Override
        public void run() {
            try {
                System.out.println("Starting thread... " + index);
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                System.out.println("Finishing Thread... " + index);
            }
        }
    }
}

실행 결과

   Starting thread... 0
    Starting thread... 1
    Starting thread... 2
    Starting thread... 3
    Done awaiting..
    Starting thread... 4
    Finishing Thread... 0
    Finishing Thread... 2
    Finishing Thread... 3
    Finishing Thread... 1
    Finishing Thread... 4

countDown이나 await메서드를 사용하지 않으면 위처럼 쓰레드를 실행하고 나서 이 메서드도 기다리지 않고 다음 줄을 실행하기 때문에 쓰레드가 종료하기 전에 "Done awaiting"이 출력된다. 이제 쓰레드 마지막에 countDown()을 넣어보자.

@Override
public void run() {
    try {
        System.out.println("Starting thread... " + index);
        Thread.sleep(3000);
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        System.out.println("Finishing Thread... " + index);
        countDownLatch.countDown(); // 이 줄 추가
    }
}

실행 결과

    Starting thread... 0
    Done awaiting..
    Starting thread... 3
    Starting thread... 4
    Starting thread... 2
    Starting thread... 1
    Finishing Thread... 0
    Finishing Thread... 3
    Finishing Thread... 2
    Finishing Thread... 4
    Finishing Thread... 1

넣어봤자 아무 소용이 없다. 왜인가? 아무도 기다리고있지 않기 때문이다. 이제 기다리는 쪽에서 await을 이용하도록 해보자.

private void runWorkers() throws InterruptedException {
    CountDownLatch countDownLatch = new CountDownLatch(5);
    IntStream.range(0, 5)
            .mapToObj(i -> new Worker(i, countDownLatch))
            .map(Thread::new)
            .forEach(Thread::start);

    countDownLatch.await(); // 이 줄 추가
    System.out.println("Done awaiting..");

}

실행 결과

    Starting thread... 0
    Starting thread... 3
    Starting thread... 2
    Starting thread... 4
    Starting thread... 1
    Finishing Thread... 0
    Finishing Thread... 1
    Finishing Thread... 4
    Finishing Thread... 2
    Finishing Thread... 3
    Done awaiting..

await을 추가하고 돌리자 Done awaiting이 가장 마지막에 실행된다. await()메서드가 CountDownLatch의 count가 0이 될 때 까지 기다리기 때문이다.