자바에서의 Thread Safety

[masiljangajji]

자바에서의 Thread Safety

Parallel Programing 이란 여러개의 작업을 동시에 처리하는 프로그래밍 방식입니다.
실제로 우리가 사용하는 대부분의 서비스들은 Parallel 하게 동작합니다.

하지만 여러 작업들이 동시에 진행되기 때문에 프로그램은 복잡해지며

다음과 같은 문제들을 야기할 수 있습니다.

Race Condition 으로 인한 결과값 변동

Deadlock , Concurrency Bugs 같은 버그

Synchronized Overhead 같은 성능 문제

이 글에서는 여러가지 문제들중 Race Condition에 대한
Thread Safety를 집중적으로 다룰 것 입니다.

1. Thread Safety의 필요성

Thread Safety는 왜 필요할까요?

또 똑바로 지켜지지 않는다면 어떤 문제가 발생할 까요?

다음의 코드를 통해서 확인해보겠습니다.

public class Main {


    public static void main(String[] args) {

        Number number = new Number();

        Counter counter1 = new Counter(number, true);
        Counter counter2 = new Counter(number, false);

        counter1.start();
        counter2.start();

        while (counter1.isAlive() || counter2.isAlive()) {
        } // Thread종료 될떄 까지 기다리기


        System.out.println(counter1.getState() + " " + counter2.getState());
        System.out.println(number.count);


    }
}

class Number {

    int count;

    public Number() {
        this.count = 0;
    }

    public void increase() {
        this.count++;
    }

    public void decrease() {
        this.count--;
    }

}

class Counter extends Thread {


    private final Number number;

    private boolean flag;

    public Counter(Number number, boolean flag) {
        this.number = number;
        this.flag = flag;
    }


    @Override
    public void run() {

        for (int i = 0; i < 10000000; i++) {

            if (flag) {
                number.increase();
            } else {
                number.decrease();
            }
        }

    }
}

Counter를 통해서 Number 객체의 count를 증감시키는 코드입니다.

똑같은 Number 객체를 갖고있기 때문에 Number는 공유자원이 됩니다.

똑같은 횟수를 증감하기 떄문에 출력은 0이 되야 정상이지만

0이 아닌값이 출력됩니다.

왜 이런 일이 생기는 것일까요?

Multi Thread환경에서는 특정 자원(Shared Resources)을 공유하게 됩니다.

위의 코드에서는 Number가 공유 자원에 속합니다.
count 값이 5일때 동시에 increase , decrease 연산이 일어나는 경우를 생각해 보겠습니다.

1. incrase,decrease 둘 다 5의 count 값 가짐

2. increase 동작 전에 decrease 동작 완료 count 값 4 변경

3. increase는 이미 count 값 5를 가지고 있음으로 6으로 증가

4. 각각 한번의 increase , decrease 가 일어 났지만 결과 값은 5가 아닌 6

따라서 공유자원에 대한 접근을 제한하지 않는다면 , 즉 Thread Safety하게 만들지 않으면
예상치 못한 결과값이 만들어 질 수 있습니다.

또 공유 자원에 대한 접근이 이뤄지는 영역을 Critical Section(임계영역) 이라 부릅니다.

2. Thread Safety를 위한 방법

자바에서는 Thread Safety를 지키기 위해

크게 2가지 방법을 사용합니다.

1. Mutex

• Synchronization
• Lock Interface

2. Semaphore

Mutex , Semaphore 무슨 차이가 존재할까요?

Mutex : 공유자원에 하나의 thread만 접근하게 함

Semaphore : 공유자원에 접근 가능한 thread의 개수를 제한 함

다음과 같은 제약조건을 갖는 서점 프로그램이 있다고 했을때

1. 서점에 출고하는 생산자 수는 2명으로 제한한다.
2. 서점에서 책 구매는 동시에 1명만 가능하다.

서점에 출고하는 생산자 수를 2명으로 두는 것은
출고 가능한 스래드의 개수에 제한이기 떄문에 Semaphore를

책 구매는 책이라는 공유자원에 대한 접근을 1명만 가능하게 하는 것으로
Mutex를 이용하면 됩니다.

더 자세한 내용은 아래에서 설명하도록 하겠습니다.

3. Mutual Exclusion (Mutex)

Mutual Exclusion 은 둘 이상의 thread가 동시에 Critical Section에 접근하는 것을 막음을 의미합니다.

즉 공유자원에 대해서 한번에 하나의 thread만 접근하도록 만드는 것입니다.

대표적으로 synchronized keyword , Lock Interface 가 존재합니다.

synchronized

 public synchronized void increase() { // number에 대한 모니터락 획득
        this.count++; 
    }

    public synchronized void decrease() { //number에 대한 모니터락 획득
        this.count--;
    }

위에서 사용된 코드지만 메서드에 synchronized keyword가 붙었습니다.

