Java Concurrent 컬렉션과 병렬 스트림의 함정, 제대로 알고 사용하자

멀티스레드 환경에서 안전한 데이터 공유는 Java 개발자들이 피할 수 없는 과제다. 특히 Java 8부터 도입된 스트림과 함께 Concurrent 컬렉션을 사용할 때, 많은 개발자들이 같은 실수를 반복한다. 우리가 공부하면서 겪었던 실제 문제들과 해결 방법을 정리해보자.

Concurrent 컬렉션이 정말 안전한가?

동기화 컬렉션의 진짜 문제

처음 Java를 배울 때 Vector나 Hashtable이 스레드 안전하다고 배웠다. 맞는 말이다. 하지만 이들은 전체 컬렉션을 락으로 보호하기 때문에 성능이 끔찍하다. 모든 메서드에 synchronized가 붙어있어서 한 번에 하나의 스레드만 접근할 수 있다. 생각해보면 읽기만 하는데도 락을 기다려야 한다니, 정말 비효율적이다.

 

ArrayList나 HashMap이 나온 이유가 바로 이것이다. 동기화 없이 빠르지만 멀티스레드 환경에서 위험하다. 그래서 우리는 Collections.synchronizedList()나 Collections.synchronizedMap()을 사용해왔다. 하지만 이것도 결국 전체를 락으로 묶는 방식이다. 여전히 느리다.

 

그렇다면 진정한 해답은 무엇일까? 바로 Concurrent 컬렉션들이다.

 

ConcurrentHashMap의 마법: 세그먼트 락

ConcurrentHashMap은 혁명적인 아이디어를 가져왔다. 전체 맵을 하나의 락으로 보호하지 않고, 여러 개의 세그먼트(버킷)로 나누어 각각을 독립적으로 보호하는 것이다. 기본적으로 16개의 세그먼트가 있어서, 최대 16개의 스레드가 동시에 다른 세그먼트에 접근할 수 있다.

 

실제로 테스트해보자.

ConcurrentHashMap<String, Integer> map = new ConcurrentHashMap<>();

// 여러 스레드가 동시에 다른 키에 put할 수 있다
ExecutorService executor = Executors.newFixedThreadPool(4);

for (int i = 0; i < 1000; i++) {
    final int index = i;
    executor.submit(() -> map.put("key" + index, index));
}

executor.shutdown();
executor.awaitTermination(10, TimeUnit.SECONDS);

System.out.println("Map size: " + map.size()); // 1000이 정확하게 출력된다

하지만 여기서 주의할 점이 있다. ConcurrentHashMap은 원자적 보장이 개별 연산에만 적용된다는 것이다. 여러 연산을 함께 수행해야 할 때는 추가 동기화가 필요하다.

// 위험한 코드
if (!map.containsKey("counter")) {
    map.put("counter", 0);
}
map.put("counter", map.get("counter") + 1);

// 위 코드는 Race Condition이 발생할 수 있다
// 여러 스레드가 동시에 같은 연산을 수행할 수 있기 때문이다

// 안전한 코드
map.putIfAbsent("counter", 0);
map.computeIfPresent("counter", (k, v) -> v + 1);

CopyOnWriteArrayList: 읽기 성능을 위한 선택

CopyOnWriteArrayList는 완전히 다른 철학을 가지고 있다. 이름에서 알 수 있듯이, 쓰기 작업이 발생할 때마다 전체 배열을 복사한다. 미쳤나 싶을 수 있지만, 읽기 작업이 압도적으로 많은 경우 이것이 최선의 선택이 될 수 있다.

 

읽기는 어떤 락도 필요 없이 기존 배열에 접근하고, 쓰기할 때만 새 배열을 만들고 교체한다. 이렇게 하면 읽기 성능은 최고이고, 동시에 완벽한 스레드 안전성을 보장한다.

CopyOnWriteArrayList<String> listeners = new CopyOnWriteArrayList<>();

// 리스너 등록 (드문 연산)
listeners.add("listener1");
listeners.add("listener2");

// 리스너 호출 (자주 발생)
for (String listener : listeners) {
    notifyListener(listener);
}

