멀티 쓰레드 사용 시 꼭 알아야 할 내용 - Thread-Safe

멀티 쓰레드는 프로그램의 성능을 향상시키는 가장 확실한 방법이다. 하지만 여러 쓰레드가 자원을 공유한다면 예기치 않은 문제를 발생시킬 수 있다. 

 

Java에선 어떻게 Thread-Safe하게 개발할 수 있도록 지원을 하는지, 다양한 방법들에 대해 간단한 코드와 함께 알아보고

 

실제 개발 중이던 Thread-UnSafe한 코드를 수정해볼 것이다. 

 

 

1. Thread Safe (쓰레드 세이프) 란?

• 멀티 쓰레드 프로그래밍에서, 어떤 공유 자원에 여러 쓰레드가 동시에 접근해도, 프로그램 실행에 문제가 없는 상태를 의미한다.
• Race Condition 상황이 발생해도 실행에 문제가 생기지 않는 것을 의미한다.
  • Race Condition 이란?
  • 경쟁 상태로, 공유 자원에 대해 두 개 이상의 스레드가 동시에 읽거나 쓰는 상황을 의미한다.

 

 

Thread Safe를 지키는 방법으론 크게 4가지가 있다. 

1. Re-entrancy(재진입성)

• 정적(전역) 변수를 사용하면 안 된다.
• 정적(전역) 변수의 주소를 반환하면 안 된다.
• 호출자가 호출 시 제공한 매개 변수만으로 동작해야 한다.
• 싱글톤 객체의 잠금에 의존하면 안 된다.
• 다른 비-재진입 함수를 호출하면 안 된다.

2. Thread-local Storage

• 공유자원의 사용을 최대한 줄이고 각각의 Thread에서만 접근 가능한 저장소를 사용하여 동시 접근을 막는 것

3. Mutual Exclusion

• Lock을 통해 공유자원을 제어하는 것
• 뮤텍스, 세마포어 등을 이용한다.

4. Atomic Operation Atomic

• 연산을 사용하는 것
• 더이상 쪼갤 수 없는 가장 작은 단위의 명령어를 실행하는 것
• 도중에 인터럽트가 발생할 수 없다.

 

 

 

 

 

2. Thread-Safe가 지켜지지 않을 때

코드를 Thread-Safe하게 작성하지 않으면 어떻게 되는지 살펴보기 위해 아래와 같은 간단한 코드를 작성하였다. 

아래 코드는 다음과 같다.

• 최소 100명의 예약이 차야지 버스를 운행한다.
• 100명 이하일 경우, 경고 메세지와 함께 현재 예약 인원수를 보여준다. 

@Getter
public class Bus {
    private int minOccupancy = 100;
    private int reservation = 0;

    public void getBusTicket() {
        try {
            Thread.sleep(100);
            reservation++;
            if (reservation <= minOccupancy) {
                Thread.sleep(100);
                System.out.println("최소 인원인 100명을 넘어야 합니다. 현재 예약 인원: " + reservation);
            }
        } catch (InterruptedException e) {
            System.out.println("ERROR!");
        }
    }
}

 

 

 

만약 500명이 동시에 버스를 예매한다고 생각해보자. 

100명 이하일 때는 "최소 인원인 100명을 넘어야 합니다. 현재 예약 인원: x" 라는 문구가 출력되야 하고,

최종 예약 인원은 500명이어야 한다. 

class BusTest {
    private final ExecutorService THREAD_POOL = Executors.newFixedThreadPool(100);

    @Test
    void 다수의_사람들이_동시에_버스티켓을_구매한다() throws InterruptedException {

        int N = 500;
        CountDownLatch latch = new CountDownLatch(N);
        Bus bus = new Bus();

        for (int i = 0; i < N; i++) {
            THREAD_POOL.execute(() -> {
                bus.getBusTicket();
                latch.countDown();
            });
        }

        latch.await();

        int busReservation = bus.getReservation();
        System.out.println("======= Total reservation: " + busReservation + " =======");
        assertThat(busReservation).isNotEqualTo(N);
    }
}

 

