[Effective Java] 11-1. 동기화

조슈아 블로크 님의 "Effective Java" 책을 정리한 포스팅 입니다.

 

1. 공유 중인 가변 데이터는 동기화해 사용하라

synchronized

배타적 실행

• 원자적 실행
  • 해당 메서드나 블록을 한번에 한 스레드씩 수행하도록 보장
• 해당 블록에 접근 시, 락이 걸림

 

스레드 사이의 통신

• 로컬 캐시의 값이 메인 메모리와의 동기화를 수행함
• 필드를 읽을 때, 항상 수정이 완전히 반영된 값을 얻지 않음
  • 스레드의 로컬 캐시와 메인 스레드의 동기화가 즉각적으로 이루어지지 않음
  • 한 스레드가 저장한 값이 다른 스레드에게 보이는 가를 보장하지 않음

 

Volatile

• 스레드의 메모리 가시성을 메인 메모리로 일치시킴
• 단위 연산일 경우, 스레드 통신을 보장하므로 동기화 연산을 생략할 수 있음

 

AtomicLong

private static final AtomicLong nextSerialNum = new AtomicLong();

public static long generateSerialNumber() { return nextSerialNum.getAndIncrement(); }

• 동기화 지원
• 성능 우수

 

주의사항

증가 연산자는 동기화 필요

• 필드에 두번 접근하므로 중간 상태의 값을 다른 스레드가 읽을 수 있음

 

2. 과도한 동기화는 피하라

과도한 동기화는 성능을 떨어뜨리고, 교착상태에 빠뜨리며, 예측 불가한 동작을 낳기도 함

 

동기화가 필요한 동작의 제어를 절대 클라이언트에게 양도하면 안됨

@FunctionalInterface
public interface SetObserver<E> {
    void added(ObservableSet<E> set, E element);
}

public class ObservableSet<E> extends ForwardingSet<E> {
    private final List<SetObserver<E>> observers = new ArrayList<>();

    private void addObserver(SetObserver<E> observer) {
        synchronized(observers) {
           observers.add(observer);
        }
    }

    private void removeObserver(SetObserver<E> observer) {
        synchronized(observers) {
           observers.remove(observer);
        }
    }
    
    private void notifyElementAdded(E element) {
        synchronized(observers) {
           for (SetObserver<E> observer : observers) 
               observers.added(this, element);
        }
    }    
}

 

ConcurrentModificationException

ObservableSet<Integer> set = new ObservableSet<>(new HashSet<>());

set.addObserver((s, e) -> {
    if (e == 32) s.removeObserver(this);
});

for (int i = 0; i < 100; i++) set.add(i);

• added 메서드 호출이 일어난 시점이 notifyElementAdded가 관찰자들의 리스트를 순회하는 도중이기 때문
  • 리스트 순회 도중, 리스트로부터 자신을 제거할 수 없음
  • 자신이 콜백을 거쳐 되돌아와 수정하는 것까지 막을 수 없음

 

교착상태

ObservableSet<Integer> set = new ObservableSet<>(new HashSet<>());

set.addObserver((s, e) -> {
    if (e == 32) {
        ExecutorService exec = Executors.newSingleThreadExecutor();
        
        try {
            exec.submit(() -> s.removeObserver(this)).get();            
        } catch (ExecutorException | InterruptedException ex) {
            throw new AssertionError(ex);
        } finally {
            exec.shutdown();
        }
    }
});

• 백그라운드 스레드가 s.removeObserver를 호출하면 관찰자를 잠그려 시도하지만 락을 얻을 수 없음
  • 메인 스레드가 이미 락을 쥐고 있음 (원소를 순회중)
• 메인 스레드는 백그라운드 스레드가 관찰자를 제거하기만을 기다림
  • ExecutorService.get(): 작업이 완료될 떄까지 호출한 스레드는 블록됨

 

해결방법

외계인 메서드 호출을 동기화 블록 바깥으로 옮기기

public class ObservableSet<E> extends ForwardingSet<E> {
    private final List<SetObserver<E>> observers = new ArrayList<>();
    
    private void notifyElementAdded(E element) {
    	List<SetObserver<E>> snapshot = null;
        
        synchronized(observers) {
            snapshot = new ArrayList<>(observers);
        }
    
        for (SetObserver<E> observer : observers) 
           observers.added(this, element);
    }    
}

• 락 없이도 안전하게 순회할 수 있음
• CopyOnWriteArrayList를 사용하여 코드를 줄일 수 있음

 

3. 스레드보다는 실행자, 태스크, 스트림을 애용하라

작업 큐

• 클라이언트가 요청한 작업을 백그라운드 스레드에 위임해 비동기적으로 처리
• 안전 실패나 응답 불가가 될 여지를 없애는 코드를 추가해야 함

 

ExecutorService

• 유연한 태스크 실행 기능 제공
• Thread를 다루기 위한 작업 단위와 실행 매커니즘이 분리됨

 

get()

• 해당 작업의 결과 반환 (Future 객체)
• 작업이 완료되지 않았으면, 완료될 떄까지 호출한 스레드는 블록됨

 

invokeAny()

• 여러 작업을 제출했을 때, 가장 빨리 완료된 작업의 결과를 반환

 

invokeAll()

• 여러 작업을 제출했을 때, 모든 작업이 완료된 경우 결과를 반환

 

awaitTermination()

• 모든 작업을 완료할 때까지 대기

 

ExecutorCompletionService

• 완료된 순서대로 작업 결과를 받음
• 내부적으로 BlockingQueue를 사용

 

ThreadPool

ThreadPoolExecutor

• 스레드 개수를 고정
• 스레드 관리 최적화, 성능 향상
• 대규모 서버에 적절함

 

ScheduledThreadPoolExecutor

• 지정된 간격으로 작업 실행
• 요청받은 태스크들이 큐에 쌓이지 않고 즉시 스레드에 위임되 실행됨

 

ForkJoinTask

• ForkJoinPool을 구성하는 스레드들이 이 태스크들을 처리
• 일을 먼저 끝낸 스레드는 다른 스레드의 남은 태스크를 가져와 대신 일을 처리