ConcurrentHashMap과 동시성 컬렉션

HashMap이나 ArrayList는 여러 스레드가 동시에 고치도록 만든 클래스가 아닙니다. 16장에서 짚었듯이 여러 스레드가 같은 HashMap을 동시에 고치면 갱신이 사라지거나 맵이 망가질 수 있습니다. 그래서 Java는 여러 스레드가 함께 쓸 수 있는 컬렉션을 두 가지 방식으로 제공합니다. 하나는 Collections.synchronizedMap()처럼 기존 컬렉션의 메서드를 모두 락으로 감싸는 동기화 래퍼(synchronized wrapper)이고, 다른 하나는 java.util.concurrent 패키지에 있는 ConcurrentHashMap 같은 동시성 컬렉션(concurrent collection)입니다.

두 방식 모두 메서드 하나하나는 여러 스레드가 동시에 호출해도 안전합니다. 하지만 메서드 두 개를 이어서 호출한 것까지 하나의 동작으로 묶어 주지는 않습니다. 앞 글의 예약에서 좌석 제거와 매출 증가를 한 락 안에 넣어야 했던 것처럼, '키가 없으면 넣는다' 같은 동작도 두 호출 사이에 다른 스레드가 끼어들 수 있습니다. 이 글에서는 먼저 그 틈을 재현하고, ConcurrentHashMap이 이런 동작을 한 번에 처리하려고 제공하는 메서드를 살펴봅니다. 이어서 반복하는 도중에 다른 스레드가 컬렉션을 고치면 무엇이 보이는지가 컬렉션마다 어떻게 다른지 확인하고, 용도에 맞는 컬렉션을 고릅니다. 실행 결과는 앞 글들과 같은 코어 2개짜리 컴퓨터에서 JDK 21로 얻었습니다.

메서드마다 안전해도 두 호출 사이는 열려 있다

설정 값을 처음 쓰는 순간에 한 번만 읽어 와 맵에 저장해 두는 캐시를 생각해 봅시다. 키가 없으면 읽어 와서 넣고, 있으면 저장된 값을 씁니다. 다음 프로그램에서는 두 로더 스레드가 containsKey()로 키가 없는 것을 확인한 뒤 put()으로 값을 저장합니다. 실제로 설정을 읽는 대신 자기 스레드 이름을 저장하고, put()이 몇 번 실행되었는지를 puts로 셉니다.

두 스레드가 확인과 저장 사이에서 엇갈리는 상황을 매번 만들려고 CountDownLatch를 두 개 씁니다. CountDownLatch는 정해 둔 횟수만큼 countDown()이 호출될 때까지 await()를 호출한 스레드를 기다리게 하는 도구입니다. checked는 두 로더가 모두 키가 없는 것을 확인할 때까지 main을 기다리게 하고, proceed는 그다음에야 두 로더를 함께 put()으로 보냅니다.

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

public final class SynchronizedMapCheckThenAct {
    public static void main(String[] args) throws InterruptedException {
        Map<String, String> cache = Collections.synchronizedMap(new HashMap<>());
        AtomicInteger puts = new AtomicInteger();
        CountDownLatch checked = new CountDownLatch(2);
        CountDownLatch proceed = new CountDownLatch(1);
        Runnable task = () -> {
            if (!cache.containsKey("config")) {
                checked.countDown();
                try { proceed.await(); }
                catch (InterruptedException e) { Thread.currentThread().interrupt(); return; }
                cache.put("config", Thread.currentThread().getName());
                puts.incrementAndGet();
            }
        };
        Thread a = Thread.ofPlatform().name("loader-a").start(task);
        Thread b = Thread.ofPlatform().name("loader-b").start(task);
        checked.await();
        proceed.countDown();
        a.join(); b.join();
        System.out.println(cache + " puts=" + puts.get());
    }
}

70번 실행해 70번 모두 puts=2였고, 맵에 남은 이름은 loader-a가 33번, loader-b가 37번이었습니다. 다음은 그중 한 번의 출력입니다.

{config=loader-b} puts=2