@Test
void 다수의_사람들이_동시에_버스티켓을_구매한다() throws InterruptedException {

    int N = 500;
    CountDownLatch latch = new CountDownLatch(N);
    Bus bus = new Bus();

    for (int i = 0; i < N; i++) {
        THREAD_POOL.execute(() -> {
            bus.getBusTicket();
            latch.countDown();
        });
    }

    latch.await();

    int busReservation = bus.getReservation();
    System.out.println("======= Total reservation: " + busReservation + " =======");
    assertThat(busReservation).isEqualTo(N);
}

 

하지만 결과는 아래와 같이 나왔다. 

 

 

 

1. 100명 이상일 때도 문구가 출력된 이유

1번 쓰레드가 if문을 통과한 시점에 2번 쓰레드가 reservation 예약 필드에 접근해 ++ 을 수행했을 수도 있기 때문이다. 

 

 

 

2. 최종 구매 인원이 486명

현재 reservation 값이 n 이라 가정하면 쓰레드 1, 2 가 ++을 하기 위해 값을 가져오게 된다.

연산을 수행하고 나면

쓰레드1: n+1

쓰레드2: n+1
의 값으로 업데이트한다. 결과적으로 n+2가 아닌 n+1이 되는 것이다. 

 

 

결국 이 두 가지 상황 모두 Race Condition(경쟁 상태)가 발생한 것으로 볼 수 있다. Race Condition은 두 개 이상의 스레드가 동시에 변경 가능한 공유 자원에 접근하고 이를 변경하려고 할 때 발생한다. 

스레드 스케쥴링 알고리즘에 의해 스레드는 언제든 swap 될 수 있기 때문에 어느 시점에 어떤 순서로 스레드가 공유 자원에 접근할지 알 수 없다. 

 

따라서, 스레드 스케쥴링 알고리즘이 결정하는 스레드의 접근 시점과 순서에 따라 실행 결과가 달라지게 된다.

 

 

그렇다면 Java에서는 어떻게 Thread-Safe한 코드를 작성할 수 있을까?

 

 

 

 

 

 

 

3. 해결 방안

- Synchronized

자바의 모든 객체는 하나의 Monitor를 갖고 있다. Monitor는 상호 배제(mutual exclusion)와 협력(cooperation) 두 가지 종류의 스레드 동기화를 지원한다.

상호 배제는 Lock을 통해 다수의 스레드가 공유 자원에 대해 독립적으로 연산을 수행할 수 있도록 하는 것이며,

협력은 wait와 notify 메서드를 통해 스레드가 상호 협력하도록 하는 것이다.

 

예를 들어 버퍼로부터 데이터를 읽어오는 스레드와 버퍼에 데이터를 쓰는 스레드가 있다고 할 때, 버퍼가 비어 있다면 버퍼에 데이터를 쓰는 스레드가 데이터를 채울 때까지 wait하고, 해당 작업이 종료되면 다시 읽기를 시작하는 것이다.

 

 

 

자바의 synchrnoized 키워드는 스레드 간 동기화를 위해 사용되는 대표적인 기법이다.

synchronized 키워드는 객체 안에 존재하는 Monitor를 이용해 동기화를 수행한다. 하나의 스레드가 synchronized로 지정한 임계 영역에 들어가 있을 때 Lock이 걸리기 때문에 다른 스레드가 임계 영역에 접근할 수 없다. 이후 해당 스레드가 임계 영역의 코드를 모두 실행한 뒤 벗어나면 unlock 상태가 되어 대기하고 있던 다른 스레드가 이 임계영역에 접근해 다시 Lock을 걸어 사용할 수 있다.

 

 

앞서 진행했던 테스트를 synchronized를 사용해서 다시 진행해보았다. 

 

@Getter
public class Bus {
    private int minOccupancy = 100;
    private int reservation = 0;

    public synchronized void getBusTicket() {
        try {
            Thread.sleep(100);
            reservation++;
            if (reservation <= minOccupancy) {
                Thread.sleep(100);
                System.out.println("최소 인원인 100명을 넘어야 합니다. 현재 예약 인원: " + reservation);
            }
        } catch (InterruptedException e) {
            System.out.println("ERROR!");
        }
    }
}

 

