자바의 정석 ch13. 쓰레드(thread) 下

쓰레드의 동기화

멀티쓰레드 프로세스의 경우 여러 쓰레드가 같은 프로세스 내의 자원을 공유해서 작업하기 때문에 서로의 작업에 영향을 주게 됨. 이러한 일이 발생하는 것을 방지하기 위해서 한 쓰레드가 특정 작업을 끝마치기 전까지 다른 스레드에 의해 방해받지 않도록 하는 것이 필요함. 그래서 도입된 개념이 바로 '임계 영역'과 '잠금'이다.
공유 데이터를 사용하는 코드 영역을 임계 영역으로 지정해놓고, 공유 데이터(객체)가 가지고 있는 lock을 획득한 단 하나의 쓰레드만 이 영역 내의 코드를 수행할 수 있도록 함. 그리고 해당 스레드가 임계 영역내의 모든 코드를 수행하고 벗어나서 lock을 반납해야 비로소 다른 쓰레드가 반납된 lock을 획득하여 임계 영역의 코드를 수행할 수 있게 됨.

• 쓰레드 동기화: 한 쓰레드가 진행 중인 작업을 다른 쓰레드가 간섭하지 못하도록 막는 것
• 'java.util.concurrent.locks'와 'java.util.cpncurrent.atomic'패키지를 통해서 다양한 방식으로 동기화를 구현할 수 있도록 지원

synchronized를 이용한 동기화

• 임계 영역을 설정하는데 사용
• 임계 영역은 멀티쓰레드 프로그램의 성능을 좌우하기 때문에 가능하면 메서드 전체에 락을 거는 것보다 synchronized블럭으로 임계 영역을 최소화해서 보다 효율적인 프로그램이 되도록 해야함.

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

//특정한 영역을 임계 영역으로 지정
synchronized(락을 걸고자 하는 객체의 참조변수) {
	//...
} //블락을 벗어나면 lock을 반납

wait()과 notify()

synchronized로 동기화해서 공유 데이터를 보호하는 것까진 좋지만, 특정 쓰레드가 객체의 락을 가진 상태로 오랜 시간을 보내지 않도록 하는 것도 중요함. 이를 위해 고안된 것이 wait()과 notify()이다.

• wait(): 동기화된 임계영역의 코드를 수행하다가 작업을 더이상 진행할 상황이 아니면, 일단 wait()를 호출하여 쓰레드가 락을 반납하고 기다리게 함. 그럼 다른 쓰레드가 락을 얻어 해당 객체에 대한 작업을 수행할 수 있게됨.
• notify(): 나중에 작업을 진행할 수 있는 상황이 되면 notify()를 호출해서 작업을 중단했던 쓰레드 중 임의의 쓰레드만 통지를 받고 다시 락을 얻어 작업을 진행할 수 있게됨.
• wait(), notify(), notifyAll()
  • Object에 정의되어 있음
  • 동기화 블록(synchronized블록) 내에서만 사용 가능
  • 보다 효율적인 동기화 가능

Lock과 Condition을 이용한 동기화

동기화할 수 있는 방법은 synchronized블럭 외에도 'java.util.concurrent.locks'패키지가 제공하는 lock클래스들을 이용하는 방법이 있다.

• lock클래스 종류
  • ReentrantLock: 재진입이 가능한 lock, 가장 일반적인 배타 lock. 특정 조건에서 lock을 풀고 나중에 다시 lock을 얻고 임계영역으로 들어와서 이후의 작업을 수행할 수 있음.
  • ReentrantReadWriteLock: 읽기에는 공유적이고, 쓰기에는 배타적인 lock. 읽기 lock이 걸려져있으면, 다른 쓰레드가 읽기 lock을 중복해서 걸고 읽기를 수행할 수 있음.(읽기는 내용을 변경하지 않으므로 동시에 여러 쓰레드가 있어도 문제가 되지 않음) 그러나 읽기 lock이 걸린 상태에서 쓰기 lock을 거는 것은 허용 X. 반대의 경우도 허용X
  • StampedLock: ReentrantReadWriteLock에 낙관적인 lock의 기능을 추가. 읽기 lock이 걸려있으면, 쓰기 lock을 얻기 위해 바로 풀수 있게 해줌. 즉, 무조건 읽기 lock을 걸지 않고, 쓰기와 읽기가 충돌할 때만 쓰기가 끝난 후에 읽기 lock을 거는 것이다.

ReentrantLock의 생성자

ReentrantLock()
ReentrantLock(boolean fair)

생성자의 매개변수를 true로 주면, lock이 풀렸을 때 가장 오래 기다린 쓰레드가 lock을 획들 수 있게 함. 하지만 어떤 쓰레드가 가장 오래 기다렸는지 확인하는 과정을 거칠 수 밖에 없으므로 성능이 떨어짐.

void lock() //lock을 잠근다.
void unlock() //lock을 해지한다.
boolean isLocked() //lock이 잠겼는지 확인한다.

ReentrantLock(lock클래스)은 수동으로 lock을 잠그고 해제해야함.(synchronized블럭은 자동적으로 lock의 잠금과 해제가 관리됨)
ex.

lock.lock(); //ReentrantLock lock = new ReentrantLock();
try{
	//임계영역
} finally {
	lock.unlock();
}

