item 81 junghyunlyoo - JAVA-JIKIMI/EFFECTIVE-JAVA3 GitHub Wiki

동시성 유틸리티

요즘은 wait와 notify를 사용해야 할 이유가 많이 줄었다.

왜냐하면 자바 5에서 도입된 고수준의 동시성 유틸리티가 wait와 notify로 하드코딩해야 했던 전형적인 일들을 대신 처리해주기 때문이다.

wait와 notify는 올바르게 사용하기가 아주 까다로우니 고수준 동시성 유틸리티를 사용하자.

java.util.concurrent의 고수준 유틸리티는 아래와 같이 세 범주로 나눌 수 있다.

• 실행자 프레임워크

• 동시성 컬렉션(concurrent collection)

• 동기화 장치(synchronizer)

실행자 프레임워크는 이전에 살펴보았고(item 80) 이번 장에선 동시성 컬렉션과 동기화 장치를 다룰 것이다.

동시성 컬렉션

동시성 컬렉션은 List, Queue, Map 같은 표준 컬렉션 인터페이스에 동시성을 가미해 구현한 고성능 컬렉션이다.

높은 동시성에 도달하기 위해 동기화를 각자의 내부에서 수행한다. (item 79)

따라서 동시성 컬렉션에서 동시성을 무력화하는 건 불가능하며 e외부에서 락을 추가로 사용하면 오히려 속도가 느려진다.

---

동시성 컬렉션에서 동시성을 무력화하지 못하므로 여러 메서드를 원자적으로 묶어 호출하는 일 역시 불가능하다.

그래서 여러 기본 동작을 하나의 원자적 동작으로 묶는 '상태 의존적 수정' 메서드들이 추가되었다.

이 메서드들은 아주 유용해서 자바 8에서는 일반 컬렉션 인터페이스에도 default method(item 21) 형태로 추가되었다.

예를 들어 Map의 putIfAbsent(key, value) 메서드는 주어진 키에 매핑된 값이 아직 없을 때만 새 값을 집어넣는다.

그리고 기존 값이 있었다면 그 값을 반환하고 없었다면 null을 반환한다.

이 메서드 덕에 스레드 안전한 정규화 맵(canonicalizing map)을 쉽게 구현할 수 있다.

다음은 String.intern의 동작을 흉내 내어 구현한 메서드다.

private static final ConcurrentMap<String, String> map = new ConcurrentHashMap<>();

public static String intern(String s) {
    String previousValue = map.putIfAbsent(s, s);
    return previousValue == null ? s : previousValue;
}

아직 개선할 점이 있다. ConcurrentHashMap은 get과 같은 검색 기능에 최적화 되어 있기 때문에 get을 이용하여 필요할 때만 putIfAbsent를 호출하도록 수정하면 더 효율적이다.

public static String intern(String s) {
    String result = map.get(s);
    if (result == null) {
        result = map.putIfAbsent(s, s);
        if (result == null)
            result = s;
    }
    return result;
}

ConcurrentHashMap은 동시성이 뛰어나며 속도도 무척 빠르다. 이러한 동시성 컬렉션은 동기화한 컬렉션을 낡은 유산으로 만들어 버렸다.

Collections.synchronizedMap 보다는 ConcurrentHashMap을 사용하는 게 훨씬 효율적이다.

컬렉션 인터페이스 중 일부는 작업이 성공적으로 완료될 때까지 기다리도록(즉 차단되도록) 확장되었다.

Queue를 확장한 BlockingQueue를 살펴보자. BlockingQueue에 추가된 메서드 중 take는 큐의 첫 원소를 꺼낸다.

이때 만약 큐가 비었다면 새로운 원소가 추가될 때까지 기다린다. 이런 특성 덕에 BlockingQueue는 작업 큐(생상자-소비자 큐)로 쓰기에 적합하다.

작업 큐는 하나 이상의 생상자(producer) 스레드가 작업(work) 을 큐에 추가하고 하나 이상의 소비자(consumer) 스레드가 큐에 있는 작업을 꺼내 처리하는 형태다.

ThreadPoolExecutor를 포함한 대부분의 실행자 서비스(item 80) 구현체에서 BlockingQueue를 사용한다.

동기화 장치

동기화 장치는 스레드가 다른 스레드를 기다릴 수 있게 하여 서로 작업을 조율할 수 있게 해준다.

동기화 장치의 대표적인 예로는 CountDownLatch가 있다.

CountDownLatch는 일회성 장벽으로, 하나 이상의 스레드가 또 다른 하나 이상의 스레드 작업이 끝날 때까지 기다리게 한다.

CountDownLatch는 생성자를 1개만 가지고 있다. 이 생성자의 매개변수인 int는 CountDownLatch의 countdown 메서드를 몇 번 호출해야 대기 중인 스레드들을 깨우는지 결정한다.

이 장치를 활용하면 유용한 기능들을 쉽게 구현할 수 있다.

---

