반응형
 
 
 

Java의 ConcurrentHashMap을 이해해 보자

📌 서론

Java에서 일반적인 HashMap을 사용하면 동시성 문제가 발생할 수 있다.

HashMap은 멀티스레드 환경에서 안전하지 않기 때문에, 여러 스레드가 동시에 HashMap을 수정하려고 하면 데이터 불일치, 무한 루프, 또는 예기치 않은 결과가 발생할 수 있다.

 

이 문제를 해결하기 위한 방법 중 하나로 ConcurrentHashMap을 사용할 수 있다. ConcurrentHashMap은 멀티스레드 환경에서 안전하게 동작하도록 설계된 클래스다. 내부적으로 분할 잠금 메커니즘(lock stripping)과 CAS(Compare-And-Swap) 같은 비동기적인 동시성 제어 기법을 사용하여 여러 스레드가 동시에 데이터를 읽고 쓰는 상황에서도 안전하게 작동한다.

 

항상 이것을 사용하면서 어떻게 동작하길래 동시성 제어가 되는 것인지 궁금했는데 이번에 날을 잡고 코드 내부를 분석해 봤다.

 

코드를 따라가는 과정과 그 내용이 복잡하다 보니 필력이 부족한 제가 작성한 글로는 ConcurrentHashMap에 대해 이해하기 힘들 수도 있겠지만 천천히 읽어보시고 직접 디버깅도 해보면서 분석을 해보신다면 분명 이해하는데 도움이 될 것이라고 생각합니다.

 

그럼 지금부터 ConcurrentHashMap을 이해하러 가봅시다!

 

내용이 복잡하고 길기 때문에 단순히 우리가 자바에서 ConcurrentHashMap을 사용하면 어떻게 동작하는지 그 흐름만을 간단하게 이해하고 싶다면 7,8번 목차를 바로 확인하시는 것을 추천합니다!

 

1. ConcurrentHashMap의 소개 및 탄생 배경

ConcurrentHashMap이 뭔가요?

  • ConcurrentHashMap은 자바의 동시성 컬렉션 클래스 중 하나로, 멀티스레드 환경에서 안전하게 사용할 수 있는 해시맵이다. HashMap과 달리, ConcurrentHashMap은 여러 스레드가 동시에 데이터를 읽고 쓸 수 있도록 설계되었다.

탄생 배경

  • ConcurrentHashMap은 자바 1.5에서 도입되었으며, 주로 멀티스레드 환경에서의 성능 문제를 해결하기 위해 설계되었다. 기존의 HashMap이나 Hashtable은 멀티스레드 환경에서 데이터 무결성을 보장하지 못하거나 성능이 크게 저하되는 문제가 있었다. Hashtable은 모든 메서드에 대해 동기화를 사용하여 안전성을 확보했지만, 그로 인해 성능이 크게 떨어졌다.

  • 이를 해결하기 위해 ConcurrentHashMap이 도입되었으며, 부분적인 락(lock)을 사용하여 성능을 향상시키면서도 Thread Safety를 보장할 수 있었다.

사용 방법 (실시간 데이터 처리 시스템)

  • 실시간 데이터 처리 시스템에서는 여러 스레드가 데이터 스트림을 소비하고, 각 데이터의 처리 상태를 추적해야 할 수 있다. ConcurrentHashMap을 사용하여 각 데이터 ID의 처리 상태를 안전하게 관리할 수 있다.

  • 상태 업데이트: 데이터가 처리될 때마다 updateStatus 메서드를 사용하여 현재 상태를 업데이트한다. put 메서드는 Thread Safety 하게 상태를 변경할 수 있도록 보장한다.

  • 상태 조회: getStatus 메서드를 사용하여 데이터의 처리 상태를 실시간으로 조회한다. 만약 해당 데이터가 존재하지 않는 경우, 기본 상태 "UNKNOWN"을 반환한다.
import java.util.concurrent.ConcurrentHashMap;

public class RealTimeDataProcessor {

    private final ConcurrentHashMap<String, String> dataStatus = new ConcurrentHashMap<>();

    // 데이터 처리 상태 업데이트
    public void updateStatus(String dataId, String status) {
        dataStatus.put(dataId, status);
    }

    // 데이터 처리 상태 조회
    public String getStatus(String dataId) {
        return dataStatus.getOrDefault(dataId, "UNKNOWN");
    }

    public static void main(String[] args) {
    
        RealTimeDataProcessor processor = new RealTimeDataProcessor();

        // 스레드 A가 데이터 처리 시작
        new Thread(() -> {
            processor.updateStatus("data1", "PROCESSING");
            // 데이터 처리 로직...
            processor.updateStatus("data1", "COMPLETED");
        }).start();

        // 스레드 B가 데이터 상태를 실시간으로 조회
        new Thread(() -> {
            System.out.println("Data1 처리 상태: " + processor.getStatus("data1"));
        }).start();
    }
    
}

 

2. ConcurrentHashMap 클래스와 putVal 메서드 살펴보기

ConcurrentHashMap 클래스 살펴보기

  • ConcurrentHashMap 클래스는 다음과 같이 선언되어 있다. (자세한 코드는 IDE에서 검색해 보자!)
public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable {
 
    // ... code
 
}

 