이외에도 tryLock()이라는 메서드가 있는데, 이 메서드는 다른 쓰레드에 의해 lock이 걸려 있으면 lock을 얻으려고 기다리지 않고, 지정된 시간만큼만 기다림. lock을 얻으면 true를 반환하고, 얻지 못하면 false를 반환

boolean tryLock()
boolean tryLock(long timeout, TimeUnit unit) throws InterruptedExeption

ReentrantLock과 Condition

wait() & notify()로 쓰레드의 종류를 구분하지 않고, 공유 객체의 waiting pool에 같이 몰아넣는 대신, 각각의 쓰레드를 위한 Condition을 만들어 각각의 waiting pool에 따로 기다리도록 함.

private ReentrantLook lock = new ReetrantLock(); //lock 생성

private Condition forCook = lock.newCondition(); //lock으로 condition을 생성
private Condition forCust = lock.newCondition();

wait()과 notify() 대신, Condition의 await()와 signal()을 사용
기아 현상이나 경쟁 상태가 개선될 수 있음.

volatile

멀티 코어 프로세서에서는 코어마다 별도의 캐시를 가지고 있기 때문에 문제가 발생할 가능성이 있다. 코어는 메모리에서 읽어온 값을 캐시에 저장하고 캐시에서 값을 읽어서 작업한다. 다시 같은 값을 읽어올 때는 먼저 캐시에 있는지 확인하고 없을 때만 메모리에서 읽어온다. 그러다보니 도중에 메모리에 저장된 변수의 값이 변경되었는데도 캐시에 저장된 값이 갱신되지 않아서 메모리에 저장된 값이 다른 경우가 발생한다.

volatile boolean suspended = false;
volatile boolean stopped = false;

그럴 때 변수 앞에 volatile을 붙이면, 코어가 변수의 값을 읽어올 때 캐시가 아닌 메모리에서 읽어오기 때문에 캐시와 메모리간의 값의 불일치가 해결된다. 변수에 volatile을 붙이는 대신에 synchronized블럭을 사용해도 같은 효과를 얻을 수 있다.

public synchronized void stop() {
	stopped = true;
}

volatile로 long과 double을 원자화(작업을 더이상 나눌 수 없게 한다.)

JVM은 데이터를 4byte단위로 처리하기 때문에, int와 int보다 작은 타입들은 한번에 읽거나 쓰는 것이 가능하다. 하지만 크기가 8 byte인 long과 double타입의 변수는 하나의 명령어로 값을 읽거나 쓸 수 없기 때문에, 변수의 값을 읽는 과정에 다른 쓰레드가 끼어들 여지가 있다. 이를 방지하기 위해 변수를 선언할 때 volatile을 붙일 수 있다.

volatile long sharedVal; //long 타입의 변수를 원자화
volatile double sharedVal; //double 타입의 변수를 원자화

다만 주의할 것은 volatile은 변수의 읽거나 쓰기를 원자화할 뿐, 동기화하는 것은 아니다. 따라서 동기화가 필요할 때 synchronized블럭 대신 volatile을 쓸 수 없다.

fork & join 프레임웍

이 프레임웍은 하나의 작업을 작은 단위로 나눠서 여러 쓰레드가 동시에 처리하는 것을 쉽게 만들어준다. 먼저 수행할 작업에 따라 RecursiveAction과 RecursiveTask 두 클래스 중에서 하나를 상속받아 구현해야한다.

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

두 클래스 모두 compute()라는 추상메서드를 가지고 있는데, 우리는 상속을 통해 이 추상 메서드를 구현하기만 하면 된다.

public abstract class RecursiveAction extends ForkJoinTask<void> {
		...
    protected abstract void compute(); //상속을 통해 이 메서드를 구현
    	...
}

public abstract class RecursiveTask extends ForkJoinTask<void> {
		...
    protected abstract V compute(); //상속을 통해 이 메서드를 구현
    	...
}

ex. 1~n까지의 합을 계산해서 결과를 돌려주는 작업 구현

class SumTask extends RecursiveTask<Long> { //RecursiveTask를 상속받음
	long from;
    long to;
    
    SumTask(long from, long to) {
    	this.from = from;
        this.to = to;
    }
    
    public Long compute() {
    	//처리할 작업을 수행하기 위한 문장 넣기
    }
}

compute()가 아닌 invoke()로 시작함

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

Long result = pool.invoke(task); //invoke()를 호출해서 작업 시작

ForkJoinPool(쓰레드 풀)은 지정된 수의 쓰레드를 생성해서 미리 만들어 놓고 반복해서 재사용할 수 있게 한다. 그리고 쓰레들르 반복해서 생성하지 않아도 된다는 장점과 너무 많은 쓰레드가 생성되어 성능이 저하되는 것을 막아준다는 장점이 있다.

compute()의 구현

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

다른 쓰레드의 작업 훔쳐오기

• 작업 훔쳐오기: fork()가 호출되어 작업 큐에 추가된 작업 역시, compute()에 의해 더이상 나눌 수 없을 때까지 반복해서 나뉘고, 자신의 작업 큐가 비어있는 쓰레드는 다른 쓰레드의 작업 큐에서 작업을 가져와서 수행함.
  • 쓰레드 풀에 의해 자동적으로 이루어짐

fork()와 join()

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