puts=2는 두 로더가 모두 설정을 읽어 저장했다는 뜻입니다. 맵에는 값이 하나만 남았지만, 그것은 나중에 실행된 put()이 먼저 저장된 값을 덮어썼기 때문입니다. 그래서 어느 이름이 남는지는 실행마다 다릅니다. synchronizedMap()이 돌려준 맵의 메서드는 모두 같은 락 안에서 실행되므로 containsKey()와 put()은 각각 안전합니다. 하지만 락은 메서드가 끝날 때마다 풀리므로, 두 호출 사이에는 다른 스레드가 얼마든지 끼어들 수 있습니다. CAS로 스핀 락 만들기에서 본 검사 후 행동(check-then-act) 경쟁이 여기서도 일어난 것입니다.

이 예제는 래치로 엇갈림을 억지로 만들었지만, 래치가 없어도 같은 일은 우연히 일어날 수 있습니다. 설정을 읽는 데 시간이 오래 걸릴수록 두 호출 사이가 벌어지므로 더 자주 일어납니다. 그러면 비싼 읽기가 두 번 실행되고, 먼저 저장된 값을 이미 가져다 쓰기 시작한 스레드는 맵에서 사라진 객체를 계속 붙잡게 됩니다.

동기화 래퍼에서 두 호출을 묶는 방법

동기화 래퍼에서 두 호출을 하나로 묶으려면 래퍼 객체 자체를 락으로 잡습니다. synchronizedMap()이 돌려준 맵은 자기 자신을 락으로 삼아 모든 메서드를 실행하므로, 같은 객체로 synchronized 블록을 만들면 그동안 다른 스레드는 이 맵의 어떤 메서드도 실행할 수 없습니다. 아래의 loadConfig()는 설정을 읽어 오는 메서드라고 가정합니다.

synchronized (cache) {
    if (!cache.containsKey("config")) {
        cache.put("config", loadConfig());
    }
}

다만 앞의 재현 프로그램에서 래치를 기다리는 코드까지 이 블록 안으로 옮기면 안 됩니다. 한 로더가 락을 쥔 채 proceed를 기다리는 동안 다른 로더는 synchronized에서 막혀 checked를 줄이지 못하므로, 두 로더와 main이 서로를 영원히 기다리게 됩니다. 또 이 방법이 통하려면 맵에 대한 모든 접근이 래퍼를 거쳐야 합니다. 래퍼에 넘긴 원래 HashMap의 참조를 다른 곳에서 직접 쓰면 그 접근은 락을 거치지 않습니다. 예제처럼 new HashMap<>()을 곧바로 넘기면 원래 맵의 참조가 남지 않으므로 이런 실수를 막을 수 있습니다.

JDK 21의 동기화 래퍼는 putIfAbsent(), computeIfAbsent(), compute(), merge() 같은 메서드도 호출 전체를 같은 락으로 감쌉니다. 그러니 cache.computeIfAbsent("config", key -> loadConfig()) 한 줄로 바꿔도 됩니다. 하지만 래퍼에는 맵 전체에 락이 하나뿐이어서, 설정을 읽는 동안에는 다른 키를 읽고 쓰려는 스레드까지 모두 기다립니다. 여러 스레드가 함께 쓰는 맵이 필요하다면 보통 다음에 볼 ConcurrentHashMap을 씁니다.

ConcurrentHashMap은 확인과 변경을 한 번에 한다

ConcurrentHashMap은 맵 전체를 락 하나로 감싸지 않고, 서로 다른 키에 대한 읽기와 쓰기가 대부분 동시에 진행되도록 설계한 맵입니다. 그리고 '확인하고 바꾸기'를 한 번에 처리하는 메서드를 제공합니다. 공식 문서는 putIfAbsent(), computeIfAbsent(), compute(), merge()의 호출 전체가 원자적으로 실행된다고 정해 두었습니다. 다른 스레드는 이 메서드가 실행되기 전이나 끝난 뒤의 맵만 볼 수 있고, 그 중간에 끼어들지 못합니다.

메서드마다 무엇을 확인해서 저장하고 무엇을 돌려주는지는 조금씩 다릅니다. 다음 표로 정리했습니다.

표에서 눈여겨볼 점이 두 가지 있습니다. 첫째, computeIfAbsent(), compute(), merge()에 넘긴 함수가 null을 돌려주면 '저장하지 않는다' 또는 '매핑을 지운다'는 뜻이 됩니다. 이 동작은 Map 인터페이스의 명세가 정한 것이라 null 값을 허용하는 HashMap에서도 같습니다. 여기에 더해 ConcurrentHashMap은 키와 값으로 null을 받지 않고, 넣으려 하면 NullPointerException을 던집니다. 이 규칙 덕분에 get()이 null을 돌려주면 '키가 없다'는 뜻으로만 읽을 수 있습니다. null 값을 허용했다면 키가 없는 것인지 null이 저장된 것인지 구분하려고 containsKey()를 한 번 더 불러야 하는데, 그 두 호출 사이에 또 다른 스레드가 끼어들 수 있습니다.

