[Java] Thread(쓰레드) - 5

1. 쓰레드 동기화

싱글 쓰레드 프로세스의 경우 프로세스 내에서 단 하나의 쓰레드만 작업하기 때문에 프로세스의 자원을 가지고 작업하는데 별 문제가 없지만, 멀티 쓰레드 프로세스의 경우 여러 쓰레드가 같은 프로세스 내의 자원을 공유해서 작업하기 때문에 서로의 작업에 영향을 주게된다. 만일 쓰레드A가 작업하던 도중에 다른 쓰레드B에게 제어권이 넘어갔을 때 , 쓰레드 A가 작업하던 공유 데이터를 쓰레드 B가 임의로 변경하였다면, 다시 쓰레드 A가 제어권을 받아서 나머지 작업을 마쳤을 때 원래 의도하던 것과는 다른 결과를 얻을 수 있다.

 

이러한 일이 발생하는 것을 막기 위해서 한 쓰레드가 특정 작업을 끝마치기 전까지 다른 쓰레드에 의해 방해받지 않도록 하는 것이 필요하다. 그래서 도입된 개념이 바로 '임계 영역'과 '잠금'이다.

 

공유 데이터를 사용하는 코드 영역을 임계 영역으로 지정해놓고 공유 데이터(객체)가 가지고 있는 lock을 획득한 단 하나의 쓰레드만 이 영역 내의 코드를 수행할 수 있게 한다. 그리고 해당 쓰레드가 임계 영역 내의 모든 코드를 수행하고 벗어나서 lock을 반납 해야만 다른 쓰레드가 반납된 lock을 획득하여 임계 영역 코드를 수행할 수 있게 된다.

 

이처럼 한 쓰레드가 진행중인 작업을 다른 쓰레드가 간섭하지 못하도록 막는 것을 '쓰레드의 동기화'라고한다. 자바에서는 synchronized 블럭을 이용해서 쓰레드의 동기화를 지원했지만, JDK1.5부터는 java.util.concurrent.locks와 java.util.concurrent.atomic 패키지를 통해서 다양한 방식으로 동기화를 구현할 수 있다.

 

2. synchronized를 이용한 동기화

먼저 가장 간단한 동기화 방법인 synchronized 키워드를 이용한 동기화에 대해서 알아본다. 

1. 메서드 전체를 임계 영역으로 지정
public synchronized void calcSum() {
    // 임계 영역
}

2. 특정한 영역을 임계 영역으로 지정
synchronized(객체의 참조변수) {
    // 임계 영역
}

첫 번째 방법은 메서드 앞에 synchronized 를 붙이는 것인데, synchronized를 붙이면 메서드 전체가 임계 영역으로 설정된다. 쓰레드는 synchronized 메서드가 호출된 시점부터 해당 메서드가 포함된 객체의 lock을 얻어 작업을 수행하다가 메서드가 종료되면 lock을 반환한다.

 

두 번째 방법은 메서드 내의 코드 일부를 블럭 {} 으로 감싸고 블럭 앞에 'synchronized(참조 변수)'를 붙이는 것인데, 이때 참조변수는 락을 걸고자하는 객체를 참조하는 것이어야 한다. 이 블럭을 synchronized 블럭이라고 부르며, 이 블럭의 영역 안으로 들어가면서부터 쓰레드는 지정된 객체의 lock을 얻게 되고, 이 블럭을 벗어나면 lock을 반납한다.

 

두 방법 모두 lock을 하나씩 가지고 있으며, 해당 객체의 lock을 가지고 있는 쓰레드만 임계 영역의 코드를 수행할 수 있다. 그리고 다른 쓰레드들은 lock을 얻을 때까지 기다리게 된다.

 

임계 영역은 멀티쓰레드 프로그램의 성능을 좌우하기 때문에 가능하면 메서드 전체에 락을 거는 것보다 synchronized 블럭으로 임계 영역을 최소화해서 보다 효율적인 프로그램이 되도록 노력해야한다. withdraw 메소드에 synchronized를 붙여주어서 한 쓰레드에 의해서 먼저 withdraw()를 호출하면 다른 쓰레드는 withdraw()를 호출하더라도 대기상태에 머물게 된다.

package thread;

public class ThreadEx21 {

    public static void main(String[] args) {
        Runnable r = new RunnableEx21();
        new Thread(r).start(); // ThreadGroup에 의해 참조되므로 gc 대상이 아니다.
        new Thread(r).start(); // ThreadGroup에 의해 참조되므로 gc 대상이 아니다.
    }

    static class Account {
        private int balance = 1000;
        public int getBalance() {
            return balance;
        }

