CAS - 동기화와 원자적 연산

황상익·2024년 10월 22일

Inflearn JAVA

목록 보기
52/61

원자적 연산 소개

멀트 스레드 상황에서 다른 스레드 간섭 없이 안전하게 처리되는 연산 

volatile int i = 0

i = 1은 쪼갤 수 없는 연산 = 원자적 연산 

i = i + 1;

순서가 나뉘어 진행되기 때문에 원자적 연산이 아니다.
원자적 연산이 아닌 경우, synchromized 블럭이나, lock등을 사용하는데 안전한 임계 영역을 생성해야 함

원자적 연산 - 시작

package chap47.cas.increment;

public interface IncrementInteger {

    void increment();

    int get();
}

package chap47.cas.increment;

public class BasicInteger implements IncrementInteger{

    private int value; //인터페이스 필드이기 떄문에, 여러 스레드 공유 가능

    @Override
    public void increment() {
        value++;
    }

    @Override
    public int get() {
        return value;
    }
}

package chap47.cas.increment;

import java.util.ArrayList;
import java.util.List;

import static chap41.util.ThreadUtils.sleep;

public class IncrementThreadMain {

    public static final int THREAD_COUNT = 1000;

    public static void main(String[] args) throws InterruptedException {

        //test(new BasicInteger());
        //test(new VolatileInteger());
        //test(new SyncInteger());
        test(new MyAtomicInteger());
    }

    //원자적이지 않은 value++을 호출했기 때문
    private static void test(IncrementInteger incrementInteger)
            throws InterruptedException {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                sleep(10); //다른 스레드와 동시에 실행을 위해 잠깐 쉼
                incrementInteger.increment();
            }
        };

        //1000개가 동시에 수행
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < THREAD_COUNT; i++) {
            Thread thread = new Thread(runnable);
            threads.add(thread);
            thread.start();
        }

        for (Thread thread : threads){
            thread.join();
        }

        int result = incrementInteger.get();
        System.out.println(incrementInteger.getClass().getSimpleName() + "result : " + result);
    }
}

원자적 연산 - volatile, synchronized

package chap47.cas.increment;

public class VolatileInteger implements IncrementInteger{

    private volatile int value; //인터페이스 필드이기 떄문에, 여러 스레드 공유 가능

    @Override
    public void increment() {
        value++;
    }

    @Override
    public int get() {
        return value;
    }
}

package chap47.cas.increment;

import java.util.ArrayList;
import java.util.List;

import static chap41.util.ThreadUtils.sleep;

public class IncrementThreadMain {

    public static final int THREAD_COUNT = 1000;

    public static void main(String[] args) throws InterruptedException {

        //test(new BasicInteger());
        //test(new VolatileInteger());
        //test(new SyncInteger());
        test(new MyAtomicInteger());
    }

    //원자적이지 않은 value++을 호출했기 때문
    private static void test(IncrementInteger incrementInteger)
            throws InterruptedException {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                sleep(10); //다른 스레드와 동시에 실행을 위해 잠깐 쉼
                incrementInteger.increment();
            }
        };

        //1000개가 동시에 수행
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < THREAD_COUNT; i++) {
            Thread thread = new Thread(runnable);
            threads.add(thread);
            thread.start();
        }

        for (Thread thread : threads){
            thread.join();
        }

        int result = incrementInteger.get();
        System.out.println(incrementInteger.getClass().getSimpleName() + "result : " + result);
    }
}

volatile은 여러 CPU 사이에 발생하는 캐시 메모리와 메인 메모리가 동기화 되지 않는 문제 해결
CPU의 캐시 메모리를 무시하고 메인 메모리를 직접 연결
캐시 메모리가 영향은 줄 수 있지만, 캐시메모리를 사용하지 않고 메인 메모리를 사용해도 문제

synchronized

package chap47.cas.increment;

public class SyncInteger implements IncrementInteger{

    private int value; //인터페이스 필드이기 떄문에, 여러 스레드 공유 가능

    @Override
    public synchronized void increment() {
        value++;
    }

    @Override
    public synchronized int get() {
        return value;
    }
}

package chap47.cas.increment;

import java.util.ArrayList;
import java.util.List;

import static chap41.util.ThreadUtils.sleep;

public class IncrementThreadMain {

    public static final int THREAD_COUNT = 1000;

    public static void main(String[] args) throws InterruptedException {

        //test(new BasicInteger());
        //test(new VolatileInteger());
        //test(new SyncInteger());
        test(new MyAtomicInteger());
    }