핵심 코드인 putVal 메서드를 살펴보자

  • ConcurrentHashMap의 핵심 기능은 동시성을 안전하게 관리하는 것이다. ConcurrentHashMap에서는 읽기 작업은 별다른 동기화 없이 빠르게 수행되도록 설계되어 있어 문제가 없다. 하지만 쓰기 작업에서는 여러 스레드가 동시에 같은 키에 접근하거나 데이터를 수정하려고 할 때 동시성 문제가 발생할 수 있다.

  • 이 때문에 쓰기 작업을 처리하는 putVal 메서드가 굉장히 중요한 역할을 한다. 이 메서드는 여러 스레드가 같은 키에 쓰기를 시도할 때 동시성 제어를 통해 안전하게 데이터를 삽입하거나 업데이트할 수 있도록 설계되어 있다. 따라서 putVal 메서드를 자세히 살펴보는 것이 중요하다. 이 메서드는 락(Lock), CAS(Compare-And-Swap) 연산, 노드 동기화 등의 기법을 사용하여 멀티스레드 환경에서도 데이터의 일관성과 안전성을 보장한다.
final V putVal(K key, V value, boolean onlyIfAbsent) {
    // 키와 값이 null일 수 없으므로 NullPointerException을 던집니다.
    if (key == null || value == null) throw new NullPointerException();
    
    // 키의 해시코드를 계산하고 이를 스프레드(spread)하여 해시 테이블에서의 위치를 분산시킵니다.
    int hash = spread(key.hashCode());
    int binCount = 0;

    // 해시 테이블이 초기화되었는지, 그리고 각 버킷을 처리할 수 있는지 확인합니다.
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh; K fk; V fv;
        // 해시 테이블이 아직 초기화되지 않았거나 크기가 0이면 초기화합니다.
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // 계산된 인덱스의 버킷이 비어 있으면 CAS(Compare-And-Swap)로 새로운 노드를 추가합니다.
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
                break;  // 락 없이 빈 버킷에 노드를 추가한 경우 반복을 종료합니다.
        }
        // 버킷이 리사이징 중인 경우 해당 버킷을 리사이징 작업에 참여시킵니다.
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f);
        // 락 없이 첫 번째 노드를 확인하여 조건부로 노드를 삽입할 수 있는지 확인합니다.
        else if (onlyIfAbsent 
                 && fh == hash
                 && ((fk = f.key) == key || (fk != null && key.equals(fk)))
                 && (fv = f.val) != null)
            return fv; // 조건에 맞는 기존 값을 반환합니다.
        else {
            V oldVal = null;
            // 첫 번째 노드가 비어있지 않다면, 노드를 수정하기 위해 해당 버킷을 동기화합니다.
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) {
                        binCount = 1;
                        // 연결 리스트를 순회하여 키가 존재하는지 확인하고, 값을 갱신하거나 새 노드를 추가합니다.
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value; // 기존 값이 존재하면 덮어씁니다.
                                break;
                            }
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {
                                pred.next = new Node<K,V>(hash, key, value);
                                break; // 새 노드를 연결 리스트에 추가합니다.
                            }
                        }
                    }
                    // 노드가 트리 구조인 경우 트리 노드에 대해 처리합니다.
                    else if (f instanceof TreeBin) {
                        Node<K,V> p;
                        binCount = 2;
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value; // 트리 노드의 값을 갱신합니다.
                        }
                    }
                    // 예약된 노드에 대한 재귀적 업데이트 오류를 방지합니다.
                    else if (f instanceof ReservationNode)
                        throw new IllegalStateException("Recursive update");
                }
            }
            if (binCount != 0) {
                // 노드 수가 특정 임계치(TREEIFY_THRESHOLD)를 초과하면 연결 리스트를 트리 구조로 변환합니다.
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal; // 기존 값을 반환합니다.
                break;
            }
        }
    }
    // 전체 노드 수를 증가시키고, 필요 시 테이블을 리사이징합니다.
    addCount(1L, binCount);
    return null; // 새로운 값을 추가한 경우 null을 반환합니다.
}

 

3. ConcurrentHashMap의 putVal 메서드의 호출 및 동작 원리

put, putIfAbsent 메서드 호출

public V put(K key, V value) {
    return putVal(key, value, false);
}

public V putIfAbsent(K key, V value) {
    return putVal(key, value, true);
}
  • ConcurrentHashMap에서 키-값 쌍을 추가하거나 업데이트할 때, 일반적으로 사용하는 메서드는 put, putIfAbsent, replace 등이다. 이러한 공개(public) 메서드는 내부적으로 실제 작업을 수행하기 위해 다른 메서드를 호출한다.
    • put(K key, V value) 메서드는 putVal(K key, V value, false)를 호출한다.
    • putIfAbsent(K key, V value) 메서드는 putVal(K key, V value, true)를 호출한다.

내부 putVal 메서드

final V putVal(K key, V value, boolean onlyIfAbsent) {
    // ...
}
  • putVal은 ConcurrentHashMap에서 삽입 또는 업데이트 로직을 처리하는 비공개(private) 메서드다.
  • 이 메서드의 세 번째 인자인 onlyIfAbsent는 키가 없을 때만 값을 삽입할지(true일 때) 아니면 키가 이미 있더라도 값을 무조건 삽입할지(false일 때)를 결정한다.
  • put이나 putIfAbsent에서 세 번째 인사를 true, false로 보냄에 따라 다른 동작을 한다.

헷갈릴 수 있는 부분

put이든 putIfAbsent든 모든 메서드는 내부적으로 putVal을 호출하여 처리한다.

  • put(K key, V value)를 호출할 때
    • putVal에 onlyIfAbsent 매개변수로 false를 전달한다.
    • 이 경우 putVal은 기존 값을 덮어쓰고 새로운 값으로 업데이트한다.

  • putIfAbsent(K key, V value)를 호출할 때
    • putVal에 onlyIfAbsent 매개변수로 true를 전달한다.
    • 이 경우 putVal은 키가 존재하지 않을 때만 값을 삽입하고, 키가 이미 존재하면 기존 값을 그대로 유지한다.

 

4. ConcurrentHashMap의 putVal 메서드에서 사용되는 CAS 기법이란 무엇인가?