synchronized가 붙음으로써 메서드와 이 메서드가 사용하는 인스턴스에 대해
하나의 thread만 접근 가능하도록 제한합니다.

이를 모니터 락을 획득했다고 합니다.

이렇게 메서드를 수정하고 난 후에는
count값이 정상적으로 0 이 나오는 것을 볼 수 있습니다.

메서드에 적용하는 방법 뿐 아니라 , 블록에도 적용 가능합니다.

@Override
    public void run() {

        for (int i = 0; i < 10000000; i++) {

            if (flag) {
                synchronized (number) { // number 모니터 락 획득
                    number.increase();
                }
            } else {
                synchronized (number) { // number 모니터 락 획득
                    number.decrease();
                }
            }
        }

    }

이렇게 사용해도 number 객체에 대한 모니터 락을 획득해

접근을 제어하는것이 됩니다.

Lock

Lock Interface는 java.util.concurrent.locks.Lock 에 존재하며

동기화 문제를 해결하기 위한 메커니즘 중 하나입니다. 이를 사용하여

코드 블록을 동기화하고 잠금을 걸 수 있습니다.

주로 Lock 인터페이스를 구현한 클래스 중 하나인 ReentrantLock이 사용됩니다.

Lock lock = new ReentrantLock();

public void increase() {
        lock.lock(); // lock 획득 
        this.count++; // 보호받음
        lock.unlock(); // lock 해제
    }

    public void decrease() {
        lock.lock(); // lock 획득
        this.count--; // 보호받음
        lock.unlock(); // lock 해제
    }

synchronized를 쓰지 않아도 정상적인 결과값 0을 얻습니다.
또한 synchronized와 같은 기능을 합니다.

여기서 사용된 lock() , unlock() 메서드 외에도
tryLock() , lockInterruptibly() , Condition을 통해서

synchronized 보다 더 유연하게 사용 가능합니다.

4. Semaphore

앞서 말했듯이 Semaphore는 공유자원에 접근 가능한 thread의 수를 제한 합니다.

java.util.concurrent.Semaphore 에 존재하며
acquire() , release() 메서드를 통해 사용됩니다.

다음과 같이 사용될 수 있습니다.

import java.util.concurrent.Semaphore;

public class Main {


    public static void main(String[] args) {

        Number number = new Number();

        Counter[] counters = new Counter[5];

        for (int i = 0; i < 5; i++) {
            counters[i] = new Counter(number);
            counters[i].start(); // 5개의 스래드가 돌아가기 때문에 원래는 5씩 증가해야 한다. 
        }
        

        while (counters[0].isAlive() || counters[1].isAlive() || counters[2].isAlive() || counters[3].isAlive() ||
                counters[4].isAlive()) {

            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                System.out.println(e.getMessage());
            }
            System.out.println(number.getCount());
        }

        System.out.println("종료");
    }


}

class Number {


    private Semaphore semaphore = new Semaphore(3); // 스래드 3개까지 동시 엑세스 허용

    public int getCount() {
        return count;
    }

    private int count;

    public Number() {
        this.count = 0;
    }

    public void increase() {
        try {
            semaphore.acquire(); // 스래드 엑세스 허용
            System.out.println("증가 시작");
            this.count++; 
            Thread.sleep(500);

        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        }
        semaphore.release(); // 스래드 엑세스 해제

    }

}

class Counter extends Thread {


    private final Number number;


    public Counter(Number number) {
        this.number = number;
    }


    @Override
    public void run() {

        for (int i = 0; i < 10000; i++) {
            number.increase();
        }

    }
}

원래라면 5개의 스래드가 돌아가고 있기 때문에

5씩 count가 증가해야 하지만 Semaphore를 이용해 3개의 스래드만 허용했음으로
3씩 증가하는 것을 볼 수 있습니다.

이렇게 Thread Safety를 위한 동기화 기법을 알아봤습니다.

마지막으로 Mutex와 Semaphore를 동시에 사용하는

서점 예시를 통해 동기화 기법을 정리하고 마무리 하겠습니다.

서점 조건

1. 서점에는 5개의 책이 존재한다.

2. 생산자는 5개의 책중 하나를 랜덤하게 납품한다.

3. 책은 이름과 번호를 가진다.

4. 똑같은 종류의 책 재고는 최대 10권까지 가능하며 , 그 이상 부터는 납품을 미룬다.

5. 재고가 없는 경우에는 출고(판매)하기 위해 납품을 기다린다.

6. 생산자는 최대 2명까지 허용한다.

7. 소비자는 책의 번호별로 구매가능하다.
   즉, 여러 사람이 각기 다른 번호의 책을 구매할 경우 동시 구매가 가능하다.

필요한 Class

Main , Book , Library , Buyer(소비자) , Seller(생산자)

Main Class

public class Main {

