[Java] 생산자 소비자 문제 (2)

기존에 사용했던 synchronized와 Object.wait(), Object.nofity()만으로 해결하려고 하면 생산자와 소비자 그룹을 "분리"시켜 대기시킬 수 있는 기능이 없다. 이번에는 자바 1.5부터 도입된 Lock 인터페이스와 ReentrantLock 구현체를 사용해서 개선시켜보자.

 

우선 클래스에서 ReentrantLock을 사용하기 위해 ReentrantLock 객체를 생성해준다.

private final Lock lock = new ReentrantLock();

 

그 다음, lock.newCondition()을 통해 스레드의 대기 집합을 만들어줄 수 있다.

private final Condition condition = lock.newCondition();

 

위와 같이 하나의 대기 집합만 사용한다면, Java의 모니터 락과 동일하게, 객체 하나에서 하나의 대기 집합만 가지는 상태와 동일하다. 이제는 ReentrantLock을 사용해서 대기 집합을 분리시키고, 이전의 비효율 문제를 개선해보자.

 

1. 대기 집합 분리

private final Condition producerCond = lock.newCondition(); // 셍산자 대기 집합
private final Condition consumerCond = lock.newCondition(); // 소비자 대기 집합

위와 같이 생산자 대기 집합과 소비자 대기 집합을 따로 만들어줄 수 있다. 이제 분리한 대기 집합을 어떻게 활용할지 알아보자.

 

2. Object.wait() 교체하기

Object.wait()을 대신해서 condition.await()을 사용하면 ReentrantLock의 스레드 대기 기능을 사용할 수 있다.

    public void put(String data) {
        lock.lock();
        try {
            while (queue.size() == max) {
                log("[put] 큐가 가득 참, 생산자 대기");
                try {
                    producerCond.await(); 	// this.wait() 대신 "await" 
                    // wait()은 이미 모든 객체가 가지고 있기 때문에 await을 쓴듯. wait쓰지 않도록 주의.
                    log("[put] 생산자 깨어남");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            queue.offer(data);
            log("[put] 생산자 데이터 저장, 소비자 호출");
            consumerCond.signal();
        } finally {
            lock.unlock();
        }

    }

 

위와 같이 현재 실행 중인 스레드를 producerCond.await() 메서드를 통해 producerCond의 스레드 대기 집합에 넣고 스레드를 대기 시킬 수 있다.

 

wait()은 모든 객체가 가지고 있기 때문에 await()을 사용한 듯 하다. condition.wait()을 사용하지 않도록 주의. 비동기 await도 아님.

 

3. Object.notify() 교체하기

Object.notify()를 대신해서 condition.signal()을 사용하면 ReentrantLock의 스레드 대기 집합 중, 원하는 스레드 대기 집합의 스레드만 깨울 수 있다.

public void put(String data) {
    lock.lock();
    try {
        while (queue.size() == max) {
            log("[put] 큐가 가득 참, 생산자 대기");
            try {
                producerCond.await();
                log("[put] 생산자 깨어남");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        queue.offer(data);
        log("[put] 생산자 데이터 저장, 소비자 호출");
        consumerCond.signal(); // this.notify() 대신
    } finally {
        lock.unlock();
    }

}

 

위와 같이 consumerCond.signal() 메서드를 실행하면, consumerCond의 스레드 대기 집합에 있는 스레드만 깨워준다.

 

4. 생산 소비 전체 코드

public void put(String data) {
    lock.lock();
    try {
        while (queue.size() == max) {
            log("[put] 큐가 가득 참, 생산자 대기");
            try {
                producerCond.await(); // this.wait() 대신 "await"이다. wait()은 이미 모든 객체가 가지고 있기 때문에 await을 쓴듯. wait쓰지 않도록 주의.
                log("[put] 생산자 깨어남");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        queue.offer(data);
        log("[put] 생산자 데이터 저장, 소비자 호출");
        consumerCond.signal(); // this.notify() 대신
    } finally {
        lock.unlock();
    }

}

public String take() {
    lock.lock();
    try {
        while (queue.isEmpty()) {
            log("[take] 큐에 데이터가 없음, 소비자 대기");
            try {
                consumerCond.await();
                log("[take] 소비자 깨어남");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        String data = queue.poll();
        log("[take] 소비자 데이터 획득, 생산자 호출");
        producerCond.signal();
        return data;
    } finally {
        lock.unlock();
    }
}

 

 

put에서는 자신을 생산자 대기 그룹에 넣고, 소비자만 깨우고

take에서는 자신을 소비자 대기 그룹에 넣고, 생산자만 깨우고

 

스레드 대기 그룹을 생산자와 소비자로 분리시킴으로서, 소비자가 소비자 스레드를 깨운다거나 생산자가 생산자 스레드를 깨우는 비효율을 제거했다.

 

 

BlockingQueue

위에서 구현한 코드는 사실 BlockingQueue에 구현되어 있다. 생산자 소비자 문제를 느끼고 해결해나가는 과정을 경험하기 위해 직접 구현해본 것이다.

 

이제는 생산자 소비자 문제가 있을 경우, BlockingQueue를 사용하면 된다. BlockingQueue에 대한 내용은 다음 글을 확인