CAS(Compare-And-Swap)란 무엇인가?

  • CAS(Compare-And-Swap)는 컴퓨터 과학에서 비동기적 동시성 제어 기법 중 하나로, 여러 스레드가 동시에 데이터를 수정하려고 할 때 데이터의 일관성을 보장하는 방법이다. CAS는 원자적(atomic) 연산으로, 멀티스레드 환경에서 안전하게 데이터를 읽고 쓸 수 있게 한다.

CAS의 작동 원리: 세 가지 값을 사용하여 연산을 수행한다.

  1. 메모리 주소: 변경하고자 하는 데이터의 위치를 가리킨다.
  2. 기존 예상 값(Old Value): 현재 메모리 주소에 저장된 값으로 예상되는 값이다.
  3. 새로운 값(New Value): 메모리 주소에 저장하고자 하는 새로운 값이다.

CAS 연산은 다음과 같이 동작합니다:

  • CAS 연산은 먼저 특정 메모리 위치의 현재 값을 읽는다. 이 값을 기대값(expected value)이라고 한다.
  • 읽어들인 현재 값이 스레드가 기대한 값과 동일한지 비교한다.
    • 만약 현재 값이 기대값과 일치한다면, 메모리 위치의 값을 새로운 값으로 원자적(atomic)으로 교체한다.
    • 만약 현재 값이 기대값과 일치하지 않으면, 다른 스레드가 그 사이에 값을 변경했음을 의미하므로 교체를 실패하고, 다시 시도하거나 다른 적절한 처리를 한다.

CAS 연산은 하드웨어 수준에서 원자적으로 수행된다.

 

원자적(Atomic) 연산의 정의

  • 원자적 연산이란 완전히 실행되거나 전혀 실행되지 않는 연산을 말한다. 즉, 연산이 시작되면 중간에 다른 연산이 개입할 수 없고, 연산 도중에 중단되지 않고 완료된다.

하드웨어 수준에서 CAS 연산이 원자적인 이유

  • 대부분의 현대 CPU는 CAS 연산을 지원하는 특수한 명령어(예: x86 아키텍처의 CMPXCHG 명령어)를 가지고 있다.
  • CMPXCHG 명령어는 메모리의 특정 주소에서 값을 읽고, 기대값과 비교한 후, 두 값이 일치하면 새 값으로 교체하는 과정을 단일 명령어로 처리한다.
  • 이 과정이 단일 명령어로 처리되기 때문에, CAS 연산이 실행되는 동안에는 다른 어떤 연산도 해당 메모리 주소에 접근할 수 없다. 이는 하드웨어적으로 보장된 원자성을 의미한다.

CAS 연산 중 다른 스레드가 값을 바꾸지 못하는 이유

  • CAS 명령어가 실행되는 동안, 해당 메모리 주소에 대한 독점적 접근권을 가진다. 즉, CAS 명령어가 실행되는 동안에는 다른 스레드나 프로세스가 해당 메모리 주소를 읽거나 쓸 수 없다.
  • 캐시 일관성 프로토콜(예: MESI 프로토콜)과 버스 락킹(Bus Locking) 같은 기술을 통해 여러 CPU 코어가 동시에 접근하려고 시도해도, CAS 명령어가 끝날 때까지 해당 메모리 주소는 보호된다.

결론:

  • CAS 연산은 하드웨어 수준에서 원자적으로 수행되며, CPU 명령어가 중간에 끊기지 않고 한 번에 수행되기 때문에, 메모리 주소를 읽고, 비교하고, 값을 교체하는 전체 과정이 한 번에 이루어진다.
  • 이 과정은 다른 어떤 스레드나 프로세스도 개입할 수 없게 보호되므로, CAS 연산은 항상 완전하게 수행되거나 전혀 수행되지 않는 두 가지 상태 중 하나만 가질 수 있다. 이를 통해 데이터의 일관성을 유지하며, 동시성 문제를 효과적으로 방지할 수 있다.

 

왜 CAS가 중요한가?

  1. 락이 필요 없음: CAS는 전통적인 락(lock) 메커니즘을 사용하지 않고도 동시성을 제어할 수 있다. 락을 사용하지 않기 때문에 데드락(교착 상태)이 발생하지 않고, 락을 획득하고 해제하는 비용을 줄일 수 있다.

  2. 높은 성능: CAS는 락보다 성능이 뛰어난 경우가 많다. 특히, 데이터 충돌이 적은 환경에서는 락을 사용하는 것보다 더 빠르게 동작한다.

  3. 비동기적 처리: CAS는 동기화 블록 내에서 코드가 실행되는 것을 방지하기 때문에, 다른 스레드가 기다릴 필요 없이 계속해서 실행할 수 있다. 이를 통해 어플리케이션의 전반적인 처리 속도가 향상된다.

CAS의 한계점

  • Busy-waiting: CAS는 반복적으로 메모리를 확인하고 수정하려고 시도하기 때문에, 값이 변경될 때까지 CPU를 사용할 수 있다. 이로 인해 Busy-waiting(반복 대기)가 발생하여 CPU 자원을 낭비할 수 있다.

  • ABA 문제: CAS는 메모리 값이 변경되었는지 확인하지만, 그 값이 중간에 변경되었다가 다시 원래 값으로 돌아오는 상황(예: A에서 B로 변경되고 다시 A로 변경됨)을 처리하지 못한다. 이를 ABA 문제라고 하며, 해결하기 위해 값과 함께 버전 정보를 사용하는 방법이 있다.