예를 들어 어떤 동작들을 동시에 시작해 모두 완료하기까지의 시간을 재는 간단한 프레임워크를 구축한다고 해보자.

이 프레임워크는 메서드 하나로 구성되며, 이 메서드는 동작들을 실행할 실행자와 동작을 몇 개나 동시에 수행할 수 있는지를 뜻하는 동시성 수준을 매개변수로 받는다.

타이머 스레드가 시계를 시작하기 전에 모든 작업자 스레드는 동작을 수행할 준비를 마친다.

마지막 작업자 스레드가 준비를 마치면 타이머 스레드가 '시작 방아쇠'를 당겨 작업자 스레드들이 일을 시작하게 한다.

마지막 작업자 스레드가 동작을 마치자마자 타이머 스레드는 시계를 멈춘다.

이상의 기능을 wait와 notify만으로 구현하려면 아주 난해하고 지저분한 코드가 만들어지지만 CountDownLatch를 쓰면 직관적으로 구현할 수 있다.

public static long time(Executor executor, int concurrency,
                        Runnable action) throws InterruptedException {
    CountDownLatch ready = new CountDownLatch(concurrency);
    CountDownLatch start = new CountDownLatch(1);
    CountDownLatch done  = new CountDownLatch(concurrency);

    for (int i = 0; i < concurrency; i++) {
        executor.execute(() -> {
            ready.countDown(); // 타이머에게 준비를 마쳤음을 알린다.
            try {
                start.await(); // 모든 작업자 스레드가 준비될 때까지 기다린다.
                action.run();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            } finally {
                done.countDown();  // 타이머에게 작업을 마쳤음을 알린다.
            }
        });
    }

    ready.await();     // 모든 작업자가 준비될 때까지 기다린다.
    long startNanos = System.nanoTime();
    start.countDown(); // 작업자들을 깨운다.
    done.await();      // 모든 작업자가 일을 끝마치기를 기다린다.
    return System.nanoTime() - startNanos;
}

이 코드는 카운트 다운 래치를 3개 사용한다. ready 래치는 작업자 스레드들이 준비가 완료됐음을 타이머 스레드에 통지할 때 사용한다.

통지를 끝낸 작업자 스레드들은 두 번째 래치인 start가 열리기를 기다린다.

마지막 작업자 스레드가 ready.countDown을 호출하면 타이머 스레드가 시작 시각을 기록하고 start.countDown을 호출하여 기다리던 작업자 스레드들을 깨운다.

그 직후 타이머 스레드는 세 번째 래치인 done이 열리기를 기다린다. done 래치는 마지막 남은 작업자 스레드가 동작을 마치고 done.countDown을 호출하면 열린다.

타이머 스레드는 done 래치가 열리자마자 깨어나 종료 시각을 기록한다.

---

time 메서드에 넘겨진 실행자(executor)는 concurrency 매개변수로 지정한 동시성 수준만큼의 스레드를 생성할 수 있어야 한다.

그렇지 못하면 이 메서드는 결코 끝나지 않을 것이다. 이런 상태를 스레드 기아 교착상태라 한다.

InterruptedException을 캐치한 작업자 스레드는 Thread.currentThread().interrupt() 관용구를 사용해 인터럽트를 되살리고 자신은 run 메서드에서 빠져나온다.

이렇게 해야 실행자가 인터럽트를 적절하게 처리할 수 있다.

위 코드에서 System.nanoTime 메서드를 사용해 시간을 잰 것에 주목하자. 

시간 간격을 잴 때는 항상 System.currentTimeMillis가 아닌 System.nanoTime을 사용하자. 

System.nanoTime은 더 정확하고 정밀하며 시스템의 실시간 시계의 시간 보정에 영향받지 않는다.

한편, 이 예제의 코드는 작업에 충분한 시간(ex 1초 이상)이 걸리지 않는다면 정확한 시간을 측정할 수 없을 것이다. 

정밀한 시간 측정은 매우 어려운 작업이라, 꼭 해야 한다면 jmh같은 특수 프레임워크를 사용해야 한다.

동시성 유틸리티를 사용하지 못 하는 경우

새로운 코드를 짜는 경우라면 항상 wait와 notify가 아닌 동시성 유틸리티를 써야 한다.

하지만 wait과 notify를 사용하는 레거시 코드를 다뤄야 하는 경우도 있을 것이다.

wait 메서드는 스레드가 어떤 조건이 충족되기를 기다리게 할 때 사용한다.

락 객체의 wait 메서드는 반드시 그 객체를 잠근 동기화 영역 안에서 호출해야 한다.

wait을 사용하는 표준 방식은 아래와 같다.

synchronized (obj) {
    while (<조건이 충족되지 않음>){
        obj.wait();
    }
    // 조건이 충족됐을 때의 동작을 수행
}

wait 메서드를 사용할 때는 반드시 대기 반복문(wait loop) 관용구를 사용하자.

그리고 반복문 밖에서는 절대로 호출하면 안된다. 이 반복문은 wait 호출 전후로 조건이 만족하는지를 검사한다.