[아이템 81] wait와 notify보다는 동시성 유틸리티를 애용하라

wait(), notify(), notifyAll()

• 자바 5+ 버전에서 도입된 고수준의 동시성 유틸리티가 wait(), notify()로 하드 코딩해야 했던 일들을 대신 처리해 주기 때문에 현재는 wait()와 notify(), notifyAll() 메서드를 직접 호출해야 할 일이 많이 줄어듦
• 또한, wait()와 notify()는 올바르게 사용하기 까다롭기 때문에 java.util.concurrent의 고수준 동시성 유틸리티를 사용하는 것을 권장
• 고수준 동시성 유틸리티의 예는 다음과 같음 
  • ExecutorService
  • ConcurrentHashMap과 같은 Concurrent Collection
  • Synchronizer

 

1. wait()와 notify()를 사용해야하는 케이스

 

• 새로운 코드라면 언제나 wait()와 notify()가 아닌 동시성 유틸리티를 사용하는 것을 권장하지만 어쩔 수 없이 레거시 코드를 다루어야 할 때가 있음
• wait() 메서드는 쓰레드가 어떤 조건이 충족되기를 기다리게 할 때 사용하며 락을 해제, 다른 쓰레드가 종료되었다고 notify() 메서드를 호출할 때까지 Waiting 상태 유지
• 락 객체의 wait() 메서드는 반드시 해당 객체를 잠근 동기화 영역 안에서 호출해야 함
• wait()를 사용하는 표준 방식은 다음과 같으며 wait() 메서드를 사용할 때는 반드시 대기 반복문(wait loop) 관용구를 사용하고 반복문 밖에서는 절대로 호출해서는 안됨
  • 대기 전에 조건을 검사하여 조건이 이미 충족되었을 경우 wait()를 건너뛰게 한 것은 응답 불가 상태를 예방하는 조치
  • 만약 조건이 이미 충족되었는데도, 쓰레드가 notify() 또는 notifyAll()을 호출한 후에 대기 상태로 들어간다면, 다른 쓰레드가 그 쓰레드를 깨울 수 있는 보장이 없음
  • 다시 말해, 이 쓰레드가 깨어나지 않을 수 있으며 데드락(deadlock)과 같은 문제를 발생시킬 수 있으며, 프로그램의 정확성을 위협할 수 있음
  • 이를 해결하기 위해서는 wait() 메서드 호출 전에 조건을 다시 확인하고, 조건이 이미 충족되었다면 wait()을 호출하지 않아야 함

 

 

 

• 안전 실패를 막는 조치로 대기 후에 조건을 검사하여 조건이 충족되지 않았다면 다시 대기하게 코드를 작성하는 방법도 있음
  • 만약 조건이 충족되지 않았는데도 불구하고 쓰레드가 깨어나 동작을 이어나가면 락이 보호하는 불변식이 깨질 수 있음

 

1.1 조건이 만족되지 않았는데도 불구하고 쓰레드가 깨어날 수 있는 케이스

• 쓰레드가 notify()를 호출한 뒤 대기 중이던 쓰레드가 깨어나는 사이에 다른 쓰레드가 락을 얻어 그 락이 보호하는 상태를 변경
  
  • 깨어난 쓰레드는 해당 공유 자원에 대한 상태를 알지 못하고 있을 수 있으며, 변경된 상태를 고려하지 않은 상태로 작업을 진행할 수 있음

 

• 조건이 만족되지 않았음에도 다른 쓰레드가 실수로 혹은 악의적으로 notify()를 호출
  • 공개된 객체를 락으로 사용해 대기하는 클래스는 위험에 노출
  • 객체의 락을 획득한 상태에서 wait()를 호출할 수 있지만, 이러한 호출은 다른 쓰레드가 잘못된 시기에 notify()를 호출할 경우 위험에 노출
  • 깨우는 쓰레드는 지나치게 관대하기 때문에 대기 중인 쓰레드 중 일부만 조건이 충족되어도 notifyAll() 메서드를 호출해 모든 쓰레드를 깨울 수 있음
  • 대기 중인 쓰레드가 드물게 notify() 없이도 깨어나는 경우가 있으며 이를 spurious wakeup이라고 지칭
    • 안전 실패를 막는 조치로 대기 후에 조건을 검사하여 조건이 충족되지 않았다면 다시 대기하게 코드를 작성해야 하는 이유

 

2. notify() vs notifyAll()

 