Java에서의 CAS 사용

  • Java에서는 java.util.concurrent 패키지의 여러 클래스에서 CAS를 사용하여 동시성을 제어한다.
  • AtomicInteger, AtomicReference, ConcurrentHashMap 등이 대표적인 예시다.
  • 이러한 클래스는 내부적으로 Unsafe 클래스의 CAS 메서드를 사용하여 비동기적으로 데이터를 업데이트한다.

5. ConcurrentHashMap에서는 CAS 기법을 어떻게 사용하고 있을까?

ConcurrentHashMap의 putVal 메서드에서 CAS (Compare-And-Swap)가 사용되는 부분은 다음과 같다.
  • putVal 메서드의 일부를 잘라온 것이므로 전체코드는 상단 또는 검색해보도록 하자.
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
    // 여기서 사용된다.
    if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
        break; // 락 없이 빈 버킷에 노드를 추가한 경우 반복을 종료합니다.
}

ConcurrentaHashMap의 putVal 메서드 내부의 CAS 동작 분석

  • 코드를 보면 먼저 tabAt(tab, i)로 계산된 인덱스 i에 해당하는 버킷이 null인지를 확인한다.
    • tabAt(tab, i)는 ConcurrentHashMap 내부에서 특정 인덱스 i에 있는 버킷(bucket)을 확인하는 메서드다.
    • 여기서 tab은 ConcurrentHashMap의 내부 배열을 나타내며, i는 해당 배열의 특정 위치(인덱스)를 의미한다.
    • tabAt(tab, i)의 결과가 null이면, 해당 위치에 아직 노드가 없다는 의미입니다. 즉, 해당 인덱스에 데이터가 삽입되지 않은 상태다.

  • tabAt(tab, i)의 결과가 null이면 해당 인덱스에 노드가 없음을 의미한다.
    • 버킷이 비어 있는 경우: tabAt(tab, i)의 결과가 null이라면, 이 인덱스 위치에 노드가 없다는 것을 의미한다.
    • 이 경우 casTabAt 메서드가 실행된다. casTabAt은 CAS 연산을 사용하여 해당 위치가 여전히 null일 때만 새로운 노드를 원자적으로 삽입하려고 시도한다.
@SuppressWarnings("unchecked")
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
    return (Node<K,V>)U.getReferenceAcquire(tab, ((long)i << ASHIFT) + ABASE);
}

@IntrinsicCandidate
public final Object getReferenceAcquire(Object o, long offset) {
    return getReferenceVolatile(o, offset);
}

@IntrinsicCandidate
public native Object getReferenceVolatile(Object o, long offset);
  • tabAt을 호출하면 네이티브 메서드가 사용된다. (실제로 위의 코드를 사용)
    • 네이티브 메서드(getReferenceAcquire와 getReferenceVolatile)는 tabAt과 casTabAt에서 메모리에 안전하게 접근하기 위해 사용된다.
    • tabAt는 Unsafe 클래스의 getReferenceAcquire 메서드를 사용해 배열의 특정 위치에서 현재 값을 읽어온다. 이 메서드는 자바 네이티브 인터페이스(JNI)를 통해 네이티브 코드로 구현된 getReferenceVolatile을 호출한다.
    • getReferenceVolatile은 자바 가상 머신(JVM)에서 직접 지원하는 네이티브 메서드로, 메모리의 가시성을 보장하며 동시성 문제를 방지한다.

putVal 메서드 내부의 (casTabAt) 호출

static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                    Node<K,V> c, Node<K,V> v) {
    return U.compareAndSetReference(tab, ((long)i << ASHIFT) + ABASE, c, v);
}

@IntrinsicCandidate
public final native boolean compareAndSetReference(Object o, long offset,
                                                   Object expected,
                                                   Object x);
  • casTabAt(tab, i, null, new Node<K,V>(hash, key, value))가 CAS 연산이다
  • casTabAt 메서드는 다음을 수행한다.
    • 현재 tab[i]의 값이 여전히 null인지 확인한다. (즉, 다른 스레드가 아직 해당 위치에 노드를 삽입하지 않았는지 확인)
    • 값이 null이면, 새로운 노드로 원자적으로 대체한다.
    • 성공하면 true를 반환하고, 그렇지 않으면(다른 스레드가 그 사이에 노드를 삽입한 경우) false를 반환한다.

  • 만약 casTabAt이 호출되면 내부적으로 compareAndSetReference라는 메서드를 호출하는데 이 메서드는 JVM에서 네이티브 코드로 구현되어 있으며, JNI(Java Native Interface)를 통해 자바 코드에서 네이티브 코드를 호출한다.

"쉽게 생각하면 여러 사람이 동시에 빈자리에 물건을 놓으려고 할 때, CAS는 "이 자리가 아직 비어있으면 내가 먼저 놓고, 이미 누가 놓아버렸으면 나는 안 놓는다."라는 원칙을 지킨다고 생각하면 된다. 그래서 서로 겹치지 않고 일관되게 행동할 수 있게 해 준다."

 

ConcurrentHashMap의 putVal 메서드에서 CAS를 사용하는 이유

  • 여기서 CAS를 사용하는 이유는 명시적 락(lock)을 사용하지 않고도 스레드 안전성(Thread Safety)을 보장하기 위해서다. 여러 스레드가 동시에 같은 버킷에 노드를 삽입하려고 할 수 있지만, CAS는 그중 오직 하나의 스레드만이 빈 버킷에 노드를 성공적으로 삽입할 수 있도록 보장한다.

  • CAS 연산이 실패하면(다른 스레드가 노드를 삽입한 경우), 메서드는 다른 방법으로 상황을 처리하도록 넘어간다. 예를 들어, 컨텐션이 발생한 경우에는 락을 사용하거나 진행 중인 리사이즈 작업을 돕는다.

 