둘째, 표 아래의 설명처럼 putIfAbsent(k, expensive())는 키가 있든 없든 인자를 먼저 계산합니다. 이 차이를 확인하려고, 앞의 설정 캐시를 먼저 computeIfAbsent()로 만들어 실행한 뒤 한 줄만 putIfAbsent()로 바꿔 비교하겠습니다. 다음 프로그램은 설정을 읽는 load()가 몇 번 실행되는지 셉니다. 두 로더는 start 래치가 열리기를 기다렸다가 함께 출발합니다.

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

public final class ComputeIfAbsentCache {
    private static final AtomicInteger loads = new AtomicInteger();

    static String load(String key) {
        loads.incrementAndGet();
        return Thread.currentThread().getName();
    }

    public static void main(String[] args) throws InterruptedException {
        ConcurrentHashMap<String, String> cache = new ConcurrentHashMap<>();
        CountDownLatch start = new CountDownLatch(1);
        Runnable task = () -> {
            try { start.await(); }
            catch (InterruptedException e) { Thread.currentThread().interrupt(); return; }
            cache.computeIfAbsent("config", ComputeIfAbsentCache::load);
        };
        Thread a = Thread.ofPlatform().name("loader-a").start(task);
        Thread b = Thread.ofPlatform().name("loader-b").start(task);
        start.countDown();
        a.join(); b.join();
        System.out.println(cache + " loads=" + loads.get());
    }
}

40번 실행해 40번 모두 loads=1이었고, 맵에 남은 이름은 loader-a가 21번, loader-b가 19번이었습니다. load()는 자기를 실행한 스레드의 이름을 값으로 돌려주므로, 맵의 값을 보면 어느 로더가 계산했는지 알 수 있습니다. 다음은 그중 한 번의 출력입니다.

{config=loader-a} loads=1

computeIfAbsent()를 먼저 실행한 스레드가 load()를 실행해 값을 넣습니다. 다른 스레드가 그 계산 도중에 같은 키로 computeIfAbsent()를 호출하면 계산이 끝날 때까지 기다렸다가, 이미 들어 있는 값을 돌려받고 load()는 실행하지 않습니다. 이것은 운이 좋아서 나온 결과가 아닙니다. 공식 문서는 '키가 없으면 함수를 호출마다 정확히 한 번 실행하고, 키가 있으면 전혀 실행하지 않는다'고 정해 두었고, 호출 전체가 원자적이므로 두 호출이 함께 '키가 없다'고 판단할 수도 없습니다.

이번에는 task 안의 computeIfAbsent() 줄을 cache.putIfAbsent("config", load("config"));로 바꿔 봅니다. 40번 실행해 40번 모두 loads=2가 나왔습니다. Java는 메서드를 호출하기 전에 인자부터 계산하므로, 두 스레드는 putIfAbsent()에 들어가기도 전에 각자 load()를 실행합니다. putIfAbsent()가 원자적으로 처리하는 것은 이미 계산된 값을 저장할지 말지뿐입니다. 맵에 남는 값은 하나지만, 비싼 계산은 중복됩니다.

computeIfAbsent()를 쓸 때도 지켜야 할 점이 있습니다. 공식 문서에 따르면 계산하는 동안 이 맵에 대한 다른 스레드의 갱신 일부가 막힐 수 있으므로, 계산은 짧고 단순해야 합니다. 함수 안에서 이 맵을 고쳐서도 안 됩니다. 또 '정확히 한 번'은 키가 없을 때 호출 한 번에 대한 약속입니다. 함수가 null을 돌려주거나 예외를 던지면 아무것도 저장되지 않으므로, 다음 호출이 다시 계산합니다. 매핑이 지워진 뒤의 호출도 다시 계산합니다.

더 깊이: 서로 다른 키는 왜 서로를 기다리지 않을까