• 일반적으로 언제나 notifyAll()을 호출하는 것이 합리적이고 안전한 조언
  • 꺠어나야 하는 모든 쓰레드가 깨어남을 보장하기 때문에 항상 정확한 결과를 얻을 수 있음
  • 다른 쓰레드가 깨어날 수도 있긴 하지만, 그 것이 프로그램의 정확성에는 영향을 주지 않음
  • 깨어난 쓰레드들은 기다리던 조건이 충족되었는지 확인하며, 충족되지 않을 경우 다시 대기 (안전 실패를 막는 조치를 취했다고 가정)

 

• 모든 쓰레드가 같은 조건을 기다리고, 조건이 한 번 충족될 때마다 단 하나의 쓰레드만 혜택을 받을 수 있을 경우 notifyAll() 대신 notify()를 호출해 최적화할 수 있음
• 하지만 이상의 전제 조건들이 만족될지라도 notify() 대신 notifyAll()을 사용해야 하는 이유가 있음
  • 외부로 공개된 객체에 대해 실수로 혹은 악의적으로 notify()를 호출하는 상황에 대비하기 위해 wait()를 반복문 안에서 호출했듯이 notify() 대신 notifyAll()을 사용하면 관련 없는 쓰레드가 실수로 혹은 악의적으로 wait()를 호출하는 공격으로부터 보호할 수 있음
  • 해당 공격으로부터 보호를 하지 않을 경우 깨어났어야 할 쓰레드들이 영원히 대기하는 불상사가 발생할 수 있음

 

동시성 컬렉션(Concurrent Collection)

• List, Queue, Map과 같은 표준 컬렉션 인터페이스에 동시성을 가미해 구현한 고성능 컬렉션
  • 높은 동시성에 도달하기 위해 동기화를 각자의 내부에서 수행

 

• Collections.synchronizedMap보다는 ConcurrentHashMap을 사용하는 것이 훨씬 좋음
• 동시성 컬렉션의 동시성을 무력화하는 것은 불가능하기 때문에 외부에서 락을 추가로 사용할 경우 오히려 속도만 느려짐
  • 동시성을 무력화하지 못하기 때문에 여러 메서드를 원자적으로 묶어 호출할 수 없음
  • 따라서 자바 8+ 버전부터 여러 기본 동작을 하나의 원자적 동작으로 묶는 `상태 의존적 수정` 메서드들이 일 반 컬렉션 인터페이스에 디폴트 메서드 형태로 추가됨
  • ex) Map의 putIfAbsent(key, value) 메서드는 주어진 키에 매핑된 값이 아직 없을 때만 새 값을 집어넣고 기존 값이 있었다면 그 값을 반환하고 없었다면 null을 반환
    • 해당 메서드 덕분에 thread-safe한 정규화 맵(canonicalizing map)을 쉽게 구현 가능

 

예제 #1 ConcurrentMap으로 String.intern의 동작을 흉내 내어 구현한 메서드 (최적화 전)

 

 

 

부연 설명

• `String.intern()` 메서드는 문자열을 String Pool에서 찾아서 반환하거나,  String Pool 에 문자열을 추가하여 해당 문자열의 참조를 반환하며 해당 메서드는 문자열의 메모리 사용을 최적화하고 문자열 비교 시에 참조 비교를 통해 성능 향상할 수 있음
  • 자세한 내용은 https://simple-ing.tistory.com/3 참고

 

예제 #2 ConcurrentMap으로 String.intern의 동작을 흉내 내어 구현한 메서드 (최적화 후)

 

 

 

부연 설명

• ConcurrentHashMap은 get 같은 검색 기능에 특화되어 있기 때문에 get() 메서드를 먼저 호출하여 필요할 때만 putIfAbsent를 호출하여 성능 최적화
• 해당 메서드는 String.intern() 메서드보다 훨씬 빠름 (String.intern() 메서드에는 메모리 누수 방지 기능도 있다는 점을 고려해야 하긴 함)

 

 

BlockingQueue

• 컬렉션 인터페이스 중 일부는 작업이 성공적으로 완료될 때까지 기다리도록 확장됨
  • ex) Queue를 확장한 BlockingQueue에 추가된 메서드 중 take()는 큐의 첫 원소를 꺼내고 이때 만약 큐가 비어있을 경우 새로운 원소가 추가될 때까지 대기
    
    • 이러한 특성 때문에 BlockingQueue는 작업 큐(생산자-소비자 큐)로 쓰기 적합함
    • 작업 큐는 하나 이상의 생산자(producer) 쓰레드가 작업(work)을 큐에 추가하고, 하나 이상의 소비자(consumer) 쓰레드가 큐에 있는 작업을 꺼내 처리하는 형태
    • 이를 통해 생산자와 소비자 간의 작업 조율이 용이해지며, 멀티스레드 환경에서 안전하고 효율적인 작업 처리가 가능
    • ThreadPoolExecutor를 포함한 대부분의 ExecutorService 구현체에서 BlockingQueue를 사용

 