참고: "컨텐션(Contention)"은 멀티스레드 프로그래밍에서 여러 스레드가 동시에 특정 자원(예: 메모리, 데이터 구조, 파일)에 접근하거나 수정하려고 할 때 발생하는 경쟁 상황을 말한다. 이 자원에 대한 접근이나 수정은 한 번에 하나의 스레드만 안전하게 수행할 수 있기 때문에, 여러 스레드가 동시에 접근하려고 하면 충돌이나 대기 상태가 발생할 수 있다.

 

6. ConcurrentHashMap의 putVal 메서드 분석

1. 입력 값 검증 및 해시 계산

// 추가된 코드 (설명할 부분)
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
  • 입력 값 검증: ConcurrentHashMap은 키와 값에 null을 허용하지 않는다. 따라서, key와 value가 null인 경우에는 NullPointerException을 던지도록 설계되었다. 이는 자바 컬렉션에서의 일반적인 제약 사항으로, null 값으로 인해 발생할 수 있는 예기치 않은 동작을 방지하기 위함이다.

  • 해시 계산: key.hashCode()를 호출하여 해시코드를 가져오고, spread 메서드를 통해 해시를 더 균등하게 분산시킨다. 이 과정은 해시 충돌을 줄여 해시 테이블의 성능을 최적화하기 위함이다.

 

2. 무한 루프와 테이블 초기화

// 이전 설명 코드
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;

// 추가된 코드 (설명할 부분)
for (Node<K,V>[] tab = table;;) {
    if (tab == null || (n = tab.length) == 0)
        tab = initTable();
    ...
}
  • 무한 루프 설정: 이 코드에서는 for 문의 조건을 생략하여 무한 루프를 만든다. 이는 특정 조건이 충족될 때까지 반복하기 위함이다. 반복을 종료하는 조건은 루프 내에서 명시적으로 break가 호출될 때다.

  • 테이블 초기화 확인: 해시 테이블 tab이 초기화되지 않았거나 길이가 0인 경우에는 initTable() 메서드를 호출하여 초기화를 한다. initTable() 메서드는 해시 테이블이 처음 사용되기 전에 준비되도록 한다. (아래의 메서드가 호출된다.)
private final Node<K,V>[] initTable() {

    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        if ((sc = sizeCtl) < 0)
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}
  • 참고로 ConcurrentHashMap에서 사용하는 "테이블"은 Node<K, V>[] 배열이다. 이는 ConcurrentHashMap 내부에서 사용되는 해시 테이블을 의미한다. (아래와 같이 ConcurrentHashMap 클래스 내부에 static class로 선언되어 있다.)
static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
}
  • 테이블 (table): ConcurrentHashMap의 해시 테이블로서, 키-값 쌍이 저장되는 노드(Node<K, V>) 객체들의 배열이다. 이 배열의 각 요소는 버킷(bucket)으로, 충돌된 키들이 체인이나 트리 구조로 연결될 수 있다.

  • Node 객체: Node<K, V>는 ConcurrentHashMap 내에서 각 엔트리를 나타내는 객체다. 키(K)와 값(V), 그리고 연결된 다음 노드를 가리키는 포인터를 가지고 있다. 이것은 연결 리스트나 트리 구조를 형성할 때 사용된다.

 

3. 비어있는 버킷에 노드 추가 (CAS 연산)

// 이전 설명 코드
for (Node<K,V>[] tab = table;;) {
    if (tab == null || (n = tab.length) == 0)
        tab = initTable();
        
    // 추가된 코드 (설명할 부분)
    else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
        if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
            break; // 락 없이 빈 버킷에 노드를 추가한 경우 반복을 종료합니다.
    }
    ...
    
}
  • 버킷 검사: 계산된 인덱스 i에 있는 버킷이 비어 있는지 확인한다. 비어 있다면 tabAt 메서드를 통해 해당 위치의 값을 가져온다.

  • CAS(Compare-And-Swap) 연산을 통한 노드 추가: 버킷이 비어 있으면 casTabAt 메서드를 사용해 원자적으로 새로운 노드를 추가한다. CAS는 멀티스레드 환경(동시성 환경)에서 락을 사용하지 않고도 안전하게 값을 업데이트할 수 있는 메커니즘이다. 만약 CAS가 성공하면, 노드가 추가되었으므로 break를 통해 반복문을 종료한다.

 

4. 리사이징 상태 확인 및 지원

for (Node<K,V>[] tab = table;;) {

    // ...

    // 이전 설명 코드
    else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
        if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
            break; // 락 없이 빈 버킷에 노드를 추가한 경우 반복을 종료합니다.
    }

    // 추가된 코드 (설명할 부분)
    else if ((fh = f.hash) == MOVED) {
        tab = helpTransfer(tab, f);
        ...
    }
    
}
  • 리사이징 상태 확인: 만약 f.hash가 MOVED 값이라면, 이는 테이블이 현재 리사이징 중임을 의미한다. 리사이징은 해시 테이블이 특정 임계치에 도달하여 크기를 늘려야 할 때 발생한다.

  • 리사이징 지원: helpTransfer 메서드를 호출하여, 현재 스레드가 리사이징 작업을 돕도록 한다. 이는 병렬 리사이징을 통해 여러 스레드가 동시에 테이블 크기를 조정할 수 있게 한다.

 

5. 조건부로 기존 노드 확인 및 반환

