[Java 멀티스레드] 동시성 문제 그리고 ReentrantLock, Condition, BlockingQueue

자바에서 멀티 스레드의 등장과 동시에 배워야만 하는 것이 있다.
바로 동시성 문제.
동시성 문제는 자바의 멀티 스레드에 대한 개념을 선행해야 한다.

• 자바 스레드, 멀티 스레드

동시성 문제

동시성 문제란, 여러 스레드가 하나의 자원에 접근하면서 발생하는 문제를 말한다.
여기서 자원은 지역 변수가 아닌 인스턴스 필드를 말한다.
(지역 변수는 스레드 간에 절대 공유되지 않는다.)
예시로 두 명의 소비자가 동시에 물건을 구매하고, 재고량을 감소시키는 상황을 들어보자.

class Item {

	int quantity;
    
    public Item(int quantity) {
    	this.quantity = quantity;
    }
    
    public void consume() {
    	quantity -= 1;
    }
}

공유 자원

이 때 여러 스레드가 접근하는 자원을 공유 자원이라고 말한다.
말 그대로 여러 스레드가 공유하는 자원이라는 의미이다.

int quantity;

위 예시에서는 재고량이 바로 공유 자원이 되겠다.

임계 영역

동시에 접근하면서 문제가 발생하는 부분을 의미한다.

예두 명의 소비자가 동시에 물건을 구매할 때,
재고량을 100에서 동시에 1을 감소시킨다면 98이 되어야 맞지만
동시에 접근했기 때문에 100이라는 숫자에서 서로 -1을 카운트해 99라는 결과가 나온다.

public void consume() {
    	quantity -= 1;
}

여기서 임계 영역은 재고량을 1을 감소시키는 부분이다.

synchronized

동시에 접근함으로써 생기는 문제를 막기 위해서는
동시에 접근했을 때 하나씩 처리될 수 있도록 구분하면 되겠다.

이 문제를 해결하기 위해서 자바는 synchronized 키워드를 사용한다.
synchronized 키워드는 메서드 레벨 또는 코드 블럭으로 선언할 수 있는데,
synchronized 키워드를 사용한 메서드 또는 코드 블럭은 스레드가 동시에 접근해도 하나씩 처리된다.

public synchronized void consume() {
	quantity -= 1;
}

public void consume() {
	synchronized (this) {
    	quantity -= 1;
    }
}

한번에 하나씩 수행되기 때문에 반드시 필요한 부분에만 선언해야 한다.
불필요한 부분까지 임계 영역으로 설정하는 경우,
미리 처리할 수 있는 부분도 순차적으로 처리되어 성능 저하로 이어진다.
임계 영역은 반드시 동시성 문제가 발생하는 부분만 설정해야 한다.

락(lock)

그렇다면 자바는 synchronized 키워드만으로 어떻게 동시에 접근하는 스레드를 구분할 수 있을까?
그 비밀은 바로 락(lock)이다.

임계 영역에 스레드가 접근하는 경우 공유 자원을 보유한 인스턴스의 락을 가져다 사용한다.
락은 임계 영역에 입장할 수 있는 열쇠와 같은 역할을 한다.
열쇠를 가진 스레드만 임계 영역에 입장할 수 있다.
임계 영역을 마치면 자동으로 락을 반납하고 다른 스레드가 접근해서 수행할 수 있다.

public void consume() {
	synchronized (this) {  // ** 락을 획득할 인스턴스 설정 **
    	quantity -= 1;
    }
}

synchronized 코드 블럭에서 작성한 this 부분이
락을 획득할 인스턴스를 지정하는 부분이다.

락(lock) 덕분에 동시에 공유자원에 접근해도 한개씩만 실행된다.

synchronized 문제점

synchronized의 락 기능은 간편하면서도 실용적인 것 같지만, 큰 문제가 있다.

무한 대기

public void consume() {
	synchronized (this) {
    	quantity -= 1;
        // ** 오류가 발생한다면? **
    }
}

락을 획득하지 못하고 대기하는 스레드는 BLOCKED 상태로 대기한다.
다른 스레드가 반납한 락을 획득해야만 BLOCKED 상태가 해제되어 임계 영역에 접근할 수 있다.

하지만 임계 영역을 마치기 전에 오류가 발생하여 락을 반납하지 못하게 된다면?
BLOCKED 스레드의 상태를 변경하는 방법은 락을 부여하는 방법밖에 없다.
오류가 발생하여 락을 반납하지 못하면 다른 스레드는 무한 대기 상태에 빠져버린다.

공정성 문제

락 획득을 대기하는 스레드들이 락을 얻는 방식은 무작위 방식이다.
100명의 소비자가 물건을 구매하기 위해 대기하고 있는 상황에서
구매는 순서대로 이루어지는 것이 아닌 무작위 방식으로 이루어진다.