JDK 8 이후의 ConcurrentHashMap은 내부 배열의 칸(bin)마다 따로 동기화합니다. 비어 있는 칸에 첫 원소를 넣을 때는 락 없이 CAS 한 번으로 넣고, 원소가 이미 있는 칸을 고칠 때는 그 칸의 첫 노드만 락으로 잡습니다. 그래서 서로 다른 칸에 들어가는 키들은 동시에 갱신할 수 있고, 읽기는 대부분 락 없이 진행됩니다. computeIfAbsent()의 계산이 길면 다른 갱신이 막힐 수 있다는 문서의 설명도 여기서 나옵니다. 계산하는 동안 그 키가 들어갈 칸이 잡혀 있으므로, 같은 칸에 들어가는 다른 키를 갱신하려는 스레드도 함께 기다립니다. 함수 안에서 같은 맵을 고치지 말라는 것도 같은 이유입니다. 잡힌 칸을 다시 건드리면 예외가 나거나 끝나지 않을 수 있습니다.

개수 세기는 merge 한 번으로

ConcurrentHashMap을 쓰더라도 메서드를 나눠 호출하면 같은 문제가 생깁니다. 단어가 나올 때마다 개수를 1씩 늘리는 다음 record()는 흔히 보는 모양이지만, 여러 스레드가 함께 부르면 틀립니다.

void record(String word) {
    counts.put(word, counts.getOrDefault(word, 0) + 1); // 읽기와 쓰기가 따로 일어난다
}

getOrDefault()와 put()은 각각 원자적이지만, 한 스레드가 값을 읽고 1을 더해 쓰기 전에 다른 스레드가 같은 키에 값을 쓸 수 있습니다. 그러면 나중에 쓰는 스레드가 낡은 값에 1을 더한 결과로 덮어써서 증가 하나가 사라집니다. 21장 첫 글에서 volatile 필드에 value++를 했을 때와 같은 갱신 손실입니다. 아래 프로그램의 record()를 이 코드로 바꿔 30번 실행했더니, 30번 모두 적어도 한 단어의 개수가 기대보다 적었습니다. 세 단어가 모두 적게 나온 실행이 대부분이었고, 한두 단어만 정확한 값이 나온 실행도 두 번 있었습니다. 예를 들어 한 번은 다음과 같았습니다.

{cas=71076, lock=31569, map=30955}

고친 프로그램은 다음과 같습니다. merge(word, 1, Math::addExact)는 키가 없으면 1을 넣고, 있으면 기존 값과 1을 Math.addExact()로 더한 결과로 바꿉니다. 확인과 변경이 호출 하나 안에서 원자적으로 일어납니다.

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

public final class ConcurrentWordCounter {
    private final ConcurrentHashMap<String, Integer> counts = new ConcurrentHashMap<>();

    void record(String word) { counts.merge(word, 1, Math::addExact); }

    public static void main(String[] args) throws InterruptedException {
        ConcurrentWordCounter counter = new ConcurrentWordCounter();
        List<String> words = List.of("cas", "lock", "cas", "map");
        Thread[] workers = new Thread[4];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = Thread.ofPlatform().start(() -> {
                for (int round = 0; round < 10_000; round++) {
                    for (String word : words) {
                        counter.record(word);
                    }
                }
            });
        }
        for (Thread worker : workers) {
            worker.join();
        }
        System.out.println(counter.counts);
    }
}

18번 실행해 18번 모두 같은 결과가 나왔습니다. 출력은 맵 전체를 한 줄로 찍은 것이며, 단어마다 기록된 횟수가 들어 있습니다.

{cas=80000, lock=40000, map=40000}

단어 목록에 cas가 두 번 들어 있으므로 cas는 작업자 4개 × 1만 라운드 × 2번 = 8만 번 기록되고, lock과 map은 4만 번씩 기록됩니다. 모든 작업자를 join()한 뒤에 출력하므로 이 값은 언제나 같습니다. Integer::sum 대신 Math::addExact를 쓴 것은 개수가 int 범위를 넘을 때 조용히 음수가 되는 대신 예외가 나게 하려는 것입니다. 출력에서 키의 순서는 이 컴퓨터에서 매번 같았지만, ConcurrentHashMap은 반복 순서를 약속하지 않으므로 순서에 기대는 코드를 쓰면 안 됩니다.

더 깊이: 아주 자주 세는 키에는 LongAdder 값

