BlockingQueue 메서드 고르기: offer·put·poll·take

앞 글에서는 ArrayBlockingQueue의 put과 take로 생산자와 소비자를 연결했습니다. 둘 다 큐가 가득 찼거나 비어 있으면 상황이 바뀔 때까지 기다립니다. 그런데 기다리는 것이 언제나 옳은 선택은 아닙니다. 웹 요청을 처리하던 스레드가 통계 이벤트를 큐에 넣으려는데 큐가 가득 찼다고 해 봅시다. 여기서 put으로 기다리면 사용자는 통계 때문에 응답을 늦게 받습니다. 반대로 주문을 다음 처리 단계로 넘기는 스레드라면, 큐가 가득 찼다고 주문을 버려서는 안 됩니다.

그래서 BlockingQueue는 같은 '넣기'와 '꺼내기'를 네 가지 방식으로 제공합니다. 지금 할 수 없을 때 예외를 던지는 방식, false나 null을 돌려주는 방식, 될 때까지 기다리는 방식, 정해진 시간까지만 기다리는 방식입니다. 메서드 이름은 비슷하지만, 어느 것을 고르느냐에 따라 큐가 가득 찼을 때 프로그램이 하는 일이 달라집니다. 이 글에서는 네 방식을 직접 실행해 비교하고, 자주 하는 실수와 실무에서 자주 쓰는 두 가지 방법을 살펴봅니다.

큐가 가득 찼거나 비었을 때 메서드마다 다른 반응

먼저 네 방식이 실제로 어떻게 다른지 실행해 보겠습니다. 다음 프로그램은 용량이 1인 큐를 두 개 만들어, 하나는 원소 하나를 넣어 가득 채우고 다른 하나는 비워 둡니다. 그런 다음 가득 찬 큐에는 원소를 더 넣으려 하고, 빈 큐에서는 원소를 꺼내려 합니다.

import java.util.NoSuchElementException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public final class QueueFailureStyles {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> full = new ArrayBlockingQueue<>(1);
        full.add("first");
        try {
            full.add("second");
        } catch (IllegalStateException e) {
            System.out.println("add -> " + e);
        }
        System.out.println("offer -> " + full.offer("second"));
        System.out.println("offer(50ms) -> " + full.offer("second", 50, TimeUnit.MILLISECONDS));

        BlockingQueue<String> empty = new ArrayBlockingQueue<>(1);
        try {
            empty.remove();
        } catch (NoSuchElementException e) {
            System.out.println("remove -> " + e);
        }
        System.out.println("poll -> " + empty.poll());
        System.out.println("poll(50ms) -> " + empty.poll(50, TimeUnit.MILLISECONDS));
        System.out.println("full=" + full + ", empty=" + empty);
    }
}

add -> java.lang.IllegalStateException: Queue full
offer -> false
offer(50ms) -> false
remove -> java.util.NoSuchElementException
poll -> null
poll(50ms) -> null
full=[first], empty=[]

add는 빈칸이 없으면 IllegalStateException을 던지고, offer는 false를 돌려줍니다. 시간 인자를 받는 offer는 50밀리초 동안 빈칸이 생기기를 기다리다가, 아무도 원소를 꺼내지 않으니 false를 돌려줍니다. 꺼내는 쪽도 구조가 같습니다. remove는 NoSuchElementException을 던지고, poll은 null을 돌려주며, 시간 제한 poll은 50밀리초를 기다린 뒤 null을 돌려줍니다. 마지막 줄을 보면 실패한 호출은 큐를 전혀 바꾸지 않았습니다. put과 take는 이 상황에서 끝없이 기다리므로 예제에 넣지 않았습니다.

도표는 공식 문서의 요약표 가운데 넣기와 꺼내기 부분을 옮긴 것입니다. 네 방식 가운데 기다리는 두 방식, 즉 put·take와 시간 제한 offer·poll은 기다리는 동안 인터럽트되면 InterruptedException을 던집니다. 이 예외는 검사 예외이므로 부르는 쪽에서 처리 방법을 정해야 하는데, 앞 글에서 본 것처럼 삼키지 말고 호출한 쪽에 넘기거나 인터럽트 상태를 되살려야 합니다. 맨 앞 원소를 꺼내지 않고 보기만 하는 element()와 peek()도 같은 식으로 나뉩니다. 큐가 비었을 때 element()는 예외를 던지고 peek()는 null을 돌려주며, 이 둘에는 기다리는 버전이 없습니다.