// 반복 중에 다른 스레드가 추가/제거해도 ConcurrentModificationException이 발생하지 않는다

하지만 쓰기가 빈번한 환경에서는 이 방식이 매우 비효율적이다. 계속 배열을 복사해야 하기 때문이다. 따라서 읽기가 쓰기보다 훨씬 많은 시나리오 - 예를 들어 이벤트 리스너, 캐시 데이터, 설정값 읽기 같은 경우에만 사용해야 한다.

병렬 스트림의 함정

병렬 스트림이 항상 빠를까?

Java 8에서 스트림이 도입되면서 우리는 이렇게 생각했다: "병렬 스트림을 쓰면 자동으로 빨라지겠지?" 아니다. 이것이 가장 큰 착각 중 하나다.

 

병렬 스트림은 자신의 ForkJoinPool을 사용해서 작업을 여러 스레드에 분배한다. 하지만 이 과정에는 오버헤드가 있다.

• 작업을 분할하는 비용
• 각 스레드 간 컨텍스트 스위칭 비용
• 결과를 병합하는 비용

따라서 작은 데이터셋이나 간단한 연산에서는 병렬 스트림이 순차 스트림보다 훨씬 느릴 수 있다.

List<Integer> numbers = IntStream.range(0, 100).boxed().collect(Collectors.toList());

// 순차 스트림이 더 빠를 가능성이 높다
long sum1 = numbers.stream()
    .filter(n -> n > 50)
    .map(n -> n * 2)
    .reduce(0, Integer::sum);

// 병렬 스트림 - 오버헤드가 크면 더 느릴 수 있다
long sum2 = numbers.parallelStream()
    .filter(n -> n > 50)
    .map(n -> n * 2)
    .reduce(0, Integer::sum);

// 병렬이 의미있으려면 충분히 큰 데이터와 비싼 연산이 필요하다
List<Integer> millionNumbers = IntStream.range(0, 1_000_000).boxed().collect(Collectors.toList());

long expensiveSum = millionNumbers.parallelStream()
    .filter(n -> isPrime(n)) // 비싼 연산
    .map(n -> n * complexCalculation(n)) // 비싼 연산
    .reduce(0L, Long::sum);

공유 상태의 위험성

병렬 스트림에서 가장 주의해야 할 것은 공유 상태다. 람다 내에서 외부 변수를 수정하면 동기화 없이 여러 스레드가 접근하게 되고, Race Condition이 발생한다.

// 위험한 코드
List<Integer> numbers = IntStream.range(0, 1000).boxed().collect(Collectors.toList());
List<Integer> results = new ArrayList<>();

numbers.parallelStream()
    .map(n -> n * 2)
    .forEach(n -> results.add(n)); // 여러 스레드가 동시에 add 호출
    
// results의 크기가 1000이 아닐 수 있다!

// 안전한 코드 1: 순차 스트림 사용
List<Integer> results1 = numbers.stream()
    .map(n -> n * 2)
    .collect(Collectors.toList());

// 안전한 코드 2: 병렬 + collect 사용
List<Integer> results2 = numbers.parallelStream()
    .map(n -> n * 2)
    .collect(Collectors.toList());

// 안전한 코드 3: ConcurrentHashMap 활용
ConcurrentHashMap<Integer, Integer> resultMap = new ConcurrentHashMap<>();
numbers.parallelStream()
    .forEach(n -> resultMap.put(n, n * 2));

Lazy Evaluation의 함정

스트림의 중간 연산(intermediate operation)은 최종 연산(terminal operation)이 호출될 때까지 실행되지 않는다. 이것을 Lazy Evaluation이라고 한다. 매우 좋은 기능이지만, 이것도 함정이 있다.

List<String> names = Arrays.asList("Alice", "Bob", "Charlie", "David");

// 이 코드는 아무것도 출력하지 않는다!
Stream<String> stream = names.stream()
    .filter(n -> {
        System.out.println("Filtering: " + n);
        return n.length() > 3;
    })
    .map(n -> {
        System.out.println("Mapping: " + n);
        return n.toUpperCase();
    });

// 최종 연산이 없어서 위의 filter와 map은 실행되지 않는다