ThreadPoolExecutor의 workQueue

 

Synchronizer

• 쓰레드가 다른 쓰레드를 기다릴 수 있게 하여 서로 작업을 조율할 수 있도록 지원
• 가장 자주 쓰이는 동기화 장치는 CountDownLatch와 Semaphore
  • CyclicBarrier와 Exchanger는 그보다 덜 쓰이며, 가장 강력한 동기화 장치는 Phaser

 

1. CountDownLatch

 

• 일회성 장벽으로 하나 이상의 쓰레드가 또 다른 하나 이상의 쓰레드 작업이 끝날 때까지 대기해야 함
• CountDownLatch의 유일한 생성자는 int 값을 받으며, 해당 값이 래치의 countDown 메서드를 몇 번 호출해야 대기 중인 쓰레드들을 깨우는지를 결정
• 이 간단한 장치를 활용할 경우 유용한 기능들을 쉽게 구현 가능
  • ex) 어떤 동작들을 동시에 시작해 모두 완료하기까지의 시간을 재는 간단한 프레임워크를 구축한다고 가정
  • 이 프레임워크는 메서드 하나로 구성되며 해당 메서드는 동작들을 실행할 실행자와 동작을 몇 개나 동시에 수행할 수 있는지를 뜻하는 동시성 수준(concurrency)를 매개변수로 받음
  • 타이머 쓰레드가 count down을 시작하기 전에 모든 작업자 쓰레드는 동작을 수행할 준비를 마침
  • 마지막 작업자 쓰레드가 준비를 마치면 타이머 쓰레드가 `시작 방아쇠`를 당겨 작업자 쓰레드들이 일을 시작하게 함
  • 마지막 작업자 쓰레드가 동작을 마치자마자 쓰레드는 시계를 멈춤
  • 위 내용을 wait()와 notify()만으로 구현하려면 아주 난해하고 지저분한 코드가 탄생하겠지만 CountDownLatch를 사용하면 다음과 같이 직관적으로 구현 가능

 

 

 

코드 부연 설명

• 위 코드는 CountDownLatch를 3개 사용
• ready Latch는 작업자 쓰레드들이 준비가 완료되었음을 타이머 쓰레드에 통지할 때 사용
• 통지를 끝낸 작업자 쓰레드들은 두 번째 Latch인 start가 열릴 때까지 대기
• 마지막 작업자 쓰레드가 ready.countDown()을 호출할 경우 타이머 쓰레드가 시작 시간을 기록하고 start.countDown()을 호출하여 기다리던 작업자 쓰레드들을 깨움
• 그 직후 타이머 쓰레드는 세 번째 Latch인 done이 열릴 때까지 대기
  • done Latch는 마지막 남은 작업자 쓰레드가 동작을 마치고 done.countDown()을 호출하면 열림
  • 타이머 쓰레드는 done Latch가 열리자마자 깨어나 종료 시각을 기록

 

추가 설명

• time 메서드에 넘겨진 실행자(executor)는 concurrency 매개변수로 지정한 동시성 수준만큼의 쓰레드를 생성할 수 있어야 함
  • 그렇지 못할 경우 해당 메서드는 끝나지 않을 것
  • 이런 상태를 쓰레드 기아 교착상태(Thread Starvation Deadlock)이라 지칭

 

• InterruptedException을 잡은 작업자 쓰레드는 Thread.currentThread().interrupt() 관용구를 사용해 interrupt를 되살리고 자신은 run 메서드에서 빠져나오는데 이렇게 해야 인터럽트를 적절히 처리 가능
  • InterruptedException 발생 시 interrupt flag가 false로 바뀌기 때문에 다시 true로 변경시키는 과정

 

• 시간 간격을 잴 때는 항상 System.currentTimeMillis가 아닌 System.nanoTime을 사용하는 것을 권장
  • System.nanoTime은 더 정확하고 정밀하며 시스템의 실시간 시계의 시간 보정에 영향받지 않음

 

2. Semaphore

 