네 방식은 '지금은 할 수 없다'는 상황을 프로그램이 어떻게 받아들일지에 따라 고릅니다. 예외 방식인 add와 remove는 실패가 곧 버그를 뜻할 때 어울립니다. 예를 들어 큐 용량을 넣을 항목 수보다 크게 잡아 가득 찰 수 없게 설계했다면, add의 예외는 설계가 틀렸다는 신호입니다. 반대로 과부하가 정상적으로 일어날 수 있는 곳에서 add의 예외로 흐름을 제어하면, 예외 객체를 만들 때마다 스택 트레이스를 채우는 비용이 들고 로그도 쌓입니다. 공식 문서도 용량이 제한된 큐에서는 대개 add보다 offer를 쓰는 편이 낫다고 적고 있습니다.

false나 null을 돌려주는 offer와 poll은 실패가 정상적인 결과 가운데 하나일 때 씁니다. 버려도 되는 통계 이벤트라면 offer가 false를 돌려줄 때 버린 건수만 세면 됩니다. 이때 반환값은 반드시 확인해야 합니다. offer의 결과를 무시하면 꼭 전달해야 할 데이터도 소리 없이 사라집니다. poll이 null로 '꺼낼 원소가 없다'는 것을 알리기 때문에 BlockingQueue는 원소로 null을 받지 않습니다.

기다리는 put과 take는 호출한 쪽이 기다려도 되고, 데이터를 반드시 전달해야 할 때 씁니다. 앞 글의 생산자와 소비자처럼 작업자 스레드끼리 데이터를 주고받는 경우가 대표적입니다. 다만 끝없이 기다릴 수 있으므로 인터럽트로 그 스레드를 멈출 방법이 있어야 합니다. 시간 제한 버전은 기다릴 수는 있지만 상한이 있을 때 씁니다. 응답 시간 예산이 있는 요청 처리에서 정해진 시간만 기다려 보고, 실패하면 다른 방법으로 넘어가는 식입니다. 시간이 다 지났다는 false·null과 인터럽트로 생긴 InterruptedException이 구별되므로 '바빠서 못 넣었다'와 '취소됐다'를 따로 처리할 수 있습니다.

남은 용량을 확인한 뒤 add하면 생기는 경쟁

add가 예외를 던지는 것이 싫어서, 넣기 전에 remainingCapacity()로 빈칸이 있는지 먼저 확인하는 코드를 떠올릴 수 있습니다. remainingCapacity()는 큐에 지금 몇 칸이 비어 있는지 돌려주는 메서드입니다. 한 스레드만 쓰는 큐라면 이 방법도 맞지만, 여러 스레드가 함께 쓰는 큐에서는 확인과 삽입 사이에 다른 스레드가 끼어들 수 있습니다. 다음 프로그램은 그 순간을 일부러 만들어 냅니다.

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);
    }
}

용량이 1인 큐에 생산자 A와 B가 각각 자기 이름을 넣으려 합니다. CountDownLatch는 카운트가 0이 될 때까지 await()를 부른 스레드를 기다리게 하는 도구로, Runnable과 데몬 스레드에서 처음 썼습니다. 여기서는 두 래치를 경쟁이 매번 일어나게 하려고 넣었습니다. 각 생산자는 빈칸을 확인한 뒤 checked의 카운트를 하나 줄이고 go에서 기다립니다. main은 checked.await()로 두 생산자가 모두 확인을 마칠 때까지 기다렸다가, go.countDown()으로 둘을 동시에 출발시킵니다. 그러니 두 생산자는 모두 '빈칸이 1개'라는 것을 본 상태에서 add를 호출합니다.

Exception in thread "B" java.lang.IllegalStateException: Queue full
	at java.base/java.util.AbstractQueue.add(AbstractQueue.java:98)
	at java.base/java.util.concurrent.ArrayBlockingQueue.add(ArrayBlockingQueue.java:329)
	at CheckThenAddRace.lambda$main$0(CheckThenAddRace.java:15)
	at java.base/java.lang.Thread.run(Thread.java:1583)
queue=[A]