        public synchronized void withdraw(int money) {
            if(balance >= money) {
                try{
                    Thread.sleep(1000);
                    balance -= money;
                } catch (InterruptedException e) {

                }
            }
        }
    }

    static class RunnableEx21 implements Runnable {
        Account acc = new Account();

        @Override
        public void run() {
            while(acc.getBalance() > 0) {
                int money = (int) (Math.random() * 3 + 1) * 100;
                acc.withdraw(money);
                System.out.println("balance:"+acc.getBalance());
            }
        }
    }

}

3. wait과 notify()

synchronized로 동기화해서 공유 데이터를 보호하는 것 까지는 좋은데, 특정 쓰레드가 객체의 락을 가진 상태로 오랜 시간을 보내지 않도록 하는 것도 중요하다. 이런 상황을 개선하기 위해서 고안된 것이 바로 wait()과 notify()이다. 동기화된 임계 영역 코드를 수행하다가 작업을 더 이상 진행할 상황이 아니면, 일단 wait()을 호출하여 쓰레드가 락을 반납하고 기다리게한다. 그러면 다른 쓰레드가 락을 얻어 해당 객체에 대한 작업을 수행할 수 있게 된다. 나중에 작업을 진행할 수 있는 상황이 되면 notify()를 호출해서, 작업을 중단했던 쓰레드가 다시 락을 얻어 작업을 진행할 수 있게 한다.

 

꼭 오래 기다린 쓰레드가 락을 얻는다는 보장은 없다. wait()이 호출되면, 실행 중이던 쓰레드는 해당 객체의 대기실(waiting pool)에서 통지를 기다린다. notify()가 호출되면 해당 객체의 대기실에 있던 모든 쓰레드 중에서 임의의 쓰레드만 통지를 받는다. notifyAll()은 기다리고 있는 모든 쓰레드에게 통보를 하지만, 그래도 Lock을 얻을 수 있는 것은 하나의 쓰레드일 뿐이고, 나머지 쓰레드는 통보를 받긴 했지만, lock을 얻지 못하면 다시 lock을 기다리는 신세가 된다.

 

wait()과 notify(), notifyAll()는 Object 클래스에 정의되어 있다. 또한 동기화 블록(synchronized 블록)내에서만 사용할 수 있다. 이를 통해서 보다 효율적인 동기화를 가능하게한다.

 

4. Lock과 Condition을 이용한 동기화

synchronized 블럭으로 동기화를 하면 자동으로 lock이 잠기고 풀리기 때문에 편리하지만, 같은 메서드 내에서만 lock을 걸 수 있다는 제약이 불편하기도 하다.

 

• ReetranLock : 재진입이 가능한 lock. 가장 일반적인 배타 lock. lock이 풀렸을 때 가장 오래 기다린 쓰레드가 lock을 획득한다.
• ReetrantReadWriteLock : 읽기에는 공유적이고, 쓰기에는 배타적인 lock. 읽기 lock이 걸려있으면, 다른 쓰레드가 읽기 Lock을 중복해서 걸고 읽기를 수행할 수 있다. 읽기 Lock이 걸린 상태에서 쓰기 lock을 거는 것은 허용되지 않는다. 대신에 동시에 여러 쓰레드가 중복해서 lock을 걸고 읽기를 수행할 수 있다. 쓰기 할 때는 쓰기 Lock을 거는 것일뿐 lock을 거는 방법은 똑같다.
• StampedLock : ReentrantReadWriteLock에 낙관적인 Lock의 기능을 추가. 무조건 읽기 lock을 걸지 않고, 쓰기와 읽기가 충돌할 때만 쓰기가 끝난 후에 읽기 Lock을 거는 것이다. 낙관적 읽기 lock은 쓰기 lock에 의해 바로 풀린다.

 

wait(), notify()예제에서는 요리사 쓰레드와 손님 쓰레드를 구분해서 통지하지 못한다는 단점이 있었다. Condition은 요리사 쓰레드와 손님 쓰레드를 위한 waiting pool을 따로 만들어서 해결한다. 

그 다음에는 wait(), notify() 대신에 await(), signal()을 사용한다.

이를 통해서 요리사 쓰레드가 통지를 받아야하는 상황에서ㅓ 손님 쓰레드가 통지를 받는 경우를 없앨 수 있다. '기아 현상'이나 '경쟁 상태'가개선이된다. 하지만 여전히 특정 쓰레드를 선택할 수 없기 때문에 같은 종류의 쓰레드간의 '기아 현상'이나 '경쟁 상태'가 발생할 가능성은 남아있다.

5. volatile