물건이 너무 인기가 많아서 구매자가 계속 몰려
1000명이 대기하는 상태가 되어도 무작위로 이루어지는데,
최악의 경우에는 2번째로 구매한 고객에게 마지막에 락이 부여되는 상황이 일어난다.
선착순 이벤트와 같은 상황에서는 이같은 공정성 문제가 발생한다.

synchronized 문제 해결 방법

무한 대기 : BLOCKED → WAITING

무한 대기가 발생하는 원인은 락 획득을 대기하는 스레드가 BLOCKED 상태이기 때문이다.
BLOCKED 상태는 Interrupt가 발생해도 빠져나갈 수 없다.

문제를 해결하기 위한 방식은 간단하다.
락 획득을 대기하는 스레드를 BLOCKED 상태가 아닌 WAITING 상태로 기다리게 만드는 것이다!
WAITING 상태는 Interrupt가 발생하면 빠져나갈 수 있기 때문에,
오류가 발생하거나 시간이 지연되는 경우에는 Interrupt를 발생시켜 흐름을 제어할 수 있다.

LockSupport

LockSupport.park()  // ** thread를 WAITING 상태로 변경 **
LockSupport.parkNanos(nanoTimes)  // ** thread를 TIMED_WAITING 상태로 변경 **
LockSupport.unpark(thread)  //  ** thread를 WAITING → RUNNABLE 상태로 변경 **

자바에서는 스레드를 WAITING 상태로 변경하기 위해 LockSupport를 지원한다.

하지만 LockSupport에는 치명적인 단점이 있다.
락을 획득했지만, 내부적인 요인에 의해 로직을 처리할 수 없을 때,
park()는 락을 보유하면서 WAITING 상태로 변경된다는 점이다.

예를 들어 재고가 0인 경우, 물건을 판매해도 재고를 차감시킬 수 없다.
재고가 추가되기 전까지 대기하는 로직이 있어,
스레드는 park() 메서드를 활용하여 WAITING 상태로 대기하는데
해당 스레드가 락을 보유하고 있어 재고를 추가하기 위해 접근하는 스레드도
락을 획득하기 위해 대기하는 상황이 벌어진다!

무한 대기를 해결하기 위해 도입한 park()로 인해 또 다른 무한 대기 문제가 발생한다.

Object

park() 대신 Object의 메서드를 활용하면 위와 같은 문제를 해결할 수 있다.

wait();  //  ** WAITING 상태로 변경하면서 락 반납 **
notify();  //  ** WAITING 상태 스레드 중 하나 무작위 RUNNABLE 상태 변경 **
notifyAll();  //  ** 모든 스레드를 WAITING 상태에서 RUNNABLE 상태로 변경 **

Object의 메서드는 WAITING 상태로 변경함과 동시에 락을 반납한다.
대기하는 스레드들은 이제 Interrupt로 흐름을 제어할 수 있으면서도
락을 보유하고 있지 않아 무한 대기 문제에 유연하게 대처할 수 있다!

하지만 아래에 등장하는 생산자 소비자 문제로 결국에는 ReentrantLock의 기능을 사용하게 된다.

공정성 문제 : ReentrantLock 공정 모드

ReentrantLock은 synchronized 키워드에서 발생한 문제들을 보완한 방법이다.
멀티 스레드의 동시성 문제를 해결하기 위해 사용해야 하는 자바 기능으로, 바로 뒤에서 학습한다.

ReentrantLock의 기능을 활용해서 동시성 문제를 해결하게 되는데,
공정성 문제는 ReentrantLock 객체를 생성할 때 적용할 수 있다.

Lock lock = ReentrantLock(true)  // ** 공정 모드로 ReentrantLock 객체 생성 **
Lock lock = ReentrantLock()  // ** 비공정 모드로 ReentrantLock 객체 생성 **

공정 모드로 ReentrantLock을 생성할 경우,
락 획득을 위해 대기하는 스레드에 순서대로 락을 부여한다.
순서대로 락을 부여하기 위한 로직이 추가된 만큼, 성능이 저하된다는 단점이 있다.

반대로 비공정 모드로 ReentrantLock을 생성할 경우,
추가된 로직이 없어 락 획득 속도가 빠르다는 장점이 있지만 공정성 문제를 해결할 수 없다.

생산자 소비자 문제

공정 모드도 한계가 있다.
바로 생산자 소비자 문제이다.

재고를 추가하는 생산자가 있고, 재고를 차감하는 소비자가 있다.
재고가 0이 되면 소비자는 재고가 추가될 때까지 기다리고,
재고가 max가 되면 생산자는 재고가 차감될 때까지 기다린다.