먼저 add를 실행한 A가 마지막 빈칸을 차지했고, B의 add는 확인할 때와 달리 빈칸이 없어 IllegalStateException을 던졌습니다. B는 이 예외를 잡지 않으므로 스레드가 그대로 끝나고, 기본 예외 처리기가 Exception in thread "B"로 시작하는 스택 트레이스를 표준 오류에 출력합니다. 다른 스레드에서 처리되지 않은 예외는 그 스레드만 끝낼 뿐이어서, main은 두 join 뒤에 큐 내용을 출력하고 종료 코드 0으로 끝납니다. 누가 이기는지는 실행마다 다릅니다. 12번 실행했을 때 위처럼 B가 실패한 것이 5번, A가 실패하고 queue=[B]가 나온 것이 7번이었습니다. 스택 트레이스에 나오는 JDK 내부 줄 번호는 JDK 버전에 따라 다를 수 있습니다(위 결과는 JDK 21).

원인은 확인과 삽입이 서로 다른 두 번의 호출이라는 데 있습니다. A와 B가 확인할 때는 빈칸이 정말 1개였지만, 그 정보는 확인한 순간에만 맞습니다. 이렇게 확인한 결과를 믿고 행동하는 검사 후 행동(check-then-act) 코드는 그 사이에 다른 스레드가 상태를 바꿀 수 있어서 동시성 버그의 흔한 원인이 됩니다. 공식 문서도 remainingCapacity()만 보고는 삽입이 성공할지 알 수 없다고 적고 있습니다. 다른 스레드가 곧 원소를 넣거나 뺄 수 있기 때문입니다.

고치는 방법은 확인과 삽입을 한 번의 호출로 합치는 것입니다. offer는 큐의 내부 락을 잡은 상태에서 빈칸을 확인하고 원소를 넣으므로, 그 사이에 다른 스레드가 끼어들 수 없습니다. 성공했는지는 반환값이 알려 줍니다. 다음 코드는 앞의 프로그램에서 확인 단계를 없애고 add 대신 offer를 쓴 것입니다. 확인 단계가 없으니 checked 래치도 필요 없습니다.

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;

public final class CheckThenOffer {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
        CountDownLatch go = new CountDownLatch(1);
        Runnable producer = () -> {
            try {
                go.await();
                String name = Thread.currentThread().getName();
                if (!queue.offer(name)) {
                    System.out.println(name + " rejected");
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        };
        Thread a = Thread.ofPlatform().name("A").start(producer);
        Thread b = Thread.ofPlatform().name("B").start(producer);
        go.countDown();
        a.join();
        b.join();
        System.out.println("queue=" + queue);
    }
}

B rejected
queue=[A]

이번에는 어느 쪽도 예외로 끝나지 않습니다. 진 쪽은 offer가 돌려준 false로 거절당했다는 것을 알고, 그에 맞게 처리할 수 있습니다. 10번 실행했을 때 A rejected와 queue=[B]가 나온 것이 3번이었고 나머지는 위와 같았습니다. 그러니 remainingCapacity()는 큐가 얼마나 찼는지 화면에 보여 주거나 디버깅할 때 쓰고, 삽입이 성공할지 미리 알아보는 데는 쓰지 않습니다.

확인과 삽입을 묶겠다고 큐 바깥에 락을 하나 더 두는 것도 피합니다. offer가 이미 그 일을 하고 있기 때문입니다. 게다가 바깥 락을 잡은 채 put에서 기다리면, 같은 락을 잡아야 원소를 꺼낼 수 있는 소비자까지 들어오지 못합니다. 잠든 스레드와 교착을 다룬 글에서 본 것처럼 모두가 멈출 수 있습니다.

여러 항목을 하나의 시간 예산 안에서 넣기

시간 제한 offer를 여러 번 부를 때는 기다릴 시간을 어떻게 나눌지 생각해야 합니다. 요청 하나가 항목 여러 개를 큐에 넣어야 하고, 이 일에 쓸 수 있는 시간이 100밀리초라고 해 봅시다. 항목마다 offer(item, 100, TimeUnit.MILLISECONDS)를 부르면, 큐가 막혀 있을 때는 항목마다 100밀리초를 다 기다립니다. 실제로 용량 1인 빈 큐에 네 항목을 이렇게 넣어 보면 첫 항목만 들어가고, 나머지 셋이 차례로 100밀리초씩 기다려 모두 약 300밀리초가 걸렸습니다. 항목이 늘어날수록 요청 시간도 그만큼 길어지는 것입니다.

해결책은 시작할 때 마감 시각을 한 번 정하고, 매번 남은 시간만 넘기는 것입니다. 다음 DeadlineBatchSubmitter는 이 방식으로 목록의 항목을 차례로 넣고, 넣지 못한 항목을 돌려줍니다.

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
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 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());
    }
}

