notFull·notEmpty 조건 분리
두 Condition이 반대 역할만 깨우도록 신호 방향을 설계하고, 유한 큐의 불변식·종료·통계를 포함한 구현으로 각성 낭비를 줄입니다.
유한 큐에는 서로 반대인 두 조건이 있습니다.
생산자는 size < capacity를 기다리고, 소비자는 size > 0을 기다립니다.
한 대기 집합에 두 역할을 섞으면 신호가 같은 역할을 깨울 수 있습니다.
ReentrantLock.newCondition()을 두 번 호출하면 같은 공유 상태와 락을 사용하면서도 대기실을 역할별로 나눌 수 있습니다.
이름은 기다리는 상태가 아니라 통과 가능한 조건을 붙이는 편이 읽기 좋습니다.
notFull은 생산자가 기다리는 곳이고, 소비가 빈칸을 만든 뒤 신호합니다.
notEmpty는 소비자가 기다리는 곳이며, 생산이 항목을 넣은 뒤 신호합니다.
자신과 같은 역할을 깨우는 것이 아니라 방금 만족 가능해진 반대 술어의 대기자를 깨운다는 방향이 핵심입니다.
신호 방향을 뒤집으면 모두 잠드는 실패
아래 클래스는 대기실을 분리했지만 생산 후 notFull을, 소비 후 notEmpty를 알립니다.
빈 큐에서 소비자가 기다리는 동안 생산자는 값을 넣고 생산자용 조건에 신호합니다.
소비자 대기실에는 알림이 도착하지 않아 값이 있는데도 작업이 끝나지 않습니다.
import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public final class ReversedConditionSignal {
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
private final Queue<String> queue = new ArrayDeque<>();
void put(String value) throws InterruptedException {
lock.lockInterruptibly();
try {
while (queue.size() == 1) notFull.await();
queue.add(value);
notFull.signal();
} finally {
lock.unlock();
}
}
String take() throws InterruptedException {
lock.lockInterruptibly();
try {
while (queue.isEmpty()) notEmpty.await();
String value = queue.remove();
notEmpty.signal();
return value;
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
ReversedConditionSignal buffer = new ReversedConditionSignal();
Thread consumer = Thread.ofPlatform().start(() -> {
try { System.out.println(buffer.take()); }
catch (InterruptedException e) { Thread.currentThread().interrupt(); }
});
Thread.sleep(30);
buffer.put("stored-but-not-signaled");
consumer.join(150);
System.out.println("consumerAlive=" + consumer.isAlive());
consumer.interrupt();
}
}관찰 결과는 보통 consumerAlive=true입니다.
공유 상태와 잠금은 올바른데 사건과 신호의 연결이 틀렸습니다.
생산은 “더 이상 비어 있지 않음”을 만들고, 소비는 “더 이상 가득 차지 않음”을 만듭니다.
코드 리뷰에서는 queue.add 다음 notEmpty.signal, queue.remove 다음 notFull.signal이 짝을 이루는지 확인합니다.
술어·변경자·대기실 대응표
조건 분리는 성능 최적화 이전에 모델 명료화입니다.
어떤 메서드가 술어를 거짓에서 참으로 바꾸는지 적으면 신호 대상이 자동으로 나옵니다.
삽입은 notEmpty를 참으로 만들 수 있고, 제거는 notFull을 참으로 만들 수 있습니다.
크기가 여러 칸 변하지 않는 단일 연산에서는 반대 역할 한 명을 깨우는 signal로 충분합니다.
Condition두 개는 반드시 큐 필드를 보호하는 동일한ReentrantLock에서 만들어야 합니다.- 생산 대기는 notFull에서 이루어지고 삽입 완료 알림은 notEmpty로 향합니다.
- 소비 대기는 notEmpty에서 이루어지며 제거 직후 notFull의 한 작업자를 깨웁니다.
- 복귀한 스레드는 지정된 대기실에서도 경쟁이나 임의 복귀가 가능하므로 술어를 반복 확인합니다.
- 종료처럼 많은 대기자의 조건을 동시에 바꾸는 사건은 두 조건 모두
signalAll해야 합니다. - 배치로 여러 슬롯을 만들거나 채우는 연산은 몇 명을 깨울지 정확성·비용을 함께 검토합니다.
signal을 쓰더라도 공정한 실행 순서를 얻는 것은 아닙니다.
같은 조건에 여러 생산자가 있다면 어떤 생산자가 깨어날지 구현과 스케줄러에 달려 있습니다.
큐의 요구가 선착순 보장을 포함한다면 단순 Condition 신호만으로 규칙을 선언하지 말고 별도의 요청 순번이나 검증된 공정 구조를 설계해야 합니다.
역할별 신호와 정상 종료를 갖춘 큐
완전 구현에는 닫힘 상태를 추가합니다.
닫힌 뒤 생산은 거부하고, 소비자는 이미 들어온 항목을 모두 배출합니다.
마지막 항목까지 제거한 뒤 다음 take는 빈 결과를 반환합니다.
close()는 생산자와 소비자 모두의 대기 조건을 바꿀 수 있으므로 양쪽 대기실 전체를 깨웁니다.
import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public final class DualConditionQueue<E> {
private final Queue<E> queue = new ArrayDeque<>();
private final int capacity;
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition notEmpty = lock.newCondition();
private boolean closed;
public DualConditionQueue(int capacity) {
if (capacity < 1) throw new IllegalArgumentException("capacity");
this.capacity = capacity;
}
public void put(E value) throws InterruptedException {
if (value == null) throw new NullPointerException("value");
lock.lockInterruptibly();
try {
while (queue.size() == capacity && !closed) notFull.await();
if (closed) throw new IllegalStateException("queue closed");
queue.add(value);
notEmpty.signal();
} finally {
lock.unlock();
}
}
public Optional<E> take() throws InterruptedException {
lock.lockInterruptibly();
try {
while (queue.isEmpty() && !closed) notEmpty.await();
if (queue.isEmpty()) return Optional.empty();
E value = queue.remove();
notFull.signal();
return Optional.of(value);
} finally {
lock.unlock();
}
}
public void close() {
lock.lock();
try {
closed = true;
notFull.signalAll();
notEmpty.signalAll();
} finally {
lock.unlock();
}
}
public static void main(String[] args) throws InterruptedException {
DualConditionQueue<String> jobs = new DualConditionQueue<>(2);
jobs.put("compile");
jobs.put("run");
jobs.close();
System.out.println(jobs.take().orElseThrow());
System.out.println(jobs.take().orElseThrow());
System.out.println("finished=" + jobs.take().isEmpty());
}
}출력은 두 작업과 finished=true입니다.
닫힌 큐가 가득 찬 상태에서 기다리던 생산자는 close의 전체 신호를 받고 예외로 빠져나옵니다.
빈 상태에서 기다리던 소비자는 종료 결과를 얻습니다.
한쪽만 깨우면 애플리케이션 종료 시 일부 작업이 남아 JVM을 붙잡을 수 있습니다.
단일 조건과 이중 조건의 각성 수 비교
성능 비교는 동일한 업무량과 용량에서 해야 합니다. 아래 탐침은 두 조건 큐로 400개 정수를 전달하고 합계를 검사합니다. 정확한 처리 시간은 환경 의존이므로 고정 임계값을 두지 않고 실행 시간과 총합을 함께 출력합니다. 단일 Condition 버전과 같은 머신에서 여러 차례 측정해 중앙값을 비교합니다.
import java.util.ArrayList;
import java.util.List;
public final class DualConditionThroughputProbe {
public static void main(String[] args) throws InterruptedException {
DualConditionQueue<Integer> queue = new DualConditionQueue<>(8);
List<Thread> workers = new ArrayList<>();
long[] partial = new long[4];
long started = System.nanoTime();
for (int producer = 0; producer < 4; producer++) {
int base = producer * 100;
workers.add(Thread.ofPlatform().start(() -> {
try {
for (int n = 1; n <= 100; n++) {
queue.put(base + n);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}));
}
for (int consumer = 0; consumer < 4; consumer++) {
int slot = consumer;
workers.add(Thread.ofPlatform().start(() -> {
try {
long local = 0;
for (int n = 0; n < 100; n++) {
local += queue.take().orElseThrow();
}
partial[slot] = local;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}));
}
for (Thread worker : workers) {
worker.join();
}
long elapsedMicros = (System.nanoTime() - started) / 1_000;
long total = 0;
for (long value : partial) total += value;
System.out.println("sum=" + total);
System.out.println("elapsedMicros=" + elapsedMicros);
}
}기대 합계는 각 생산 구간을 더한 80,200입니다. 작은 표본의 시간 차이는 JIT 컴파일, OS 스케줄, 백그라운드 부하에 흔들리므로 벤치마크 결론으로 사용하지 않습니다. 이 탐침의 첫 목적은 역할 분리 후에도 총량이 보존되고 모든 작업이 종료되는지 확인하는 것입니다.
조건 분리의 운영 판단
| 징후 | 단일 대기 집합 영향 | 두 Condition의 가치 |
|---|---|---|
| 생산자·소비자가 다수 | 반대 아닌 역할 각성 | 정확한 신호 방향 제공 |
| 큐 점유가 경계에 자주 닿음 | 재대기 증가 | 잠금 경쟁 감소 가능 |
| 종료가 복잡함 | 전체 알림 한 곳 | 두 대기실 모두 종료 신호 필요 |
| 구현 유지 인력이 적음 | 저수준 규약 부담 | BlockingQueue 위임 우선 |
직접 만든 이중 조건 큐가 표준 ArrayBlockingQueue보다 낫다는 결론은 나오지 않습니다.
표준 구현도 같은 핵심 구조를 검증된 형태로 제공하며 다양한 연산 규칙이 있습니다.
이 구현은 신호 방향을 이해하고 특별한 상태 조건이 필요한 경우에만 확장 기반으로 삼습니다.
연습 문제
용량이 하나인 처리 슬롯에 높은 우선순위 작업과 일반 작업이 들어옵니다. 소비자는 높은 우선순위 큐를 먼저 확인하되, 일반 작업이 영원히 굶지 않도록 높은 작업을 최대 세 번 연속 처리한 뒤 일반 작업을 하나 선택하세요. 데이터 존재 조건과 빈 슬롯 조건을 분리하고 종료 시 모든 대기자를 깨웁니다.
정답과 해설
두 데이터 큐를 합친 총 크기가 용량 조건을 결정합니다.
소비 가능 조건은 어느 큐든 비어 있지 않은 상태입니다.
선택 정책은 잠금 안에서 highStreak과 함께 바꿔야 여러 소비자가 연속 횟수를 어긋나게 보지 않습니다.
생산자는 공통 notFull, 소비자는 공통 hasWork에서 대기합니다.
import java.util.ArrayDeque;
import java.util.Optional;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
public final class FairPrioritySlotSolution<E> {
private final ReentrantLock lock = new ReentrantLock();
private final Condition notFull = lock.newCondition();
private final Condition hasWork = lock.newCondition();
private final Queue<E> high = new ArrayDeque<>();
private final Queue<E> normal = new ArrayDeque<>();
private final int capacity;
private int highStreak;
private boolean closed;
FairPrioritySlotSolution(int capacity) {
if (capacity < 1) throw new IllegalArgumentException("capacity");
this.capacity = capacity;
}
void submit(E value, boolean urgent) throws InterruptedException {
lock.lockInterruptibly();
try {
while (size() == capacity && !closed) notFull.await();
if (closed) throw new IllegalStateException("closed");
(urgent ? high : normal).add(value);
hasWork.signal();
} finally {
lock.unlock();
}
}
Optional<E> take() throws InterruptedException {
lock.lockInterruptibly();
try {
while (size() == 0 && !closed) hasWork.await();
if (size() == 0) return Optional.empty();
E value;
if (!high.isEmpty() && (highStreak < 3 || normal.isEmpty())) {
value = high.remove();
highStreak++;
} else {
value = normal.remove();
highStreak = 0;
}
notFull.signal();
return Optional.of(value);
} finally {
lock.unlock();
}
}
void close() {
lock.lock();
try {
closed = true;
notFull.signalAll();
hasWork.signalAll();
} finally {
lock.unlock();
}
}
private int size() { return high.size() + normal.size(); }
public static void main(String[] args) throws InterruptedException {
FairPrioritySlotSolution<String> queue = new FairPrioritySlotSolution<>(5);
queue.submit("normal", false);
queue.submit("h1", true);
queue.submit("h2", true);
queue.close();
System.out.println(queue.take().orElseThrow());
System.out.println(queue.take().orElseThrow());
System.out.println(queue.take().orElseThrow());
}
}예시는 높은 작업 두 개 뒤 일반 작업을 출력합니다. 높은 작업이 네 개 이상 쌓인 경우에는 세 번째 이후 일반 작업이 한 번 선택되어 기아를 제한합니다. 엄밀한 대기 시간 상한이 필요한 서비스라면 연속 횟수보다 도착 시각과 마감 시각을 사용하는 스케줄러가 적합합니다.
대기실 이름과 신호 방향을 함께 읽기
두 Condition을 도입한 코드는 각 상태 변경 다음 줄만 보아도 의도를 확인할 수 있어야 합니다.
추가는 notEmpty 또는 hasWork, 제거는 notFull을 알립니다.
종료처럼 모든 역할의 술어를 바꾸는 전이는 양쪽 전체 신호를 사용합니다.
대기 공간 분리는 잘못된 역할 각성을 줄이지만 조건 재검사와 취소 처리를 없애지 않습니다.
효율화 이후에도 while, try-finally, 인터럽트 전파가 남아 있어야 동일한 정확성 규칙을 유지합니다.