merge()는 같은 키에 대한 갱신을 한 번에 하나씩 처리합니다. 그래서 여러 스레드가 몇 개 안 되는 키를 아주 자주 늘리면 그 키에서 기다림이 생깁니다. ConcurrentHashMap의 공식 문서는 이럴 때 값으로 LongAdder를 두는 방법을 소개합니다. 맵의 타입을 ConcurrentHashMap<String, LongAdder>로 바꾸고 record()를 다음처럼 씁니다.

void record(String word) {
    counts.computeIfAbsent(word, key -> new LongAdder()).increment();
}

computeIfAbsent()는 단어마다 LongAdder를 한 번만 만들어 넣고, 그 뒤로는 이미 들어 있는 것을 돌려줍니다. increment()는 맵의 원자적 처리 밖에서 일어나지만, 21장 첫 글에서 본 것처럼 LongAdder는 여러 스레드가 함께 더해도 안전하고 경합이 생기면 내부의 칸을 나눠 씁니다. 위 프로그램을 이렇게 바꿔 10번 실행했을 때도 출력은 모두 {cas=80000, lock=40000, map=40000}였습니다. LongAdder의 toString()이 합계를 문자열로 돌려주기 때문입니다. 다만 다른 스레드가 더하는 도중에 읽은 합계는 원자 스냅샷이 아니므로, 정확한 값은 모든 작업이 끝난 뒤에 읽어야 합니다.

반복하는 도중에 컬렉션이 바뀌면

여러 스레드가 함께 쓰는 컬렉션에서는 한 스레드가 반복하는 동안 다른 스레드가 원소를 넣거나 뺄 수 있습니다. 이때 반복하는 쪽에 무엇이 보이는지는 컬렉션마다 다릅니다.

동기화 래퍼는 메서드 호출 하나하나만 락으로 보호하고, 반복자로 원소를 차례로 꺼내는 과정 전체는 보호하지 않습니다. 그래서 Collections.synchronizedMap()의 공식 문서는 반복하는 동안 사용자가 직접 래퍼를 락으로 잡아야 한다고 정해 두었습니다. 그렇게 하지 않으면 Iterable과 Iterator에서 본 ConcurrentModificationException이 날 수도 있고, 예외 없이 잘못된 결과가 나올 수도 있습니다.

ConcurrentHashMap의 반복자는 반복하는 도중에 맵이 바뀌어도 ConcurrentModificationException을 던지지 않습니다. 공식 문서의 표현으로는 '반복자를 만든 시점이나 그 이후의 어느 시점의 상태'를 반영합니다. 반복하는 도중에 다른 스레드가 넣거나 뺀 원소가 보일 수도 있고 보이지 않을 수도 있다는 뜻이며, 이런 성질을 약한 일관성(weakly consistent)이라고 합니다. 반복으로 모은 결과가 어느 한순간의 맵 전체와 같다는 약속은 없습니다. size()도 마찬가지여서, 다른 스레드가 갱신하는 중에 부른 size()는 대략적인 값으로만 써야 하고 그 값으로 프로그램의 동작을 결정하면 안 됩니다. 어느 한순간의 정확한 전체 모습이 필요하다면, 모든 쓰기와 그 읽기를 같은 락으로 묶거나 21장 첫 글처럼 불변 객체를 새로 만들어 참조를 통째로 바꾸는 방법을 씁니다.

CopyOnWriteArrayList는 또 다른 방법을 씁니다. 이름 그대로 쓸 때마다 복사합니다(copy on write). 원소를 넣거나 빼면 내부 배열 전체를 복사해 새 배열을 만들고, 내부 참조를 새 배열로 바꿉니다. 반복자는 만들어질 때의 배열을 붙잡고 끝까지 그 배열만 봅니다. 그 배열은 다시 바뀌지 않으므로 락 없이 반복해도 안전하고 ConcurrentModificationException도 나지 않지만, 반복자를 만든 뒤에 일어난 변경은 보이지 않습니다. 이처럼 어느 한 시점의 내용을 그대로 담은 것을 스냅샷(snapshot)이라고 합니다. 반복자의 remove()처럼 반복자를 통해 원소를 바꾸는 메서드는 지원하지 않아서, 호출하면 UnsupportedOperationException이 납니다.

CopyOnWriteArrayList로 만든 리스너 목록