@Test
void 다수의_사람들이_동시에_버스티켓을_구매한다() throws InterruptedException {

    int N = 500;
    CountDownLatch latch = new CountDownLatch(N);
    Bus bus = new Bus();

    for (int i = 0; i < N; i++) {
        THREAD_POOL.execute(() -> {
            bus.getBusTicket();
            latch.countDown();
        });
    }

    latch.await();

    int busReservation = bus.getReservation();
    System.out.println("======= Total reservation: " + busReservation + " =======");
    assertThat(busReservation).isEqualTo(N);
}

 

결과는 아래처럼 나온다. 

 

synchronized를 사용하면 손쉽게 Thread-Safe한 환경을 만들 수 있지만 느리다. 임계영역의 코드는 하나의 쓰레드씩만 사용할 수 있기 때문이다. 

 

 

 

 

 

- Atomic Objects

자바는 AtomicInteger, AtomicLong, AtomicBoolean 등의 atomic 클래스를 제공한다. 

atomic 클래스는 멀티 스레드 환경에서 원자성을 보장해준다. 또한, 앞에서 살펴본 synchronized와는 다르게 blocking이 아닌 non-blocking 방식으로 동작하며 핵심 동작 원리는 CAS(Compare And Swap) 알고리즘이다.

 

@Getter
public class Bus {
    private int minOccupancy = 100;
    private AtomicInteger reservation = new AtomicInteger();

    public synchronized void getBusTicket() {
        try {
            Thread.sleep(100);
            int newReservation = reservation.incrementAndGet();
            if (newReservation <= minOccupancy) {
                Thread.sleep(100);
                System.out.println("최소 인원인 100명을 넘어야 합니다. 현재 예약 인원: " + reservation);
            }
        } catch (InterruptedException e) {
            System.out.println("ERROR!");
        }
    }
}

 

 

위에서 봤던 synchronized랑 비교하기 위해 테스트 코드 수행 시간을 남겼다. 

@Test
void 다수의_사람들이_동시에_버스티켓을_구매한다() throws InterruptedException {
    long startTime = System.currentTimeMillis();


    int N = 500;
    CountDownLatch latch = new CountDownLatch(N);
    Bus bus = new Bus();

    for (int i = 0; i < N; i++) {
        THREAD_POOL.execute(() -> {
            bus.getBusTicket();
            latch.countDown();
        });
    }

    latch.await();

    int busReservation = bus.getReservation().get();
    System.out.println("======= Total reservation: " + busReservation + " =======");
    long endTime = System.currentTimeMillis();
    System.out.println("총 걸린 시간: " + (endTime - startTime) + " milliseconds.");
    assertThat(busReservation).isEqualTo(N);
}

 

 

 

원하는 결과가 나왔다. 

Atomic이 어떤 역할을 해줬길래 Thread-Safe한 코드가 됐을까??

 

자바로 작성된 한 줄의 연산인 reservation++은 변수의 값을 읽고, 연산한 뒤 저장하는 세 줄의 기계어로 쪼개질 수 있고, 멀티 스레드 환경에서 기계어 연산들이 한 번에 모두 실행될거라는 보장이 없기 때문이다.

값을 읽고 연산하는 기계어까지 실행했는데 다른 스레드로 swap 되는 상황을 생각하면 된다.

 

atomic 클래스는 원자성을 보장하기 때문에 멀티 스레드 환경에서 안전하게 사용할 수 있다.

 

무엇보다 수행 시간이 667ms로 굉장히 빨랐다. 

 

synchronized 를 사용해서 테스트를 진행하니.. 무려 64031ms 가 걸렸다. 

 

하나의 필드를 공유 자원으로 갖는다면 Atomic 을 사용하는 게 좋아보였다. 하지만 필드가 아닌 로직이 Thread-Safe해야 한다면 느리지만 synchronized를 사용해야 하지 않을까 생각이 든다.