    public static void main(String[] args) {

        Library library = new Library();


        Seller[] sellers = new Seller[5]; // 5명의 생산자

        Buyer[] buyers = new Buyer[10]; // 10명의 소비자

        for (int i = 0; i < sellers.length; i++) {
            sellers[i] = new Seller(library);
            sellers[i].start();
        }

        for (int i = 0; i < buyers.length; i++) {
            buyers[i] = new Buyer(library);
            buyers[i].start();
        }

    }
}

Book Class

public class Book {
    private final String name;
    private final int index;

    public int getIndex() {
        return index;
    }

    public Book(String name, int index) {
        this.name = name;
        this.index = index;
    }

    @Override
    public String toString() {
        return name;
    }

    public String getName() {
        return name;
    }
}

Library Class

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Semaphore;

public class Library {

    private final List[] books = new ArrayList[5];

    private final Semaphore semaphore = new Semaphore(2); // 스래드 2개까지 접근 가능하게 제한하겠다.

    public synchronized void printList() {

        for (int i = 0; i < books.length; i++) {
            if (!books[i].isEmpty()) {
                System.out.println(books[i].get(0) + " 현재 재고 : " + books[i].size());
            }
        }

    }

    public Library() {
        for (int i = 0; i < 5; i++) {
            List<Book> list = new ArrayList<>();
            books[i] = list;
        }
    }

    public void buyBook(Book book) {

        int index = book.getIndex();

        try {
            semaphore.acquire(); // 납품할 생산자는 최대 2명으로 제한하여 접근하게 함 
            synchronized (books[index]) { // books[index]에 대한 모니터 락 획득.

                while (books[index].size() == 10) { // 책이 이미 10개 이상 존재한다면 더 이상 받을 수 없음

                    try {
                        books[index].wait(); // books[index]가 비워질떄 까지 기다려야 하기 때문에
                        // books[index]에 대한 접근을 허용하기 위해 모니터 락을 반환하고 , 스래드는 일시정지 시킴
                    } catch (InterruptedException e) {
                        System.out.println(e.getMessage());
                    }
                }


                try {
                    Thread.sleep(1000); // 입고 하는데 1초 걸림.
                } catch (InterruptedException e) {
                    System.out.println(e.getMessage());
                }

                System.out.println("책 입고 완료 : " + book);
                books[index].add(book); // 책을 추가
                books[index].notifyAll(); // books[index]에 접근을 기다리는 모든 스래드를 꺠움
                printList();
            }
            semaphore.release(); // 생산자 접근 해제
        } catch (InterruptedException e) {
            System.out.println(e.getMessage());
        }


    }

    public void sellBook(int index) {

        synchronized (books[index]) { // books[index]에 대한 모니터 락 획득.

            while (books[index].isEmpty()) { // 책이 존재하지 않는다면
                try {
                    books[index].wait(); // books[index]가 채워질때 까지 기다려야 하기 떄문에
                    // books[index]에 대한 접근을 허용하기 위해 모니터 락을 반환하고 , 스래드는 일시정지 시킴

                    //책이 입고되면 notifyAll()로 인해 정지해있던 스래드가 다시 모니터 락을 얻기 위해 경쟁함
                } catch (InterruptedException e) {
                    System.out.println(e.getMessage());
                }
            }

            try {
                Thread.sleep(100); // 책 출고시 0.1초 걸림.
            } catch (InterruptedException e) {
                System.out.println(e.getMessage());
            }

            System.out.println("책 출고 완료 : " + books[index].get(0));
            books[index].remove(0); // 책 판매되서 재고 1개 삭제 됨.
            books[index].notifyAll(); // // books[index]에 접근을 기다리는 모든 스래드를 꺠움
        }

    }

}

Buyer Class

import java.util.concurrent.ThreadLocalRandom;

public class Buyer extends Thread {

    private final Library library;

    Buyer(Library library) {
        this.library = library;
    }

    @Override
    public void run() {

        while (true) {

            // multi thread환경에서 난수 생성
            int index = ThreadLocalRandom.current().nextInt(0, 5);

            library.sellBook(index);

        }

    }
}

Seller Class

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadLocalRandom;

public class Seller extends Thread {

    private final Library library;

    private final List<Book> list = new ArrayList<>();

    Seller(Library library) {
        this.library = library;

        list.add(new Book("NHN", 0));
        list.add(new Book("Academy", 1));
        list.add(new Book("Back-End", 2));
        list.add(new Book("4기", 3));
        list.add(new Book("파이팅", 4));
    }

    @Override
    public void run() {

        while (true) {

            int index = ThreadLocalRandom.current().nextInt(0, 5);
            // multi thread환경에서 난수 생성

            library.buyBook(new Book(list.get(index).getName(), list.get(index).getIndex())); // 책 입고
        }

    }
}

실행 결과