for (Node<K,V>[] tab = table;;) {

    // ...

    // 이전 설명 코드
    else if ((fh = f.hash) == MOVED) {
        tab = helpTransfer(tab, f);
    } 

    // 추가된 코드 (설명할 부분)
    else if (onlyIfAbsent 
               && fh == hash
               && ((fk = f.key) == key || (fk != null && key.equals(fk)))
               && (fv = f.val) != null) {
        return fv;
    }
    
}
  • 조건부 확인: onlyIfAbsent가 true일 경우, 즉 값이 이미 존재하는 경우 (삽입하지 않으려는 경우), 첫 번째 노드를 락 없이 확인한다. 이 최적화는 성능을 높이기 위해 락 없이 작업을 완료할 수 있을 때 사용된다.

  • 기존 값 반환: 키가 이미 존재하고, 값이 null이 아닌 경우 기존 값을 반환하고 메서드를 종료한다. 이 경우 추가 작업이 필요하지 않으므로, 빠르게 종료할 수 있다.

 

6. 락을 사용한 노드 삽입 및 업데이트

for (Node<K,V>[] tab = table;;) {

    // ...

    // 이전 설명 코드
    else if (onlyIfAbsent 
             && fh == hash
             && ((fk = f.key) == key || (fk != null && key.equals(fk)))
             && (fv = f.val) != null) {
        return fv;
    } 

    // 추가된 코드 (설명할 부분)
    else {
        V oldVal = null;
        synchronized (f) {
            if (tabAt(tab, i) == f) {
                if (fh >= 0) {
                    binCount = 1;
                    for (Node<K,V> e = f;; ++binCount) {
                        K ek;
                        if (e.hash == hash &&
                            ((ek = e.key) == key ||
                             (ek != null && key.equals(ek)))) {
                            oldVal = e.val;
                            if (!onlyIfAbsent)
                                e.val = value;
                            break;
                        }
                        Node<K,V> pred = e;
                        if ((e = e.next) == null) {
                            pred.next = new Node<K,V>(hash, key, value);
                            break;
                        }
                    }
                }
                else if (f instanceof TreeBin) {
                    Node<K,V> p;
                    binCount = 2;
                    if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) {
                        oldVal = p.val;
                        if (!onlyIfAbsent)
                            p.val = value;
                    }
                }
                else if (f instanceof ReservationNode)
                    throw new IllegalStateException("Recursive update");
            }
        }
        ...
    }
}
  • 락을 사용한 동기화: 버킷의 첫 번째 노드가 존재하면 해당 노드를 락(synchronized)으로 감싸 동기화한다. 이 과정은 멀티스레드 환경에서 여러 스레드가 동시에 같은 버킷에 접근할 경우 발생하는 충돌과 데이터 손상을 방지하기 위함이다.

  • 연결 리스트(Linked List) 처리: 버킷이 연결 리스트로 구성된 경우, 리스트의 각 노드를 순회하면서 동일한 키를 찾고, 발견하면 값을 갱신한다. 만약 키가 존재하지 않으면 리스트 끝에 새로운 노드를 추가한다.

  • 트리 구조 처리: 만약 노드가 트리 구조(TreeBin)로 되어 있다면, 트리 노드에 대해서도 키를 찾아 값을 업데이트하거나 새 노드를 추가한다. 이는 충돌이 많은 해시맵에서 성능을 향상시키기 위함이다.

  • 코드를 보면 f가 ReservationNode이면 예외를 발생시킨다. 이것은 ConcurrentHashMap에서 노드 삽입 중에 발생할 수 있는 재귀적 업데이트(recursive update)를 방지하기 위해 사용된다.
    • ConcurrentHashMap은 테이블이 확장될 때 여러 스레드가 동시에 기존 노드를 새로운 테이블로 옮겨야 할 때가 있다.
    • 이 과정에서, 스레드가 노드를 옮기면서 또 다른 노드를 추가하려고 하거나, 또는 노드가 옮겨지는 동안 다시 접근하려고 하면 재귀적 갱신(recursive update) 문제가 발생할 수 있다.
    • ReservationNode는 이런 재귀적 갱신을 방지하기 위해 해당 인덱스가 이미 처리 중임을 표시하는 역할을 한다.
    • 지금 코드에서는 현재 노드가 ReservationNode 타입인지 확인하고, 그렇다면 IllegalStateException을 던져서 재귀적인 업데이트를 방지한다.
    • ReservationNode가 발견되면 이는 해당 인덱스에 대해 특별한 처리가 이미 진행 중이라는 뜻이므로, 더 이상의 변경이 수행되지 않도록 예외를 던진다.

 

7. 트리화 검사 및 노드 수 증가

for (Node<K,V>[] tab = table;;) {

    // ...

    // 이전 설명 코드 (길어서 생략)
    else {
        V oldVal = null;
        synchronized (f) {
            if (tabAt(tab, i) == f) {
                if (fh >= 0) {
                    binCount = 1;
                    for (Node<K,V> e = f;; ++binCount) {
                        ...
                    }
                }
            else if (f instanceof TreeBin) {
                ...
            }
        }

        // 추가된 코드 (설명할 부분)
        if (binCount != 0) {
            if (binCount >= TREEIFY_THRESHOLD)
                treeifyBin(tab, i);
            if (oldVal != null)
                return oldVal;
            break;
        }
    }
}

// ...
  • 트리화 검사: 버킷에 있는 노드의 수(binCount)가 TREEIFY_THRESHOLD(기본값 8)를 초과하면, 연결 리스트를 트리 구조로 변환한다. 이는 해시 충돌이 많은 경우 검색 성능을 최적화하기 위해서다. 트리 구조는 리스트보다 빠른 검색 시간을 제공하여 성능 저하를 방지한다.

  • 기존 값 반환: 만약 이미 존재하는 키에 대한 값(oldVal)이 있다면, 새로운 값을 삽입하지 않고 기존 값을 반환한다. 이는 중복된 키에 대해 기존 값을 유지하면서 새로운 값으로 덮어쓰지 않도록 하기 위함이다.

  • 반복 종료: break는 위의 조건들이 충족되거나 필요한 작업이 완료되었을 때 루프를 종료하기 위한 것이다. 이 시점에서 더 이상 추가적인 처리가 필요하지 않으므로 반복을 종료하고 다음 단계로 넘어간다.

 

8. 노드 수 증가 및 종료 처리 (마지막)

for (Node<K,V>[] tab = table;;) {
 
     // ...
 
           // 이전 설명 코드
           if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }   
}