스냅샷 반복이 잘 맞는 곳이 이벤트 리스너 목록입니다. 리스너는 가끔 등록되지만, 이벤트가 생길 때마다 목록 전체를 반복하며 호출되므로 읽기가 쓰기보다 훨씬 많습니다. 다음 ListenerRegistry는 리스너를 CopyOnWriteArrayList에 담고, publish()에서 향상된 for문으로 모든 리스너에게 이벤트를 전달합니다. main은 첫 리스너를 등록하고 다른 스레드에서 ready를 발행하게 한 뒤, 그사이에 두 번째 리스너를 등록합니다. 발행 스레드가 끝나기를 기다린 다음에는 next를 발행합니다.

import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

public final class ListenerRegistry {
    private final CopyOnWriteArrayList<Consumer<String>> listeners = new CopyOnWriteArrayList<>();

    void add(Consumer<String> listener) { listeners.add(listener); }
    void publish(String event) {
        for (Consumer<String> listener : listeners) {
            listener.accept(event);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ListenerRegistry registry = new ListenerRegistry();
        registry.add(event -> System.out.println("first=" + event));
        Thread publisher = Thread.ofPlatform().start(() -> registry.publish("ready"));
        registry.add(event -> System.out.println("second=" + event));
        publisher.join();
        registry.publish("next");
    }
}

이 프로그램의 출력은 두 가지 중 하나입니다. 160번 실행했더니 102번은 다음처럼 두 번째 리스너도 ready를 받았습니다.

first=ready
second=ready
first=next
second=next

나머지 58번은 다음처럼 첫 번째 리스너만 ready를 받았습니다. 두 출력에서 다른 것은 second=ready 줄이 있는지뿐이고, next는 두 경우 모두 두 리스너에게 전달됐습니다.

first=ready
first=next
second=next

출력이 갈리는 이유는 publish("ready")의 향상된 for문이 반복자를 만든 시점에 있습니다. 반복자가 main의 두 번째 add()보다 먼저 만들어졌다면 첫 리스너만 든 배열을 반복하므로 second=ready는 출력되지 않습니다. add()가 먼저였다면 두 리스너가 든 새 배열을 반복합니다. 어느 쪽이든 반복하는 도중에 배열이 바뀌지는 않으므로, 리스너 목록이 반쯤 바뀐 상태를 보거나 예외가 나는 일은 없습니다. 반면 next는 언제나 두 리스너에게 전달됩니다. main은 두 번째 리스너를 등록하고 발행 스레드가 끝나기를 기다린 뒤에 next를 발행하므로, 그때 만든 반복자에는 두 리스너가 모두 들어 있습니다. 같은 join() 때문에 ready 줄은 언제나 next 줄보다 먼저 나옵니다.

그림은 ready의 반복자가 두 번째 등록보다 먼저 만들어진 실행 하나를 따라갑니다. 그림 아래의 'second=ready가 추가되는 경로는 관측되지 않았다'는 문장도 그림이 따라간 그 한 번의 실행에 대한 것입니다. 앞에서 보았듯이 160번 가운데 102번은 두 번째 등록이 먼저였고, 그때는 2단계에서 second=ready가 함께 출력됩니다.

리스너 목록에 이 컬렉션을 쓸 때 알아 둘 점이 세 가지 있습니다. 첫째, 쓸 때마다 배열 전체를 복사하므로 등록과 해제가 잦거나 원소가 많으면 비용이 커집니다. 읽기가 쓰기보다 훨씬 많을 때만 알맞습니다. 둘째, 스냅샷이 복사하는 것은 리스너 참조의 배열이지 리스너 객체가 아닙니다. 리스너 객체 안의 상태를 여러 스레드가 함께 바꾼다면 그 상태는 따로 보호해야 합니다. 셋째, 반복이 안전하다는 것과 리스너가 실패했을 때의 처리는 별개의 문제입니다. publish()는 listener.accept()를 try-catch로 감싸지 않으므로, 한 리스너가 예외를 던지면 반복이 멈추고 그 뒤의 리스너는 그 이벤트를 받지 못합니다. 한 리스너가 실패해도 나머지에게 계속 전달할지, 발행 전체를 실패로 처리할지는 이벤트의 성격에 맞게 정해야 합니다.

용도에 맞는 동시성 컬렉션 고르기

동시성 컬렉션은 같은 인터페이스를 구현하더라도 복사 비용, 정렬, 기다리는 방식이 서로 다릅니다. 주로 쓰는 것들을 하려는 일에 따라 정리하면 다음과 같습니다.

ConcurrentLinkedQueue와 ConcurrentLinkedDeque는 락 대신 CAS로 원소를 넣고 빼는 연결 구조입니다. 크기 제한이 없고, 비어 있을 때 poll()로 꺼내려 하면 기다리지 않고 바로 null을 돌려줍니다. 그래서 생산자가 소비자보다 빠르면 원소가 끝없이 쌓입니다. 소비가 따라가지 못할 때 생산을 늦추는 장치, 즉 역압(backpressure)이 필요하다면 20장에서 본 ArrayBlockingQueue처럼 용량을 정한 BlockingQueue를 씁니다. 가득 차면 넣는 쪽이 기다리고, 비면 꺼내는 쪽이 기다립니다. DelayQueue는 원소마다 정한 지연 시간이 지나야 take()로 꺼낼 수 있는 큐입니다. 크기 제한이 없으므로 예약 작업이 끝없이 쌓이지 않도록 상한을 따로 둬야 합니다.

ConcurrentSkipListSet과 ConcurrentSkipListMap은 원소를 정렬된 순서로 유지합니다. TreeSet처럼 비교 결과로 같은 원소인지 판단하므로, Comparator가 0을 돌려주면 equals()가 false인 두 원소도 같은 원소로 취급합니다. 표준 라이브러리에는 넣은 순서를 기억하는 동시성 맵이나 집합, 이를테면 ConcurrentLinkedHashMap 같은 클래스가 없습니다. CopyOnWriteArraySet은 내부의 CopyOnWriteArrayList에 원소를 덧붙이므로 실제로는 넣은 순서대로 반복되지만, 쓸 때마다 복사하므로 원소가 적고 쓰기가 드물 때만 쓸 만합니다. 넣은 순서가 필요한 맵이나 쓰기가 많은 집합이라면 LinkedHashMap을 락 하나로 감싸거나, 한 스레드만 그 컬렉션을 고치게 하는 식으로 설계합니다.

다음 프로그램은 이 성질 가운데 몇 가지를 한 번에 확인합니다.

import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArraySet;

public final class ConcurrentSetAndDequeChoices {
    public static void main(String[] args) {
        var subscribers = new CopyOnWriteArraySet<>(List.of("audit", "metrics"));
        subscribers.add("audit");

        var scores = new ConcurrentSkipListSet<>(List.of(30, 10, 20));

        var byLength = new ConcurrentSkipListSet<String>(Comparator.comparingInt(String::length));
        byLength.addAll(List.of("ab", "cd", "xyz"));

        var jobs = new ConcurrentLinkedDeque<String>();
        jobs.offerLast("normal-report");
        jobs.offerFirst("urgent-alert");

        System.out.println("subscribers=" + subscribers);
        System.out.println("scores=" + scores);
        System.out.println("byLength=" + byLength);
        System.out.println("firstJob=" + jobs.pollFirst());
    }
}

subscribers=[audit, metrics]
scores=[10, 20, 30]
byLength=[ab, xyz]
firstJob=urgent-alert

첫 줄의 subscribers에는 audit을 두 번 넣었지만 한 번만 들어 있고, 넣은 순서대로 출력됩니다. 둘째 줄의 scores는 넣은 순서와 상관없이 정렬되어 있습니다. 셋째 줄의 byLength는 문자열의 길이로 비교하는 Comparator를 썼으므로, ab와 길이가 같은 cd는 같은 원소로 취급되어 들어가지 않았습니다. "ab".equals("cd")가 false인데도 그렇습니다. 그래서 정렬된 집합에 쓰는 Comparator는 equals()와 판단이 맞아야 합니다. 마지막 줄은 offerFirst()로 맨 앞에 넣은 urgent-alert가 pollFirst()로 먼저 나온 결과입니다. 이 프로그램은 스레드 하나로만 실행하므로 출력은 언제나 같습니다.

쓰던 컬렉션을 다른 동시성 컬렉션으로 바꿀 때는 성능만 보지 말고, 반복할 때 보이는 내용이 달라지는지도 확인해야 합니다. 예를 들어 Collections.synchronizedList()를 반복할 때는 래퍼를 락으로 잡으므로, 그동안 다른 스레드의 쓰기는 기다리고 반복이 끝날 때까지 목록은 바뀌지 않습니다. CopyOnWriteArrayList로 바꾸면 그 락이 필요 없어지는 대신, 반복하는 동안에도 다른 스레드가 목록을 바꿀 수 있고 반복자는 그 변경을 보지 못합니다. 반복한 결과가 지금 목록의 내용과 같다고 가정하는 코드가 있었다면, 컴파일 오류 없이 동작만 조용히 달라집니다.

정리

• 동기화 래퍼와 동시성 컬렉션은 메서드 하나하나를 안전하게 만들 뿐, 메서드 여러 개를 이어 호출한 것을 하나의 동작으로 묶어 주지 않습니다. containsKey() 뒤의 put()처럼 확인과 변경을 나누면 그사이에 다른 스레드가 끼어듭니다.
• ConcurrentHashMap의 putIfAbsent(), computeIfAbsent(), compute(), merge()는 호출 전체가 원자적입니다. 업무 동작 하나에 맞는 메서드를 먼저 찾습니다.
• putIfAbsent(k, expensive())는 인자를 먼저 계산하므로 계산이 중복될 수 있습니다. 키가 없을 때만 계산하려면 computeIfAbsent()를 쓰고, 그 함수는 짧게 유지하며 안에서 같은 맵을 고치지 않습니다.
• ConcurrentHashMap은 null 키와 값을 받지 않으며, 반복자는 약한 일관성을 따르고 size()는 갱신 중에는 대략적인 값입니다.
• CopyOnWriteArrayList의 반복자는 만들 때의 스냅샷을 봅니다. 읽기가 대부분인 리스너 목록에 알맞지만, 쓸 때마다 배열 전체를 복사합니다.
• 큐를 고를 때는 크기 제한과 기다리는 방식을 확인합니다. 역압이 필요하면 용량을 정한 BlockingQueue를 씁니다.

연습 문제

1. 여러 서버에서 들어온 로그인 이벤트를 한 프로세스에서 처리합니다. 사용자마다 가장 이른 로그인 시각만 남기려 합니다. 같은 사용자의 이벤트가 여러 스레드에서 동시에 들어와도, 그 사용자의 첫 기록이 동시에 만들어지더라도 올바르게 동작하도록 compute()로 기존 값과 새 값을 비교해 저장하세요.

2. 다음 add()는 그룹 이름별로 회원 목록을 모읍니다. groups의 타입은 ConcurrentHashMap<String, List<String>>입니다. 여러 스레드가 같은 그룹에 동시에 회원을 추가하면 어떤 문제가 생기는지 설명하고 고치세요.

void add(String group, String member) {
    if (!groups.containsKey(group)) {
        groups.put(group, new CopyOnWriteArrayList<>());
    }
    groups.get(group).add(member);
}

3. ConcurrentHashMap에 원소가 100개를 넘지 않게 하려고 if (map.size() < 100) map.put(key, value);라고 썼습니다. 이 코드의 문제를 두 가지 설명하세요.

import java.time.Instant;
import java.util.concurrent.ConcurrentHashMap;

public final class FirstLoginRegistrySolution {
    private final ConcurrentHashMap<String, Instant> first = new ConcurrentHashMap<>();