submitAll은 먼저 지금 시각에 예산을 더해 마감 시각 deadline을 정합니다. 그리고 항목마다 남은 시간 remaining을 계산해, 이미 시간이 다 지났거나 offer가 남은 시간 안에 넣지 못하면 그 항목부터 끝까지를 rejected에 담고 반복을 멈춥니다. 예를 들어 2밀리초 예산에서 첫 항목을 넣는 데 0.6밀리초, 두 번째 항목을 넣는 데 0.8밀리초가 걸렸다면, 세 번째 offer가 기다릴 수 있는 시간은 2 − 0.6 − 0.8 = 0.6밀리초입니다. 앞에서 시간을 많이 쓸수록 뒤 항목이 기다릴 시간이 줄어들기 때문에, 각 offer에 넘기는 대기 시간을 모두 합쳐도 처음 정한 예산을 넘지 않습니다. 다만 목록을 만드는 시간이나 스레드가 다시 실행되기까지의 지연이 더해지므로, 메서드가 정확히 예산 안에 돌아온다는 뜻은 아닙니다.

accepted=2
rejected=[three]

main은 용량 2인 빈 큐에 세 항목을 2밀리초 예산으로 넣습니다. one과 two는 빈칸이 있으니 기다리지 않고 들어갑니다. three는 가득 찬 큐에서 남은 시간 동안 기다리지만 원소를 꺼내는 소비자가 없으므로 offer가 false를 돌려줍니다. 여러 번 실행해도 결과는 같았습니다. 앞에서 말한 용량 1인 큐와 네 항목에 이 메서드를 100밀리초 예산으로 쓰면, 약 300밀리초가 아니라 약 100밀리초 만에 accepted=1로 끝납니다.

도표처럼 결과는 언제나 앞쪽은 받아들여지고 뒤쪽은 거절되는 모양입니다. 이번 실행에서는 accepted=2였지만, 스레드가 첫 offer를 부르기 전에 2밀리초 넘게 멈춰 있었다면 accepted=0이 되어 세 항목이 모두 거절됩니다. 같은 이유로 accepted=1도 나올 수 있습니다. 실행해 보는 동안 나오지는 않았지만 코드상 가능한 경로입니다. 어느 경우든 이미 큐에 들어간 항목은 되돌리지 않습니다. 그래서 rejected에 남은 항목을 다시 시도할지, 오류로 응답할지, 다른 곳에 기록할지는 호출한 쪽이 정해야 합니다.

넣지 못한 부분은 items.subList(index, items.size())로 잘라 냅니다. subList는 원래 목록의 일부를 그대로 보여 주는 뷰(view)라서, 이것을 그대로 돌려주면 호출한 쪽이 원래 목록을 바꿀 때 결과도 영향을 받습니다. 그래서 새 ArrayList에 옮겨 담고, 마지막에 List.copyOf로 수정할 수 없는 복사본을 만들어 돌려줍니다. 이때 복사되는 것은 목록이고 원소 객체까지 복사하지는 않습니다. 인터럽트는 잡지 않고 throws InterruptedException으로 호출한 쪽에 넘깁니다.

더 깊이: nanoTime으로 마감 시각을 계산하는 이유

경과 시간을 잴 때는 System.currentTimeMillis() 대신 System.nanoTime()을 씁니다. currentTimeMillis()는 벽시계 시각이라서 운영체제가 시계를 맞추면 앞뒤로 움직일 수 있습니다. 반면 공식 문서는 nanoTime()이 경과 시간을 재는 데만 쓸 수 있고 벽시계 시각과는 관계가 없다고 설명합니다. 그 대신 값의 기준점이 정해져 있지 않으므로 같은 JVM 안에서 두 값의 차이를 구할 때만 의미가 있고, 두 값을 비교할 때도 t1 < t0이 아니라 t1 - t0 < 0처럼 차이로 비교해야 합니다. 값이 long 범위를 넘어 부호가 뒤집히더라도 차이는 올바르게 계산되기 때문입니다. submitAll이 deadline - System.nanoTime()으로 남은 시간을 구하는 것도 같은 이유입니다. 한편 Duration.toNanos()는 약 292년을 넘는 기간에서 ArithmeticException을 던지므로, 공개 API라면 받을 수 있는 예산의 범위를 정해 검사하는 것이 좋습니다.