예를 들어 재고가 0이 되어 소비자 스레드들이 재고 추가를 기다리고 있다.
생산자 스레드가 재고를 하나 만들어 추가하려고 하지만,
공정 모드에서 먼저 줄을 서고 있는건 소비자 스레드들이다.

절대 공정.
생산자 스레드에게 락을 부여하기 위해 순서대로 소비자 스레드들에게 먼저 락을 부여하지만,
소비자 스레드가 할 수 있는 것은 없다.
그저 순서를 지키기 위한 비효율적인 락 부여만이 수행될 뿐이다.

각자 보유한 역할이 있는 스레드들이 한 곳에 대기하여 관리되는 것은
이와 같은 비효율적인 상황을 발생시킨다.

뒤 이어 등장하는 ReentrantLock은 대기 공간을 분리할 수 있는 기능까지 장착되어 있다.

ReentrantLock

멀티 스레드의 동시성 문제를 해결하기 위해서,
자바에서는 스레드를 활용할 때 ReentrantLock을 사용해야 한다.
ReentrantLock의 다양한 기능을 살펴보자.

ReentrantLock 생성

ReentrantLock 객체를 생성한 후 락 기능을 사용할 수 있다.

Lock lock = new ReentrantLock()  // true 파라미터 전달 시 공정 모드 ON

lock(), unlock()

synchronized 키워드로 임계 영역을 설정한 것처럼,
lock() 메서드 호출 이후 선언된 코드들은 락을 반납하기 전까지 임계 영역으로 돌입한다.
lock() 메서드는 Interrupt에 응답하지 않고, 락을 획득할 때까지 무한 대기한다.

class Item {
	
    Lock lock = new ReentrantLock();
    
    int quantity;
    
    public Item(int quantity) {
    	this.quantity = quantity;
    }
    
	public void consume() {
    	lock.lock();  // ** 임계 영역 돌입 **
        try {
        	quantity -= 1;
        } finally {
        	lock.unlock();  // ** finally 블럭에서 락 반납 **
        }
	}
}   	

lock() 메서드를 호출하여 락을 획득한 이후에는 돌발 상황에서도 락을 반납할 수 있도록
반드시 finally 블럭에서 unlock() 메서드를 호출해야 한다.
lock() 메서드를 호출한 시점부터 unlock() 메서드를 호출하는 시점까지가 바로 임계 영역이다.

lockInterruptibly()

lock() 메서드는 인터럽트에 응답하지 않아 흐름을 제어할 수 없다.
lockInterruptibly()로 임계 영역을 시작하면 Interrupt에 응답한다.

class Item {
	
    Lock lock = new ReentrantLock();
    
    int quantity;
    
    public Item(int quantity) {
    	this.quantity = quantity;
    }
    
	public void consume() {
    	lock.lockInterruptibly();  // ** 임계 영역 내부에서 인터럽트에 반응 **
        try {
        	quantity -= 1;
        } finally {
        	lock.unlock();
        }
	}
}   	

tryLock()

lock() 메서드는 락을 획득할 때까지 무한 대기하기 때문에 문제가 발생할 수 있다.

tryLock()은 락을 획득하는 경우 true, 획득하지 못하는 경우 즉시 false를 반환하여
락 획득을 위해 대기하지 않는다.

class Item {
	
    Lock lock = new ReentrantLock();
    
    int quantity;
    
    public Item(int quantity) {
    	this.quantity = quantity;
    }
    
	public boolean consume() {
    	if (!lock.tryLock()) {
        	// ** 락 획득 실패 **
            
            return false;
        }
        
        try {  // ** 락을 획득한 경우에 로직 실행 **
        	quantity -= 1;
        } finally {
        	lock.unlock();
        }
        
        return true;
	}
}   	

tryLock(long time, TimeUnit unit)

lock()은 무한 대기를, tryLock()은 대기 자체를 하지 않는다.
이 두 상황 외에 일정 시간만 대기하고 싶다면,
tryLock() 메서드에 대기 시간과 시간 단위를 파라미터로 전달하면 된다.

class Item {
	
    Lock lock = new ReentrantLock();
    
    int quantity;
    
    public Item(int quantity) {
    	this.quantity = quantity;
    }
    
	public boolean consume() {
    	if (!lock.tryLock(10, TimeUnit.SECONDS)) {
        	// ** 10초 대기 이후 락 획득 실패
            
            return false;
        }
        
        try {  // ** 락을 획득한 경우에 로직 실행 **
        	quantity -= 1;
        } finally {
        	lock.unlock();
        }
        
        return true;
	}
}   	

Condition, await(), signal()

공정 모드에서도 특정 역할의 스레드에게 락을 부여하지 못하는 한계가 있었다.