    void record(String userId, Instant at) {
        first.compute(userId, (key, old) -> old == null || at.isBefore(old) ? at : old);
    }

    public static void main(String[] args) throws InterruptedException {
        FirstLoginRegistrySolution registry = new FirstLoginRegistrySolution();
        Instant early = Instant.parse("2026-01-01T00:00:00Z");
        Instant late = Instant.parse("2026-01-02T00:00:00Z");
        Thread a = Thread.ofPlatform().start(() -> registry.record("u1", late));
        Thread b = Thread.ofPlatform().start(() -> registry.record("u1", early));
        a.join(); b.join();
        System.out.println(registry.first.get("u1"));
    }
}

2026-01-01T00:00:00Z

void add(String group, String member) {
    groups.computeIfAbsent(group, key -> new CopyOnWriteArrayList<>()).add(member);
}

다음 글

지금까지는 스레드를 Thread.ofPlatform().start()로 직접 만들었습니다. 22장 실행자와 스레드 풀에서는 작업을 스레드에 직접 맡기는 대신 ExecutorService에 제출합니다. 스레드 풀이 작업자 스레드를 재사용하고, 이 글에서 본 BlockingQueue 같은 큐에 작업을 쌓아 두며 몇 개의 스레드로 실행할지를 관리합니다.