    //원자적이지 않은 value++을 호출했기 때문
    private static void test(IncrementInteger incrementInteger)
            throws InterruptedException {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                sleep(10); //다른 스레드와 동시에 실행을 위해 잠깐 쉼
                incrementInteger.increment();
            }
        };

        //1000개가 동시에 수행
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < THREAD_COUNT; i++) {
            Thread thread = new Thread(runnable);
            threads.add(thread);
            thread.start();
        }

        for (Thread thread : threads){
            thread.join();
        }

        int result = incrementInteger.get();
        System.out.println(incrementInteger.getClass().getSimpleName() + "result : " + result);
    }
}

여전히 안전한 임계 영역을 만들고 value ++ 연산을 수행했더니 정확한 값이 나온다.

원자적 연산 - AtomicInteger

package chap47.cas.increment;

import java.util.concurrent.atomic.AtomicInteger;

public class MyAtomicInteger implements IncrementInteger {

    AtomicInteger atomicInteger = new AtomicInteger(0); //초기값 지정 생략하면 0

    private int value; //인터페이스 필드이기 떄문에, 여러 스레드 공유 가능

    @Override
    public void increment() {
        atomicInteger.incrementAndGet(); // 값을 하나 증가 반환
    }

    @Override
    public int get() {
        return atomicInteger.get(); // 현재 값을 반환
    }
}

원자적 Integer 

package chap47.cas.increment;

import java.util.ArrayList;
import java.util.List;

import static chap41.util.ThreadUtils.sleep;

public class IncrementThreadMain {

    public static final int THREAD_COUNT = 1000;

    public static void main(String[] args) throws InterruptedException {

        //test(new BasicInteger());
        //test(new VolatileInteger());
        //test(new SyncInteger());
        test(new MyAtomicInteger());
    }

    //원자적이지 않은 value++을 호출했기 때문
    private static void test(IncrementInteger incrementInteger)
            throws InterruptedException {

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                sleep(10); //다른 스레드와 동시에 실행을 위해 잠깐 쉼
                incrementInteger.increment();
            }
        };

        //1000개가 동시에 수행
        List<Thread> threads = new ArrayList<>();
        for (int i = 0; i < THREAD_COUNT; i++) {
            Thread thread = new Thread(runnable);
            threads.add(thread);
            thread.start();
        }

        for (Thread thread : threads){
            thread.join();
        }

        int result = incrementInteger.get();
        System.out.println(incrementInteger.getClass().getSimpleName() + "result : " + result);
    }
}

AtomicInteger는 멀티 스레드 환경에서 안전, 증감 연산을 제공

성능 Test

package chap47.cas.increment;

import java.util.concurrent.atomic.AtomicInteger;

public class IncrementPerformanceMain {
    private static final long COUNT = 100_000_000;

    public static void main(String[] args) {
        test(new BasicInteger());
        test(new VolatileInteger());
        test(new SyncInteger());
        test(new MyAtomicInteger());
    }

    private static void test (IncrementInteger incrementInteger) {
        long startMs = System.currentTimeMillis();
        for (int i = 0; i < COUNT; i++) {
            incrementInteger.increment();
        }

        long endMs = System.currentTimeMillis();
        System.out.println(incrementInteger.getClass().getSimpleName() + ": ms = " + (endMs - startMs));
    }
}

BasicInteger: ms = 4
VolatileInteger: ms = 175
SyncInteger: ms = 250
MyAtomicInteger: ms = 178

BasicInteger가 가장 빠름 : CPU 캐시를 적극 활용하기 때문, 하지만 안전한 임계영역도 없고, volatile도 사용하지 않기 때문에 멀티 스레드 상황에는 사용 X
단일 스레드가 사용하는 경우 효과적

VolatileInteger
volatile을 사용해서 CPU 캐시를 사용하지 않고 메인 메모리를 사용
안전한 임계영역이 없기 때문에 멀티스레드 상황에는 사용 X

SyncInteger
안전한 임계 영역이 있기 때문에 멀티스레드 상황에도 안전하게 사용 가능
MyAtomicInteger 보다는 성능 느림

MyAtomicInteger
멀티스레드 상황에 안전하게 사용 가능
성능도 synchronized, lock(ReentrantLock)을 사용하는 경우보다 훨씬 빠름

CAS 연산1

원자적인 연산을 수행할 수 있는 방법, -> CAS 연산.
락을 사용하지 않기 때문에 락 프리 기법이라고 함
CAS 연산은 락을 완전히 대체하는 것은 아니고 작은 단위의 일부 영역에 적용 

package chap47.cas;

import java.util.concurrent.atomic.AtomicInteger;