// 최종 연산을 추가해야 실행된다
stream.forEach(System.out::println);

// 또 다른 문제: limit()와 distinct() 같은 연산의 순서가 중요하다
names.stream()
    .distinct()
    .limit(2) // 먼저 중복을 제거한 후 2개만 가져온다

names.stream()
    .limit(2)
    .distinct(); // 먼저 2개만 가져온 후 중복을 제거한다

ForkJoinPool 고갈의 위험

병렬 스트림은 공통 ForkJoinPool을 사용한다. 이 풀은 시스템에 하나만 존재하고, 모든 병렬 스트림이 공유한다. 만약 한 병렬 스트림에서 블로킹 작업(I/O, 네트워크 등)을 수행하면 어떻게 될까?

// 위험한 코드
List<String> urls = Arrays.asList("url1", "url2", "url3", ...);

List<String> results = urls.parallelStream()
    .map(url -> {
        // 네트워크 요청 - 블로킹 작업
        return fetchFromUrl(url); // 이 작업이 오래 걸릴 수 있다
    })
    .collect(Collectors.toList());

// ForkJoinPool의 모든 워커 스레드가 네트워크 응답을 기다리고 있으면
// 다른 병렬 스트림 작업은 실행될 수 없다!

이를 해결하려면 별도의 ExecutorService를 사용하거나, 병렬 스트림 대신 다른 방식을 고려해야 한다.

// 더 나은 코드
ExecutorService executor = Executors.newFixedThreadPool(4);
List<Future<String>> futures = new ArrayList<>();

for (String url : urls) {
    futures.add(executor.submit(() -> fetchFromUrl(url)));
}

List<String> results = new ArrayList<>();
for (Future<String> future : futures) {
    results.add(future.get());
}

executor.shutdown();

실제 사용 가이드라인

어떤 컬렉션을 선택할 것인가?

상황	추천 컬렉션	이유
읽기 매우 많음, 쓰기 드뭄	CopyOnWriteArrayList	읽기 성능 최고
균형잡힌 읽고 쓰기	ConcurrentHashMap, ConcurrentLinkedQueue	세그먼트 락으로 좋은 성능
멀티 스레드 큐 필요	ConcurrentLinkedQueue, LinkedBlockingQueue	특정 용도 최적화
레거시 코드	Vector, Hashtable	피해야 할 선택

// 이벤트 리스너 시스템 - CopyOnWriteArrayList 추천
class EventSystem {
    private CopyOnWriteArrayList<EventListener> listeners = new CopyOnWriteArrayList<>();
    
    public void addEventListener(EventListener listener) {
        listeners.add(listener); // 드문 연산
    }
    
    public void fireEvent(Event event) {
        for (EventListener listener : listeners) { // 자주 발생
            listener.onEvent(event);
        }
    }
}

// 캐시 시스템 - ConcurrentHashMap 추천
class Cache {
    private ConcurrentHashMap<String, Object> map = new ConcurrentHashMap<>();
    
    public Object get(String key) {
        return map.getOrDefault(key, null); // 읽기와 쓰기 모두 자주 발생
    }
    
    public void put(String key, Object value) {
        map.put(key, value);
    }
}

병렬 스트림 체크리스트

병렬 스트림을 사용하기 전에 이 질문들에 답해보자.

1. 데이터 크기가 충분히 큰가? (최소 수천 개 이상)
   작은 데이터는 순차 스트림이 더 빠르다
2. 연산이 충분히 비싼가? (CPU를 많이 사용하는가?)
   간단한 필터링이나 맵핑이라면 병렬의 이점이 없다
3. 공유 상태가 없는가? (stateless인가?)
   외부 변수 수정이 없어야 한다
4. 블로킹 연산이 없는가? (I/O, 네트워크 등)
   블로킹 작업이 있으면 ForkJoinPool이 고갈된다
5. 데이터 분할이 효율적인가?
   ArrayList는 좋지만 LinkedList는 나쁘다
   커스텀 Spliterator 고려 필요할 수 있다

이 질문들에 모두 Yes라면 병렬 스트림을 써도 좋다.

 

Java 멀티스레드의 악몽, 데드락·라이블락·스타베이션 완벽 정리