drainTo로 여러 항목을 한 번에 꺼내기

이번에는 꺼내는 쪽을 보겠습니다. 소비자가 원소를 하나씩 take해서 그때마다 데이터베이스에 저장한다면, 원소 수만큼 저장 요청을 보내야 합니다. 저장소는 대개 여러 건을 한 번에 쓰는 편이 효율적이므로, 큐에 쌓인 원소를 묶음으로 꺼내 한꺼번에 저장하고 싶어집니다. drainTo(collection, maxElements)는 지금 큐에 있는 원소를 최대 maxElements개까지 꺼내 주어진 컬렉션으로 옮기고, 옮긴 개수를 돌려줍니다.

그런데 drainTo는 기다리지 않습니다. 큐가 비어 있으면 아무것도 옮기지 않고 바로 0을 돌려주므로, 이것만으로 소비자 반복문을 만들면 빈 큐를 쉬지 않고 확인하게 됩니다. 그래서 첫 원소는 시간 제한 poll로 기다리고, 첫 원소가 오면 나머지를 drainTo로 한 번에 가져오는 방식을 씁니다.

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());
    }
}

nextBatch는 첫 원소를 최대 waitForFirst만큼 기다립니다. 그 안에 아무것도 오지 않으면 poll이 null을 돌려주므로 빈 목록을 반환합니다. 첫 원소를 받았다면 목록에 넣고 drainTo로 최대 maxBatch - 1개를 더 옮깁니다. 이미 하나를 꺼냈으니, 묶음 전체가 maxBatch개를 넘지 않게 하려는 것입니다. 마지막으로 List.copyOf로 수정할 수 없는 복사본을 돌려줍니다.

batch=[1, 2, 3]
remaining=2

main은 용량 10인 LinkedBlockingQueue에 1부터 5까지를 미리 넣고, 묶음 크기 3으로 한 번 꺼냅니다. poll이 1을 바로 꺼내고 drainTo가 2와 3을 더 옮기므로 묶음은 [1, 2, 3]이고, 큐에는 4와 5가 남습니다. 같은 drainer로 nextBatch를 두 번 더 부르면 두 번째 호출에서는 [4, 5]가 바로 나옵니다. 세 번째 호출에서는 큐가 비어 있으므로 1초를 기다린 뒤 빈 목록이 나옵니다.

이 방식에는 장점이 하나 더 있습니다. 큐의 락은 poll과 drainTo가 원소를 옮기는 동안에만 잡히고, 묶음을 저장하는 느린 작업은 nextBatch가 돌아온 뒤에 지역 목록을 가지고 합니다. 그래서 저장하는 동안에도 생산자는 계속 원소를 넣을 수 있습니다. 이 예제에는 저장 작업이 없지만, 실제로 만들 때도 저장은 꺼내기가 끝난 다음에 해야 합니다.

주의할 점은 꺼낸 원소가 이미 큐에서 사라졌다는 것입니다. 묶음을 저장하다 실패해도 큐는 원소를 되돌려 주지 않습니다. 실패한 원소를 다시 put하면 큐의 맨 뒤로 들어가 순서가 바뀌고, 묶음의 일부만 저장된 상태였다면 같은 원소를 두 번 처리할 수도 있습니다. 다시 시도해야 하는 데이터라면 실패한 묶음을 파일이나 데이터베이스처럼 내구성 있는 곳에 기록하거나 재처리용 큐로 따로 보내는 방법을 미리 정해 두어야 합니다.

더 깊이: drainTo가 poll 반복보다 가벼운 이유

poll을 n번 부르면 큐의 락을 n번 잡았다 놓습니다. OpenJDK 21의 ArrayBlockingQueue.drainTo는 락을 한 번만 잡은 상태에서 원소를 모두 옮기고, 그만큼 생긴 빈칸에 맞춰 기다리던 생산자를 깨웁니다. LinkedBlockingQueue는 꺼내는 쪽 락인 takeLock만 잡으므로, 옮기는 동안에도 생산자는 putLock으로 원소를 넣을 수 있습니다. 한편 공식 문서는 대상 컬렉션에 원소를 넣다가 예외가 나면, 원소가 큐와 컬렉션 어디에도 없거나 한쪽에만 있거나 양쪽 모두에 있을 수 있다고 경고합니다. 또 큐를 자기 자신에게 drainTo하면 IllegalArgumentException이 발생합니다.