public class CasMainV1 {
    public static void main(String[] args) {
        AtomicInteger atomicInteger = new AtomicInteger();
        System.out.println("start value = " + atomicInteger.get());

        boolean result1 = atomicInteger.compareAndSet(0, 1); //비교하는 값이 0이면 1로 setting
        System.out.println("result1 = " + result1 + "; value = " + atomicInteger.get());

        boolean result2 = atomicInteger.compareAndSet(0, 1); //atomicInteger에 값이 만약 0이면 1로 Setting
        System.out.println("result2 = " + result2 + "; value = " + atomicInteger.get()); //현재 0이 아니라면 변경되지 않음 / false
    }
}

compareAndSet(0, 1)
atomicInteger가 갖고 있는 값이 0이면 1로 변경, 현재 값이 0이면 1로 변경 되면서 true
현재 값이 0이 아니라면, atomicInteger 값은 변경되지 않는다. 이 경우 false

성공

CPU는 두 과정을 하나의 원자적 명령을 만들기 위해 다른 스레드가 현재 작업중인 스레드의 값을 변경하지 못하게 물리적으로 막는다.

실패

CAS 연산 메모리에 있는 값이 기대하는 값, 원하는 값으로 변경
atomicInteger 내부에 있는 value 값이 0이라면 1로 변경
하지만 value는 이미 1이기 때문에 변경 X

CAS 연산2

package chap47.cas;

import java.util.concurrent.atomic.AtomicInteger;

import static chap41.util.MyLogger.log;

public class CasMainV2 {
    public static void main(String[] args) {
        AtomicInteger atomicInteger = new AtomicInteger();
        System.out.println("start value = " + atomicInteger.get());

        int resultValue1 = incrementAndGet(atomicInteger);
        System.out.println("resultValue1 = " + resultValue1);

        int resultValue2 = atomicInteger.incrementAndGet();
        System.out.println("resultValue2 = " + resultValue2); //현재 0이 아니라면 변경되지 않음 / false
    }

    private static int incrementAndGet(AtomicInteger atomicInteger) {
        int getValue;
        boolean result;

        do {
            getValue = atomicInteger.get(); //아토믹 인티져에서 읽은 값을 getValue에 넣음 = 0 //Thread 1 : 0
            log("getValue = " + getValue); //Thread 2가 값을 갑자기 1로 변경하면 -> 실패
            result = atomicInteger.compareAndSet(getValue, getValue + 1); //앍은 값을 읽고 update 시점에 같다면 읽은 값에 + 1 (아토믹 인티져가 갖고 있는 값이 0일 때만 증가 시킬 것이다)
            log("result = " + result);

        } while (!result);

        return getValue + 1;
    }
}

CAS 연산을 사용하면 여러 스레드가 같은 값을 사용하는 상황에도 락을 걸지 않고 안전하게 값을 증가.
incrementAndGet 첫 실행

incrementAndGet 첫 실행

CAS 연산3

package chap47.cas;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import static chap41.util.MyLogger.log;
import static chap41.util.ThreadUtils.sleep;

public class CasMainV3 {
    private static final int THREAD_COUNT = 2;

    public static void main(String[] args) throws InterruptedException {
        AtomicInteger atomicInteger = new AtomicInteger(0);
        System.out.println("start Value = " + atomicInteger.get());

        Runnable runnable = new Runnable() {

            @Override
            public void run() {
                incrementAndGet(atomicInteger);
            }
        };

        List<Thread> threads = new ArrayList<>(THREAD_COUNT);
        for (int i = 0; i < THREAD_COUNT; i++) {
            Thread thread = new Thread(runnable);
            threads.add(thread);
            thread.start();
        }

        for (Thread thread : threads) {
            thread.join();
        }

        int result = atomicInteger.get();
        System.out.println(atomicInteger.getClass().getSimpleName() + " resultValue = " + result);
    }

    private static int incrementAndGet(AtomicInteger atomicInteger) {
        int getValue;
        boolean result;

        do {
            getValue = atomicInteger.get(); //아토믹 인티져에서 읽은 값을 getValue에 넣음 = 0 //Thread 1 : 0
            sleep(100); //스레드 동시 실행을 위한 대기
            log("getValue = " + getValue); //Thread 2가 값을 갑자기 1로 변경하면 -> 실패
            result = atomicInteger.compareAndSet(getValue, getValue + 1); //앍은 값을 읽고 update 시점에 같다면 읽은 값에 + 1 (아토믹 인티져가 갖고 있는 값이 0일 때만 증가 시킬 것이다)
            log("result = " + result);

        } while (!result);

        //처음 읽었던 값 그대로
        return getValue + 1;
        //다른 스레드가 값을 변경 할 수도 있다.
        //return atomicInteger.get();
    }
}