// 추가된 코드 (설명할 부분)
addCount(1L, binCount);
return null;
  • 노드 수 증가 (addCount): 새로운 노드가 추가될 때마다 해시맵의 전체 노드 수를 증가시키고, 필요에 따라 테이블 크기를 조정하여 해시 충돌을 줄이고 성능을 최적화한다.
    • addCount 메서드는 ConcurrentHashMap의 전체 노드 수를 증가시키고, 특정 조건이 만족되면 해시 테이블을 리사이징(크기 조정)하는 기능을 담당한다. 멀티스레드 환경에서 동시성 제어를 유지하면서도 효율적으로 노드 수를 관리하고 테이블 크기를 조정하기 위한 복잡한 메커니즘을 사용한다.

  • 결과 반환 (return null): 새로운 키-값 쌍이 성공적으로 추가되었음을 나타내기 위해 null을 반환한다. 이를 통해 ConcurrentHashMap의 삽입 작업이 명확히 처리되었음을 보장한다.

 

7. putVal 메서드 내부 동작: 클라이언트 호출 예시 (이미 키가 존재)

이미 키가 있을 때 (동시 접근)

  • ConcurrentHashMap에 이미 key1이라는 키가 있고, 그 값은 value0이라고 가정한다. 두 개의 스레드, Thread A와 Thread B가 동시에 key1에 대한 값을 업데이트하려고 시도한다.

내부 동작 1 (put 메서드에서 putVal을 호출하는 시나리오)

 

1. Thread A가 putVal(key1, value1, false)을 호출한다.

  • 입력 값 검증:
    • key1과 value1이 null이 아니므로 NullPointerException이 발생하지 않는다.

  • 해시 계산:
    • key1.hashCode()를 사용하여 해시 코드를 계산하고, spread 메서드를 통해 해시 값을 분산시켜 해시 테이블의 올바른 인덱스를 찾는다.

  • 테이블 초기화:
    • 만약 내부 테이블(table)이 초기화되지 않았거나 크기가 0이라면, initTable() 메서드를 호출하여 테이블을 초기화한다.

  • 노드 추가 시도:
    • key1의 해시 인덱스에 해당하는 버킷을 확인하고, 버킷이 비어있지 않음을 확인한다(이미 key1이 존재함).

  • 기존 노드 확인 및 업데이트:
    • key1에 해당하는 노드가 이미 존재하므로, Thread A는 해당 노드를 synchronized 블록으로 잠그고 값을 value1로 업데이트한다.

  • 락을 사용한 동기화:
    • 해당 노드를 락으로 감싸 동기화하여 다른 스레드가 접근하지 못하도록 한다.

2. Thread B가 동시에 putVal(key1, value2, false)를 호출한다.

  • 입력 값 검증과 해시 계산:
    • Thread B도 마찬가지로 key1과 value2가 null이 아니기 때문에 NullPointerException이 발생하지 않으며, 해시 계산을 통해 동일한 인덱스를 찾는다.

  • 노드 추가 시도:
    • Thread B는 해당 인덱스에서 노드가 이미 잠겨 있음을 발견한다.(이 시점에서 Thread A가 락을 걸고 있음).

  • 락 대기:
    • Thread A가 해당 노드의 락을 해제할 때까지 Thread B는 대기 상태에서 벗어나 다시 실행된다.

  • 락 해제 후 재실행:
    • Thread A가 업데이트를 완료하고 락을 해제하면, Thread B는 다시 실행을 재개한다.

  • 기존 값 확인 및 업데이트:
    • Thread B는 현재 노드의 값을 확인한다. Thread B가 put 메서드를 호출했다면, 현재 값을 확인하고 새로운 값 value2로 업데이트하며, 이전 값(value1)을 반환한다.

주요 포인트

  • Thread A가 노드를 락으로 잠그고 key1의 값을 value1로 업데이트한다.
  • Thread B는 Thread A가 락을 해제할 때까지 대기한다.
  • Thread B가 재실행되었을 때, Thread A가 업데이트한 value1을 확인하고, 사용된 메서드에 따라 다른 응답을 한다.
    • put 메서드를 사용한 경우(지금 상황): Thread B는 값을 value2로 덮어쓰고, 이전 값(value1)을 반환한다.
    • 참고로 put 메서드가 이전 값을 반환하는 이유는 새로운 값으로 대체되기 전의 상태를 확인할 수 있도록 하기 위함이다.
    • putIfAbsent 메서드를 사용한 경우: Thread B는 key1에 이미 값이 존재하므로 값을 변경하지 않고 기존 값(value1)을 반환한다.

 

결과

  • Thread A가 key1의 값을 value1로 업데이트한다.
  • Thread B는 이후 Thread A의 업데이트 결과를 확인하고, 사용된 메서드에 따라 적절한 작업을 수행한다: value2로 덮어쓰거나, 기존 값을 유지한다.

 

8. putVal 메서드 내부 동작: 클라이언트 호출 예시 (키가 없을 때)

키가 없을 때 (동시 접근)

  • Thread C와 Thread D는 동시에 putVal 메서드를 호출하여 key2에 대해 값을 삽입하려고 한다. 이때, ConcurrentHashMap에는 처음에 key2라는 키가 존재하지 않는다.