호출하는 곳에 따라 메서드 고르기

지금까지 본 내용을 모으면, 메서드는 큐마다가 아니라 큐를 부르는 곳마다 고른다는 결론이 나옵니다. 같은 큐라도 웹 요청 스레드는 기다리지 않고 넣고, 내부 작업자 스레드는 기다려서라도 넣는 식으로 정책이 다를 수 있습니다. 대표적인 경우를 표로 정리하면 다음과 같습니다.

put이 언제나 가장 안전한 것도, offer가 언제나 가장 빠른 답인 것도 아닙니다. 반드시 전달해야 할 이벤트를 offer로 넣고 실패를 무시하면 데이터를 잃습니다. 반대로 버려도 되는 통계를 put으로 넣으면, 큐가 막혔을 때 사용자 요청까지 함께 멈춥니다. 데이터가 얼마나 중요한지와 호출한 쪽이 얼마나 기다릴 수 있는지를 함께 보고 고르되, 어떤 메서드를 고르든 반환값과 예외를 무시하지 않는 것이 출발점입니다.

정리

• 큐가 가득 찼거나 비었을 때 add·remove는 예외를 던지고, offer·poll은 false·null을 돌려주고, put·take는 될 때까지 기다리고, 시간 제한 offer·poll은 정해진 시간까지만 기다립니다. 기다리는 메서드는 인터럽트되면 InterruptedException을 던집니다.
• remainingCapacity()로 확인한 뒤 add하면 그 사이에 다른 스레드가 빈칸을 차지할 수 있습니다. 확인과 삽입을 한 번에 하는 offer의 반환값으로 성공 여부를 판단합니다.
• 여러 항목을 시간 제한으로 넣을 때는 마감 시각을 한 번 정하고 남은 시간만 넘깁니다. 이미 들어간 항목은 되돌아오지 않으므로 거절된 나머지의 처리 방법을 정해야 합니다.
• drainTo는 기다리지 않으므로 첫 원소는 시간 제한 poll로 기다린 뒤 나머지를 한 번에 옮깁니다. 꺼낸 뒤의 처리 실패는 큐가 복구해 주지 않습니다.
• BlockingQueue는 스레드 사이의 전달을 안전하게 만들어 주지만, 꺼낸 뒤의 처리를 잃지 않게 하는 일까지 대신하지는 않습니다.

연습 문제

1. 본문의 DeadlineBatchSubmitter에 예산으로 Duration.ZERO를 주면, 용량 2인 빈 큐에 one, two, three를 넣을 때 결과가 어떻게 될까요? 이 동작이 왜 아쉬운지 설명하고, 예산이 0이면 '기다리지는 않되 빈칸이 있으면 넣는다'는 뜻이 되도록 고치세요.
2. 결제가 끝나면 웹 요청 스레드가 '결제 완료' 이벤트를 알림 발송 큐에 넣습니다. 요청은 200밀리초 안에 응답해야 하고, 이 이벤트는 잃어버리면 안 됩니다. add, offer, put, 시간 제한 offer 가운데 무엇을 쓰고, 실패하면 어떻게 할지 정하세요.
3. 첫 번째 큐에서 문자열을 꺼내 길이를 계산한 뒤 두 번째 큐에 넣는 변환기를 작성하세요. 입력의 take와 출력의 put 모두 인터럽트에 반응해야 합니다. 출력 큐가 가득 차서 기다리는 동안 취소되면, 이미 꺼낸 입력을 잃었다는 사실을 실패 기록에 남기고 스레드의 인터럽트 상태를 되살립니다.

for (int index = 0; index < items.size(); index++) {
    E item = items.get(index);
    long remaining = Math.max(0, deadline - System.nanoTime());
    if (!queue.offer(item, remaining, TimeUnit.NANOSECONDS)) {
        rejected.addAll(items.subList(index, items.size()));
        break;
    }
    accepted++;
}

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());
    }
}

failures=1
interrupted=true

다음 글

다음 장에서는 volatile 변수에 ++를 하면 왜 값을 잃는지 확인하고, 락 없이 값 하나를 안전하게 바꾸는 원자 연산을 다룹니다. 이 장의 락과 조건 대기가 여러 단계의 상태 변화를 묶는 도구였다면, AtomicInteger 같은 원자 변수는 카운터처럼 값 하나만 바꿀 때 더 가볍게 쓸 수 있는 도구입니다.