멀티코어 프로세서는 코어마다 별도의 캐시를 가지고 있다. 코어는 메모리에서 읽어온 값을 캐시에 저장하고 캐시에서 값을 읽어서 작업한다. 다시 같은 값을 읽어올 때는 먼저 캐시에 있는지 확인하고 없을 때만 메모리에서 읽어온다. 그러다보니 도중에 메모리에 저장된 변수의 값이 변경 되어도 캐시의 값이 갱신되지 않아서 메모리에 저장된 값과 다른 경우가 발생한다. 그래서 변수 stopped의 값이 바뀌었는데도 쓰레드가 멈추지 않고 계속 실행되는 것이다.

 

이때 volatile을 변수 앞에 붙이면 코어가 변수의 값을 읽어올 때 캐시가 아닌 메모리에서 읽어오기 때문에 캐시와 메모리간의 값의 불일치가 해결된다. 변수에 volatile을 붙이는 대신에 synchronized 블럭을 사용해도 같은 효과를 얻을 수 있다. 쓰레드가 synchronized 블럭으로 들어갈 때와 나올 때 캐시와 메모리간의 동기화가 이루어지기 때문이다.

 

volatile은 해당 변수에 대한 읽거나 쓰기가 원자화할 뿐 동기화하는 것은 아니라는 점에 주의하자.

 

6. fork & join 프레임웍

10년전까지만 해도 CPU의 속도는 매년 2배씩 향상되었다. 그러나 이제 그 한계에 도달하여 속도 보다는 코어의 개수를 늘려서 CPU의 성능을 향상시키는 방향으로 발전하고 있다. 이러한 하드웨어의 변화에 발맞춰 프로그래밍도 멀티 코어를 잘 활용할 수 있는 멀티 쓰레드 프로그래밍이 더  중요해지고 있다.

 

JDK 1.7 부터는 'fork & join 프레임웍'이 추가되었고, 이 프레임웍은 하나의 작업을 작은 단위로 나눠서 여러 쓰레드가 동시에 처리하는 것을 쉽게 만들어 준다.

 

• RecursiveAction : 반환값이 없는 작업을 구현할 때 사용
• RecursiveTask : 반환값이 있는 작업을 구현할 때 사용

두 클래스 모두 compute() 라는 추상 메서드를 가지는데, 상속을 통해 이 추상 메서드를 구현하기만 하면 된다. 클래스를 구현하고 그 다음에는 쓰레드풀과 수행할 작업을 생성하고, invoke()로 작업을 시작한다. compute()메서드가 아닌 invoke로 쓰레드 작업을 시작하는 것을 주의하자.

ForkJoinPool pool = new ForkJoinPool(); // 쓰레드 풀을 생성
SumTask task = nwe SumTask(from, to) // 수행할 작업을 생성
Long result = pool.invoke(task);

ForkJoinPool은 fork & join 프레임웍에서 제공하는 쓰레드 풀로 지정된 수의 쓰레드를 생성해서 미리 만들어 놓고 반복해서 재사용할 수 있다. 그래서 쓰레드를 반복해서 생성하지 않아도 된다는 장점과 너무 많은 쓰레드가 생성되어 성능이 저하되는 것을 막아준다는 장점이 있다. 쓰레드 풀은 쓰레드가 수행해야하는 작업이 담긴 큐를 제공하며, 각 쓰레드는 자신의 작업 큐에 담긴 작업을 순서대로 처리한다.

 

compute()를 구현할 때는 수행할 작업 외에도 작업을 어떻게 나눌 것인가에 대해서 알려줘야한다.

public Long compute() {

    long size = to -from +1;
    if(size <= 5) // 더할 숫자가 5개 이하면
        return sum(); // 숫자의 합을 반환. sum()은 from부터 to까지의 수를 더해서 반환
    // 범위를 반으로 나눠서 두개의 작업을 생성
    long half = (from+to)/2;
    SumTask leftSum = new SumTask(from, half);
    SumTask rightSum = new SumTask(half+1, to);
    leftSum.fork(); // 작업(leftSum)을 작업 큐에 넣는다.
    return rightSum.compute() + leftSum.join();

}

실제 수행할 작업은 sum() 뿐이고 나머지는 수행할 작업의 범위를 반으로 나눠서 새로운 작업을 생성해서 실행시키기 위한 것이다. 재귀 호출을 이용했다. join 메소드로 호출 결과를 기다린다.

 

• fork() : 해당 작업을 쓰레드 풀의 작업 큐에 넣는다. 비동기 메서드
• join() : 해당 작업의 수행이 끝날 때까지 기다렸다가, 수행이 끝나면 그 결과를 반환한다. 동기 메서드

 

멀티 쓰레드로 처리를 하는 것이 항상 빠르지 않기 때문에 반드시 테스트해보고 이득이 있을 때만, 멀티쓰레드로 처리해야한다.

 

Reference

http://www.yes24.com/Product/Goods/24259565

Java의 정석 - YES24