[Java] PriorityBlockingQueue

PriorityBlockingQueue, java.util.concurrent.PriorityBlockingQueue, PBQ, 동시성 우선순위 큐

정의

java.util.concurrent.PriorityBlockingQueue<E> 는 PriorityQueue 의 동시성 + blocking 버전. unbounded 힙 기반 BlockingQueue.

특징

• unbounded, 따라서 put 이 절대 block 하지 않음
• take 만 큐가 비어 있을 때 block
• 단일 ReentrantLock 으로 보호
• 원소는 Comparable 또는 Comparator 에 따라 정렬

동작

public void put(E e) {
    offer(e);   // 절대 block 안 함, unbounded
}

public E take() throws InterruptedException {
    lock.lockInterruptibly();
    try {
        while ((e = dequeue()) == null) notEmpty.await();
        return e;
    } finally { lock.unlock(); }
}

복잡도

작업	시간
offer/put	O(log n)
take/poll	O(log n)
peek	O(1)

사용 예

우선순위 작업 스케줄러 (멀티 스레드)

BlockingQueue<Task> queue = new PriorityBlockingQueue<>(
    100,
    Comparator.comparingInt(Task::priority).reversed()
);

// Worker pool
for (int i = 0; i < 4; i++) {
    new Thread(() -> {
        while (true) {
            Task t = queue.take();    // 우선순위 높은 것부터
            process(t);
        }
    }).start();
}

// Producer
queue.put(new Task(urgent));
queue.put(new Task(normal));

함정

1. unbounded → OOM

producer 가 빠르고 consumer 가 느리면 메모리 무한 증가. 외부 backpressure 필요.

2. 순회 순서

iterator 가 정렬 순서를 보장하지 않는다 (PriorityQueue 처럼). 정렬된 결과가 필요하면 모두 poll 후 처리.

참고

• BlockingQueue
• PriorityQueue
• Blocking