dowile에서 thread1이 0 -> 1로 값을 변경하면서 탈출
하지만 thread2는 실패 -> 값이 이미 변경되었기 때문, 다시 do while 문에서 작업 실행
현재 값은 1이기 때문에 결과 값은 2로 변경

CAS(Compare-And-Swap)와 락(Lock) 방식의 비교
1. 락(Lock) 방식
데이터에 접근하기 전에 항상 락을 획득
다른 스레드의 접근을 막음
"다른 스레드가 방해할 것이다"라고 가정

  1. CAS(Compare-And-Swap) 방식
    락을 사용하지 않고 데이터에 바로 접근
    충돌이 발생하면 그때 재시도
    "대부분의 경우 충돌이 없을 것이다"라고 가정

CAS 락 구현

package chap47.cas.spinlock;

import static chap41.util.MyLogger.log;
import static chap41.util.ThreadUtils.sleep;

//스레드 둘다 락을 획득하는 문제 발생!
//스레드 둘다 비즈니스 로직 실행
//임계영역이 보호 되지 않는다.
public class SpinLockBad {
    private volatile boolean lock = false; // 한 스레드만 락 획득

    //락 사영 여부 확인 & 락의 값 변경 -> 원시적이지 않음
    public void lock() {
        log("락 획득 시도");

        while (true) {
            if (!lock) { //락 사용 여부 확인
                sleep(100);
                lock = true; //락 값 변경
                break;
            } else {
                //락을 획득할 때까지 스핀 대기 (Runnable 상태로 계속 대기)
                log("락 획득 실패 - 스핀 대기");
            }
        }

        log("락 획득 완료");
    }

    public void unlock() {
        lock = false;
        log("락 반납 완료");
    }
}

문제점은 임계영역이 전혀 보호 되지 않고 있음, Thread 두개가 동시에 접근하는 상황 발생 

package chap47.cas.spinlock;

import static chap41.util.MyLogger.log;
import static chap41.util.ThreadUtils.sleep;

public class SpinLockMain {
    public static void main(String[] args) {
        //SpinLockBad spinLock = new SpinLockBad();
        SpinLock spinLock = new SpinLock();

        Runnable task = new Runnable() {
            @Override
            public void run() {
                spinLock.lock();
                try {
                    //critical section
                    sleep(1); //오래 걸리는 로직에서는 스핀 락 사용 X
                    log("비즈니스 로직 실행");
                } finally {
                    spinLock.unlock();
                }
            }
        };

        Thread t1 = new Thread(task, "Thread - 1");
        Thread t2 = new Thread(task, "Thread - 2");
        t1.start();
        t2.start();
    }
}

CAS 락 구현 2

package chap47.cas.spinlock;

import java.util.concurrent.atomic.AtomicBoolean;

import static chap41.util.MyLogger.log;
import static chap41.util.ThreadUtils.sleep;

public class SpinLock {
    private final AtomicBoolean lock = new AtomicBoolean(false);

    public void lock() {
        log("락 획득 시도");

        //!true -> false -> while문에서 빠져나온다.
        while (!lock.compareAndSet(false, true)) { //락이 아무도 사용하고 았지 않는다면 true & CAS 연산을 통해 원자적 변경
            log("락 획득 실패 - 스핀 대기");
        }
        log("락 획득 완료");
    }

    public void unlock() {
        lock.set(false);
        log("락 반납 완료");
    }
}

CAS를 사용해서 원자적 연산을 만든 덕분에 무거운 동기화 작업을 가벼운 락으로 해결
무겁다는 의미는 스레드가 락을 획득하지 못하면 BLOCKED , WAITING 등으로 상태가 변함
대기 상태의 스레드를 깨워야 하는 무겁고 복잡한 과정이 추가로 들어간다

단점
sleep을 걸게되면 계속 spin 하게 된다
Blocked나 waiting 상태로 들어가는 것이 아닌, runnable 상태에서 계속 대기를 해야 하기 때문

언제 사용하는게 적절할까??
안전한 임계 영역이 필요하지만, 연산이 길지 않고 매우매우매우! 짧게 끝날 때 사용해야 한다.
예를 들어 숫자 값의 증가, 자료 구조의 데이터 추가와 같이 CPU 사이클이 금방 끝나는 연산에 사용하면 효과적이다.

DB의 결과를 대기하거나, 서버 요청을 기다리는 등 오래 기다리는 작업에서 사용하면 CPU 사용이 증가, 최악의 결과 발생 가능

profile
황상익
개발자를 향해 가는 중입니다~! 항상 겸손
이전 포스트

생산자 소비자 문제2

다음 포스트

동시성 컬렉션

0개의 댓글