• 공유 자원에 대한 접근을 제어하기 위해 사용되는 신호전달 메커니즘 동기화 도구
• 정수형 변수 S와 P(), V()의 두 가지 원자적 함수로 구성된 신호전달 메커니즘 동기화 도구
  • S는 공유 자원의 개수로서 이 개수만큼 쓰레드의 접근을 허용 (Mutex는 1개만 허용)
  • P()는 임계 영역을 사용하려는 쓰레드의 진입 여부를 결정하는 연산으로 Wait 연산
  • V()는 대기 중인 프로세스를 깨우는 신호로 Signal 연산

 

• 자바에서는 java.util.concurrent 패키지에 세마포어 구현체를 포함하고 있기 때문에 직접 구현할 필요 없음

 

3. CyclicBarrier

 

• 공통된 장벽 지점에 도달할 때까지 일련의 쓰레드가 서로 기다리도록 하는 동기화 보조 도구
  • 공통된 장벽 지점은 프로그램에서 지정한 지점을 의미
  • 일련의 쓰레드는 이 장벽 지점에 도달하기 위해 일을 수행하며 모든 쓰레드가 도달할 때까지 대기
  • 모든 쓰레드가 해당 지점에 도달해야지만 모든 쓰레드가 동시에 계속해서 실행될 수 있음
  • 여러 쓰레드가 병렬로 작업을 수행하다가 특정 단계에 도달하거나 모든 쓰레드가 특정 작업을 완료하고 모이는 지점에 사용
  • CountDownLatch가 다른 쓰레드가 작업을 완료하고 CountDown을 호출했을때 대기상태를 풀어준다면, CyclicBarrier는 다른 쓰레드들이 전부 대기 상태가 되었을 때 모든 쓰레드의 대기 상태를 해제
  • ex) 병렬 계산 작업 중 중간 결과를 모두 계산한 후에 다음 단계로 진행하기 위해 쓰레드들이 모이는 경우에 유용

 

• 대기 중인 쓰레드가 해제된 후 재사용할 수 있음
• CyclicBarrier는 옵션으로 Runnable 명령을 지원하는데 해당 명령은 마지막 쓰레드가 도착한 후 각 장벽 지점마다 한 번씩 실행되는 barrierAction 역할을 수행
  • CyclicBarrier가 설정된 카운트(쓰레드 수)만큼 쓰레드가 await() 메서드를 호출하여 장벽 지점에 도달하면, 그때 barrierAction이 실행된다는 것을 의미
  • 장벽 지점을 모든 쓰레드가 도달할 때마다 실행되는 것이 아니라, 모든 쓰레드가 도착한 후 한 번 실행된다는 것을 의미
  • 이 Runnable은 쓰레드가 장벽 이후  실행을 계속하기 전 공유 상태를 업데이트하는데 유용함

 

4. Exchanger

 

• 두 개의 쓰레드가 데이터를 안전하게 교환할 수 있도록 도와주는 동기화 장치
• 이는 두 쓰레드 간에 데이터를 전달할 때 하나의 스레드가 데이터를 보내고 다른 하나가 데이터를 받을 때까지 기다리도록 만듦
• Exchanger는 쓰레드 간의 작업을 조율하고 동기화하는 데 사용될 수 있으며, 주로 producer-consumer 문제를 해결하는 데 활용

 

5. Phaser

 

• Phaser는 Java의 동시성 유틸리티 클래스 중 하나로, 여러 쓰레드 간의 동기화 및 조율을 담당
• Phaser는 CyclicBarrier와 비슷한 동작을 수행하지만 동기화에 참여할 쓰레드의 수가 동적이고 재사용 가능
• Phaser에서 주의 깊게 볼 용어는 register, arrive, phase
  • register는 동기화 과정에 참여할 쓰레드의 개수를 추가하는 과정
  • 처음 Phaser를 생성할 때 동기화에 참여하는 쓰레드의 수를 parties라고 하고 int형으로 인자를 받는데 처음 생성할 때 1개로 생성하더라도 register 메서드를 통해 동적으로 참여할 쓰레드의 개수를 조정 가능
  • Phaser는 CyclicBarrier처럼 동기화 과정에 참여한 모든 쓰레드가 대기 상태에 들어갈 경우 대기 상태를 해제하는데 각 쓰레드가 대기 상태에 들어가는 지점에 도달하는 것을 arrive라고 지칭
  • 대기 상태가 풀리는 과정을 phase라고 하며 한 phase가 끝나면 Phaser 내부의 phase 값이 하나씩 증가

 

참고하면 좋은 카테고리

 

https://jaimemin.tistory.com/category/JAVA/RxJava

 

참고

이펙티브 자바

https://javabom.tistory.com/35

Java concurrent 패키지의 동기화 장치