BlockingQueue 연산 정책
`add`·`offer`·`put`과 `remove`·`poll`·`take`의 결과 형식을 비교하고, 시간 예산·배치 배출·인터럽트가 일관된 작업 수집기를 완성합니다.
BlockingQueue에는 비슷해 보이는 삽입과 제거 메서드가 여러 개 있습니다.
차이는 문법이 아니라 실패를 표현하는 방식입니다.
가득 찼을 때 예외를 던질지, 거짓을 반환할지, 빈칸이 날 때까지 기다릴지, 일정 시간만 양보할지를 선택합니다.
소비 쪽도 빈 큐에서 예외·빈 값·무한 대기·제한 대기 중 하나를 고릅니다.
호출 경로의 지연 예산과 실패 빈도를 먼저 정하면 메서드가 자연스럽게 좁혀집니다.
과부하가 정상적으로 발생할 수 있는 웹 요청에서 add 예외를 흐름 제어로 쓰면 로그와 예외 비용이 불필요하게 커집니다.
반드시 전달해야 하는 내부 파이프라인에서 즉시 offer 결과를 무시하면 유실됩니다.
같은 큐라도 사용 지점별 정책이 다를 수 있습니다.
용량 확인과 add 사이의 경쟁
아래 코드는 두 생산자가 remainingCapacity()가 1인 것을 함께 본 뒤 둘 다 add를 호출할 수 있습니다.
한쪽은 성공하지만 다른 쪽은 IllegalStateException: Queue full을 만납니다.
관찰 메서드와 상태 변경 메서드를 분리하면 원자성이 사라진다는 예입니다.
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
public final class CheckThenAddRace {
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
CountDownLatch checked = new CountDownLatch(2);
CountDownLatch go = new CountDownLatch(1);
Runnable producer = () -> {
if (queue.remainingCapacity() > 0) {
checked.countDown();
try {
go.await();
queue.add(Thread.currentThread().getName());
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
};
Thread a = Thread.ofPlatform().name("A").start(producer);
Thread b = Thread.ofPlatform().name("B").start(producer);
checked.await();
go.countDown();
a.join();
b.join();
System.out.println("queue=" + queue);
}
}예외가 어느 스레드에서 발생할지는 정해지지 않습니다.
올바른 즉시 제출은 queue.offer(value) 한 번의 반환값으로 성공을 판단합니다.
남은 용량은 대시보드나 디버깅에는 쓸 수 있지만 미래 성공을 예약하는 토큰이 아닙니다.
검사-행동 경쟁을 없애는 API가 이미 있다면 외부 잠금으로 큐를 다시 감싸지 않습니다.
연산 계열과 업무 의미
예외 계열은 실패가 불변식 위반처럼 드문 사건일 때 적합합니다. 특수값 계열은 현재 상태에서 불가능한 일이 정상 분기일 때 사용합니다. 무한 대기 계열은 호출자 지연 상한이 없고 취소가 보장될 때만 선택합니다. 시간 제한 계열은 일정 예산을 큐에 양보한 뒤 폴백해야 하는 경계에 맞습니다.
- 삽입
add는 즉시 시도하고 가득 차면 예외를 던지므로 설정상 절대 차면 안 되는 큐의 검증에 쓸 수 있습니다. - 삽입
offer는 불리언 결과가 과부하 분기이며 반환값을 반드시 소비해야 합니다. - 삽입
put은 공간까지 기다리고 인터럽트로 취소되므로 내부 역압에 어울립니다. - 제거
remove는 빈 큐를 예외로 다루고poll은 현재 부재를null로 표현합니다. - 제거
take는 항목 도착을 기다리며 시간 제한poll은 정해진 휴지 후 제어권을 돌려줍니다. drainTo는 잠금 획득 횟수를 줄이는 배치 이동이지만 대상 컬렉션 처리 실패와 메모리 크기를 고려합니다.
null을 큐 원소로 허용하지 않는 이유는 poll의 부재 신호와 충돌하기 때문입니다.
타입이 있는 데이터에서도 이 규칙을 유지하면 소비자는 분기 하나로 현재 부재를 판단할 수 있습니다.
종료는 null이 아니라 별도 메시지나 실행기 수명주기로 표현합니다.
하나의 마감 시간을 공유하는 배치 수집기
연속된 작업마다 100밀리초씩 기다리면 전체 요청 시간이 항목 수만큼 늘어납니다. 다음 수집기는 호출 시 정한 전체 시간 예산에서 남은 나노초를 매 제출에 전달합니다. 제한이 끝나면 남은 항목을 거부 목록에 기록하고 즉시 반환합니다. 인터럽트는 상위로 그대로 전파합니다.
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
public final class DeadlineBatchSubmitter<E> {
public record Result<E>(int accepted, List<E> rejected) {}
private final BlockingQueue<E> queue;
public DeadlineBatchSubmitter(BlockingQueue<E> queue) {
this.queue = queue;
}
public Result<E> submitAll(List<E> items, Duration totalBudget) throws InterruptedException {
long deadline = System.nanoTime() + totalBudget.toNanos();
int accepted = 0;
List<E> rejected = new ArrayList<>();
for (int index = 0; index < items.size(); index++) {
E item = items.get(index);
long remaining = deadline - System.nanoTime();
if (remaining <= 0 || !queue.offer(item, remaining, TimeUnit.NANOSECONDS)) {
rejected.addAll(items.subList(index, items.size()));
break;
}
accepted++;
}
return new Result<>(accepted, List.copyOf(rejected));
}
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> queue = new java.util.concurrent.ArrayBlockingQueue<>(2);
DeadlineBatchSubmitter<String> submitter = new DeadlineBatchSubmitter<>(queue);
Result<String> result = submitter.submitAll(
List.of("one", "two", "three"), Duration.ofMillis(2));
System.out.println("accepted=" + result.accepted());
System.out.println("rejected=" + result.rejected());
}
}소비자가 없으므로 처음 두 항목이 수용되고 세 번째가 거부됩니다.
subList를 그대로 반환하지 않고 복사하여 원본 목록 변경의 영향을 막습니다.
System.nanoTime으로 경과 시간을 계산하면 시스템 시계 보정에 흔들리지 않습니다.
매우 긴 기간의 덧셈 오버플로까지 다루는 라이브러리라면 잔여 시간 계산 유틸리티를 별도로 확인합니다.
drainTo 기반 잠금·처리 구간 분리
소비자가 매번 take한 뒤 외부 저장소에 쓰면 큐 잠금 자체는 짧지만 호출 횟수가 많습니다.
drainTo는 현재 들어 있는 여러 항목을 지역 배치로 옮깁니다.
외부 I/O는 이동이 끝난 뒤 수행하므로 큐 내부 잠금을 오래 잡지 않습니다.
빈 큐에서는 시간 제한 poll로 첫 항목만 기다린 뒤 추가 항목을 배출합니다.
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
public final class TimedBatchDrainer<E> {
private final BlockingQueue<E> queue;
private final int maxBatch;
TimedBatchDrainer(BlockingQueue<E> queue, int maxBatch) {
if (maxBatch < 1) throw new IllegalArgumentException("maxBatch");
this.queue = queue;
this.maxBatch = maxBatch;
}
List<E> nextBatch(Duration waitForFirst) throws InterruptedException {
E first = queue.poll(waitForFirst.toNanos(), TimeUnit.NANOSECONDS);
if (first == null) return List.of();
List<E> batch = new ArrayList<>(maxBatch);
batch.add(first);
queue.drainTo(batch, maxBatch - 1);
return List.copyOf(batch);
}
public static void main(String[] args) throws InterruptedException {
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(10);
queue.addAll(List.of(1, 2, 3, 4, 5));
TimedBatchDrainer<Integer> drainer = new TimedBatchDrainer<>(queue, 3);
System.out.println("batch=" + drainer.nextBatch(Duration.ofSeconds(1)));
System.out.println("remaining=" + queue.size());
}
}첫 배치는 [1, 2, 3], 잔여 크기는 2입니다.
지역 배치로 옮긴 뒤 저장에 실패하면 항목은 이미 큐에서 제거되었습니다.
재시도가 필요하다면 배치 처리기가 내구성 있게 기록하거나 실패 항목을 별도 재처리 큐로 넘겨야 합니다.
무조건 원래 큐 앞에 되넣으면 순서와 중복 규칙이 바뀔 수 있습니다.
호출 경계별 연산 선택표
| 호출 경로 | 허용 지연 | 적합한 메서드 |
|---|---|---|
| UI 요청의 선택적 미리보기 | 거의 0 | offer와 즉시 폴백 |
| 내부 필수 전달 파이프라인 | 취소 전까지 | put과 인터럽트 전파 |
| 제한된 API 응답 예산 | 수십 밀리초 | 시간 제한 offer |
| 주기적 배치 소비 | 첫 값만 제한 대기 | 시간 제한 poll 뒤 drainTo |
put이 항상 가장 안전하거나 offer가 항상 가장 빠른 답은 아닙니다.
필수 이벤트를 offer로 버리면 정확성이 무너지고, 선택적 통계를 put으로 막으면 사용자 요청 지연이 전파됩니다.
데이터 중요도와 호출자의 시간 예산을 함께 봐야 합니다.
연습 문제
첫 큐에서 문자열을 가져와 길이를 계산한 뒤 두 번째 큐에 넣는 변환기를 작성하세요.
입력 take와 출력 put 모두 인터럽트에 반응해야 합니다.
출력 큐가 차서 기다리는 동안 취소되면 이미 가져온 입력을 잃었다는 사실을 실패 기록에 남기고 스레드의 인터럽트 상태를 복구합니다.
정답과 해설
큐 사이 이동은 트랜잭션이 아닙니다.
입력 제거 후 출력 삽입 전에 취소될 수 있으므로 처리 중 항목을 지역 변수로 보관하고 실패 저장소에 기록합니다.
가장 바깥 작업 접점에서 InterruptedException을 잡아 상태를 복구한 뒤 종료합니다.
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
public final class InterruptibleQueueBridgeSolution {
record Failure(String input, String reason) {}
private final BlockingQueue<String> source;
private final BlockingQueue<Integer> target;
private final ConcurrentLinkedQueue<Failure> failures = new ConcurrentLinkedQueue<>();
InterruptibleQueueBridgeSolution(BlockingQueue<String> source,
BlockingQueue<Integer> target) {
this.source = source;
this.target = target;
}
void transferOne() {
String claimed = null;
try {
claimed = source.take();
target.put(claimed.length());
} catch (InterruptedException e) {
if (claimed != null) failures.add(new Failure(claimed, "cancelled-before-target"));
Thread.currentThread().interrupt();
}
}
public static void main(String[] args) throws InterruptedException {
BlockingQueue<String> source = new ArrayBlockingQueue<>(1);
BlockingQueue<Integer> target = new ArrayBlockingQueue<>(1);
source.put("abcdef");
target.put(99);
InterruptibleQueueBridgeSolution bridge = new InterruptibleQueueBridgeSolution(source, target);
Thread worker = Thread.ofPlatform().start(bridge::transferOne);
Thread.sleep(30);
worker.interrupt();
worker.join();
System.out.println("failures=" + bridge.failures.size());
System.out.println("interrupted=" + worker.isInterrupted());
}
}출력은 실패 1건과 인터럽트 상태 true입니다.
정확히 한 번 처리가 필요하다면 메모리 실패 목록으로는 부족하며 메시지 브로커의 승인·재전달이나 데이터베이스 트랜잭션 같은 내구 프로토콜이 필요합니다.
이 예제는 BlockingQueue 두 개를 연결할 때 생기는 원자성 구분을 드러냅니다.
메서드 이름에 드러난 과부하 정책
BlockingQueue 연산표를 암기하는 대신 호출자의 의도를 읽으세요. 지금 불가능하면 예외인가, 정상 거부인가, 기다릴 수 있는가, 총 예산은 얼마인가를 답하면 적합한 메서드가 결정됩니다. 반환값과 예외를 무시하지 않는 것이 구현의 절반입니다.
큐 앞뒤의 처리까지 하나의 원자적 작업으로 착각하지 마세요. 배출 이후 실패와 취소를 어디에 기록할지 정해야 합니다. 표준 큐는 스레드 간 전달을 안전하게 만들지만 업무 처리의 내구성까지 대신하지는 않습니다.