ReentrantLock의 Condition은 스레드가 대기할 수 있는 공간으로,
원하는 스레드를 원하는 대기 공간에 대기시킬 수 있다.

Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();


condition.await()  // ** 스레드 락 반납 및 WAITING 상태 변경 
condition.signal()  // ** condition 객체 대기 스레드 중 하나 WAITING → RUNNABLE

await() 메서드와 signal() 메서드를 통해서 스레드 상태를 제어할 수 있다.

class Item {
	
    Lock lock = new ReentrantLock();
    Condition producerCondition = lock.newCondition();
    Condition consumerCondition = lock.newCondition();
    
    int quantity;
    public static final int MAX = 100;
    
    public Item(int quantity) {
    	this.quantity = quantity;
    }
    
    // 재고 소비
	public boolean consume() {
    	if (!lock.tryLock(10, TimeUnit.SECONDS)) {
        	// 10초 대기 이후 락 획득 실패
            
            return false;
        }
        
        try {
        	
            while(quantity == 0) {
            	try {
                	consumerCondition.await();  // ** 재고가 없으면 소비자 전용 공간에서 대기 **
                } catch (InterruptException e) {
                	throw new RuntimeException(e);
                }
            }
           
        	quantity -= 1;
            producerCondition.signal();  // ** 재고 공간에 여유가 생겼으므로 생산자 스레드 상태 전환 **
        } finally {
        	lock.unlock();
        }
        
        return true;
	}
    
    // 재고 추가
    public boolean produce() {
    	if (!lock.tryLock(10, TimeUnit.SECONDS)) {
        	// 10초 대기 이후 락 획득 실패
            
            return false;
        }
        
        try {
        	while(quantity == MAX) {
            	try {
                	producerCondition.await();  // ** 재고가 가득 찼으므로 생산자 전용 공간에서 대기 **
                } catch (InterruptException e) {
                	throw new RuntimeException(e);
                }
            }
            
            quantity++;
            consumerCondition.signal();  // ** 재고가 추가됐으므로 소비자 스레드 상태 전환 **
        }
        
        return true;
    }
}

이제 스레드 역할 별로 분리해서 대기 공간을 사용할 수 있다.
당연한 이야기지만, 스레드는 대기 공간에서 나오더라도 락을 획득해야 로직을 수행할 수 있다.
signal() 메서드를 호출해서 스레드를 깨워도,
unlock()으로 락을 반납해야 로직을 수행할 수 있다는 이야기이다.

BlockingQueue

위 예시에서는 재고를 단순히 필드를 선언해서 표현했지만,
실제 데이터를 보관하기 위해서는 Buffer가 필요하다.

자바에서는 멀티 스레드 환경에서 사용할 수 있는 Buffer인 BlockingQueue를 지원하는데,
말 그대로 차단하는 Queue라는 의미로,
BlockigQueue의 메서드는 데이터 추가, 획득 시에 다른 스레드의 접근을 차단하도록 설계되어 있다.

BlockingQueue 인터페이스는 구현체로 ArrayBlockingQueue, LinkedBlockingQueue 등을 지원하는데,
무한으로 대기하는 상황, 대기하지 않는 상황, 시간 별로 대기하는 상황 등
다양한 상황에 대한 메서드를 별도로 지원한다.

Blockingqueue<T> queue = new ArrayBlockingQueue<>();

Exception 발생 메서드 : add, remove, element

• add(data) : data 추가, 가득 찬 경우 Exception throw
• remove() : data 획득, 비어있는 경우 Exception throw
• element() : data를 획득하지만 Queue에서 제거하지는 않는다, 비어있는 경우 Exception throw

무한 대기 메서드 : put, take

• put(data) : data 추가, 가득 찬 경우 무한 대기
• take() : data 획득, 없는 경우 획득할 때까지 대기

대기하지 않는 메서드 : offer, poll, peek

• offer(data) : data 추가, 가득 찬 경우 false
• poll() : data 획득, 비어있는 경우 null
• peek() : data를 획득하지만 Queue에서 제거하지는 않는다, 비어있는 경우 null

일정 시간 대기하는 메서드 : offer, poll

• offer(data, time, unit) : data 추가, time이 지나면 false
• poll(time, unit) : data 획득, time이 지나면 null

마무리

참고로 멀티 스레드의 동기화 문제에서
필드의 메모리 가시성 문제는 volatile을 선언하지 않아도 자동으로 해결된다.
메모리 가시성 문제에 대해서는 아래 게시글을 참고하자.

• volatile, 메모리 가시성

멀티 스레드를 활용하기 위해서 ReentrantLock을 사용한다는 점,
대기 공간으로는 Condition 객체를 사용한다는 점,
그리고 멀티 스레드의 Buffer로 BlockingQueue를 사용한다는 점을 잊지 말자.