내부 동작 1 (put 메서드에서 putVal을 호출하는 시나리오)

 

1. Thread C가 putVal(key2, value3, false)을 호출한다.

  • 입력 값 검증:
    • key2와 value3이 null이 아니므로 예외가 발생하지 않는다. (key2와 value3은 외부에서 ConcurrentHashMap에 삽입하려고 전달한 값이기 때문에 null이 아니라는 의미다.)

  • 해시 계산:
    • key2의 해시 코드를 계산하고, spread 메서드를 사용하여 이 해시 값을 테이블 내 위치로 변환한다.

  • 테이블 초기화:
    • 테이블이 초기화되지 않았거나 길이가 0이라면, initTable() 메서드를 호출하여 테이블을 초기화한다.

  • 노드 추가 시도: 
    • 계산된 인덱스 위치에 버킷이 비어 있는지 확인한다. (tabAt 메서드를 사용)
    • 버킷이 비어 있으면 CAS(Compare-And-Swap) 연산을 사용하여 새로운 노드를 원자적으로 추가한다.
    • Thread C가 성공적으로 노드를 추가하면 반복문을 종료한다.

2. Thread D가 동시에 putVal(key2, value4, false)를 호출한다.

  • 입력 값 검증과 해시 계산:
    • Thread D도 마찬가지로 key2와 value4가 null이 아니므로 해시 계산을 통해 동일한 위치를 찾는다.

  • 노드 추가 시도:
    • Thread D는 계산된 인덱스 위치에서 Thread C가 CAS 연산을 통해 이미 key2를 추가했음을 발견한다. 즉, Thread D는 동일한 인덱스에 key2가 이미 존재하는 것을 감지한다.

  • 기존 노드 확인 및 업데이트:
    • Thread D는 이미 추가된 key2에 대한 노드를 확인한다.
    • put 메서드를 호출한 경우 Thread D는 기존 값을 덮어쓰고 새 값을 value4로 업데이트하고, 이전 값(value3)을 반환한다.
    • 참고로 put 메서드가 이전 값을 반환하는 이유는 새로운 값으로 대체되기 전의 상태를 확인할 수 있도록 하기 위함이다.

 

내부 동작 2 (putIfAbsent 메서드에서 putVal을 호출하는 시나리오)

 

1. Thread C가 putVal(key2, value3, true)를 호출

  • 입력 값 검증:
    • key2와 value3이 null이 아니므로 NullPointerException이 발생하지 않는다.

  • 해시 계산 및 테이블 초기화:
    • key2의 해시 코드를 계산하고, 이를 spread 메서드를 통해 해시 값을 분산시켜 해시 테이블의 인덱스를 찾는다.
    • 테이블이 초기화되지 않았거나 길이가 0이면 initTable() 메서드를 호출하여 테이블을 초기화한다.

  • 노드 추가 시도:
    • 계산된 인덱스 위치에 있는 버킷이 비어 있는지 확인한다 (tabAt 메서드를 사용).
    • 버킷이 비어 있으면, CAS(Compare-And-Swap) 연산을 사용해 새로운 노드를 원자적으로 추가한다. Thread C가 성공적으로 노드를 추가하면 반복문을 종료한다.

2. Thread D가 동시에 putVal(key2, value4, true)를 호출

  • 입력 값 검증과 해시 계산:
    • Thread D도 마찬가지로 key2와 value4가 null이 아니며, 해시 계산을 통해 동일한 인덱스를 찾는다.

  • 노드 추가 시도:
    • Thread D는 계산된 인덱스 위치에서 Thread C가 CAS 연산을 통해 이미 key2를 추가했음을 발견한다. 즉, Thread D는 동일한 인덱스에 key2가 이미 존재하는 것을 감지한다.

  • 기존 노드 확인 및 처리:
    • Thread D는 key2에 대한 노드가 이미 존재하는 것을 확인한다.
    • putIfAbsent 메서드를 호출했기 때문에, Thread D는 기존 값이 이미 존재할 경우 새로운 값을 삽입하지 않고, 기존 값(value3)을 그대로 반환한다.
    • 따라서, Thread D는 새로운 값을 삽입하지 않으며, 기존 값 value3을 반환한다.

결과

  • Thread C가 먼저 실행되어 key2를 추가한다.
  • Thread D는 Thread C가 이미 key2를 추가했음을 확인하고, 사용된 메서드에 따라 적절한 조치를 취한다.
    • put 메서드를 사용하는 경우: Thread D는 기존 값을 덮어쓰고 새 값을 삽입하며, 이전 값을 반환한다.
    • putIfAbsent 메서드를 사용하는 경우: Thread D는 값이 이미 존재하기 때문에 값을 업데이트하지 않고 기존 값을 반환한다.

 

 

 

오라클 가이드 자료

 

ConcurrentHashMap (Java SE 21 & JDK 21)

Type Parameters: K - the type of keys maintained by this map V - the type of mapped values All Implemented Interfaces: Serializable, ConcurrentMap , Map A hash table supporting full concurrency of retrievals and high expected concurrency for updates. This

docs.oracle.com

 

반응형

'JAVA' 카테고리의 다른 글

[Java] Stream의 mutable, immutable 리스트 변환 (toList)  (3) 2024.08.30
[Java] 동시성 제어가 가능한 CopyOnWriteArrayList와 일반 ArrayList의 차이점  (3) 2024.08.30
[OOP] 단일 책임 원칙, 응집도, 관심사  (12) 2024.03.06
[Java] HTTP 서버 만들기: 멀티스레딩과 네트워크 최적화하기  (39) 2024.01.13
[JMeter] 멀티스레딩 vs 스레드 풀: Java로 만든 HTTP 서버 성능 테스트  (27) 2024.01.03

티스토리툴바