ArrayBlockingQueue로 생산자·소비자 만들기

두 Condition으로 대기실을 나눈 글에서는 ReentrantLock 하나와 notFull·notEmpty 두 Condition으로 크기가 정해진 큐를 직접 만들었습니다. 큐가 가득 차면 생산자를 재우고, 비면 소비자를 재우고, 상태가 바뀌면 반대편을 깨우는 코드였습니다. 여기에 시간 제한 대기나 즉시 실패하는 삽입, 종료 처리까지 붙이면 코드는 금방 길어집니다. 게다가 신호 하나만 빠뜨려도 스레드가 영원히 잠듭니다.

자바 표준 라이브러리에는 이런 큐가 이미 있습니다. java.util.concurrent 패키지의 BlockingQueue 인터페이스는 큐가 가득 찼거나 비었을 때 기다릴 수 있는 큐를 정의하고, ArrayBlockingQueue는 그 구현 가운데 배열을 쓰며 크기가 고정된 것입니다. 실무에서는 앞에서 만든 큐 대신 이런 표준 구현을 씁니다. 그렇다고 직접 만들어 본 일이 헛수고는 아닙니다. 이 클래스가 안에서 무엇을 하는지 알아야 메서드를 정확히 고르고, 문제가 생겼을 때 원인을 짐작할 수 있습니다.

다만 큐를 바꿔도 라이브러리가 대신 정해 주지 않는 것이 있습니다. 기다리던 생산자가 인터럽트되면 어떻게 할지, 소비자에게 일이 끝났다는 것을 어떻게 알릴지, 큐를 몇 칸으로 만들지는 여전히 프로그램을 만드는 쪽이 정해야 합니다. 이 글은 먼저 두 큐를 비교한 다음 이 세 가지를 차례로 다룹니다.

직접 만든 큐와 ArrayBlockingQueue 비교하기

ArrayBlockingQueue의 내부 구조는 앞에서 만든 큐와 거의 같습니다. OpenJDK 21의 구현을 열어 보면 원소를 담는 배열 하나, ReentrantLock 하나, 그리고 이름까지 같은 notEmpty와 notFull 두 Condition을 필드로 가지고 있습니다. put은 큐가 가득 찬 동안 notFull에서 기다렸다가 원소를 넣은 뒤 notEmpty에 신호를 보내고, take는 정확히 그 반대로 움직입니다. 우리가 손으로 작성했던 while 재검사와 신호 방향을 이미 검증된 코드가 맡는 셈입니다.

생성자에는 용량을 넘깁니다. new ArrayBlockingQueue<>(100)은 100칸짜리 큐를 만들고, 1보다 작은 용량을 주면 IllegalArgumentException이 발생합니다. 이 크기는 만든 뒤에 바꿀 수 없는데, 그 점이 오히려 장점입니다. 생산자가 아무리 빨라도 큐에 쌓이는 원소는 100개를 넘지 않으므로 메모리 사용량에 상한이 생깁니다. 또 큐가 가득 차면 생산자가 put에서 기다리므로 생산 속도가 저절로 소비 속도에 맞춰지는데, 이렇게 뒤쪽 단계가 느릴 때 앞쪽 단계를 늦추는 효과를 역압(backpressure)이라고 합니다.

생성자의 두 번째 인자로는 공정성(fairness)을 정할 수 있습니다. new ArrayBlockingQueue<>(100, true)처럼 true를 주면 put이나 take에서 기다리는 스레드들이 먼저 기다린 순서대로 큐에 접근하고, 기본값인 false에서는 그 순서가 정해져 있지 않습니다. 내부의 ReentrantLock을 공정 모드로 만드는 것이므로 ReentrantLock의 공정성을 다룬 글에서 본 것과 같은 개념입니다. 공식 문서에 따르면 공정 모드는 대개 처리량을 떨어뜨리는 대신, 대기 시간의 편차를 줄이고 기아(starvation)를 막아 줍니다. 기아란 특정 스레드가 계속 순서를 빼앗겨 진행하지 못하는 현상입니다. 그래서 보통은 기본값으로 시작하고, 일부 스레드가 오래 밀려나는 문제가 실제로 확인될 때 공정 모드를 검토합니다.

BlockingQueue에는 null을 넣을 수 없다는 점도 알아 둡니다. put(null)이나 offer(null)은 NullPointerException을 던집니다. 큐가 비었을 때 poll()이 null을 돌려주어 '꺼낼 원소가 없다'는 뜻을 나타내므로, 원소로 null을 허용하면 두 경우를 구별할 수 없기 때문입니다. 이 규칙은 뒤에서 종료 메시지를 설계할 때 다시 나옵니다.

put과 take 말고도 바로 결과를 돌려주는 offer와 poll, 정해진 시간만 기다리는 시간 제한 버전, 예외를 던지는 add와 remove가 있습니다. 이 메서드들은 다음 글에서 자세히 비교하고, 이 글에서는 주로 기다리는 put과 take를 씁니다.

InterruptedException을 삼키면 작업이 사라진다

큐를 표준 구현으로 바꿨다고 모든 문제가 사라지지는 않습니다. 흔한 실수는 put이 던지는 InterruptedException을 중간에서 조용히 처리해 버리는 것입니다. InterruptedException은 검사 예외(checked exception)라서 put을 부를 때마다 throws를 붙이거나 try로 감싸야 합니다. 이것이 번거로워서 다음처럼 큐를 감싸는 클래스를 만들고, 그 안에서 예외를 삼키는 경우가 많습니다.

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

public final class SwallowingQueueFacade {
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);

    void submit(String value) {
        try {
            queue.put(value);
        } catch (InterruptedException e) {
            System.err.println("ignored cancellation: " + value);
        }
    }

    public static void main(String[] args) throws InterruptedException {
        SwallowingQueueFacade facade = new SwallowingQueueFacade();
        facade.submit("occupy-slot");
        Thread producer = Thread.ofPlatform().start(() -> facade.submit("lost-task"));
        Thread.sleep(30);
        producer.interrupt();
        producer.join();
        System.out.println("queued=" + facade.queue.size());
        System.out.println("producerInterrupted=" + producer.isInterrupted());
    }
}

main은 먼저 occupy-slot을 넣어 한 칸짜리 큐를 가득 채웁니다. 그다음 생산자 스레드가 lost-task를 넣으려 하면 빈칸이 없으므로 put 안에서 기다립니다. 30밀리초 뒤 main이 이 생산자를 인터럽트하고, join으로 생산자가 끝나기를 기다린 다음 큐에 든 원소 수와 생산자의 인터럽트 상태를 출력합니다.

ignored cancellation: lost-task
queued=1
producerInterrupted=false

첫 줄은 catch 블록이 표준 오류(stderr)로 출력한 메시지이고, 나머지 두 줄은 main이 표준 출력으로 출력한 것입니다. 여러 번 실행해도 결과는 같습니다. 아무도 원소를 꺼내지 않으니 put은 성공할 수 없고, 인터럽트가 put에 들어가기 전에 도착하든 기다리는 도중에 도착하든 put은 곧바로 InterruptedException을 던지기 때문입니다.

queued=1은 큐에 처음 넣은 occupy-slot만 들어 있다는 뜻입니다. lost-task는 들어가지 못했는데도 submit은 아무 일 없었다는 듯이 정상적으로 반환했습니다. 반환 타입이 void이니 호출한 쪽은 제출이 성공했는지 알 수 없고, 대개는 성공했다고 믿고 다음 일로 넘어갑니다. 작업 하나가 아무도 모르게 사라진 것입니다. stderr의 메시지는 누군가 로그를 뒤져야 눈에 띌 뿐, 호출한 코드에 실패를 알려 주지는 못합니다.

producerInterrupted=false는 두 번째 문제를 보여 줍니다. put처럼 기다리다가 InterruptedException을 던지는 메서드는 예외를 던지면서 스레드의 인터럽트 상태를 지웁니다. 예외를 잡은 쪽이 상태를 다시 설정하지 않으면, 이 스레드에 취소 요청이 왔었다는 사실 자체가 사라집니다. 그러면 이 스레드에서 이어서 실행되는 코드나 바깥의 작업 반복문은 취소를 알아채지 못하고 하던 일을 계속합니다.

고치는 방법은 두 가지입니다. 가장 간단한 방법은 예외를 잡지 않고 submit에 throws InterruptedException을 붙여 호출한 쪽에 그대로 넘기는 것입니다. 그런데 그렇게 하면 submit은 queue.put을 이름만 바꿔 부르는 메서드가 되므로, 이런 감싸는 클래스를 따로 둘 이유가 없어집니다. 그럴 때는 BlockingQueue를 직접 쓰면 됩니다.

검사 예외를 던질 수 없는 자리도 있습니다. 예를 들어 Runnable의 run() 메서드에는 throws InterruptedException을 붙일 수 없습니다. 이때는 인터럽트 상태를 다시 설정하고, 실패했다는 사실을 반환값으로 알려야 합니다. 다음 코드는 submit이 성공 여부를 boolean으로 돌려주도록 고친 것입니다. 람다 안에서는 바깥 지역 변수에 값을 대입할 수 없으므로, 생산자가 받은 반환값은 원소 하나짜리 배열 accepted에 담아 main으로 전달합니다.

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

public final class ReportingQueueFacade {
    private final BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);

    boolean submit(String value) {
        try {
            queue.put(value);
            return true;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ReportingQueueFacade facade = new ReportingQueueFacade();
        facade.submit("occupy-slot");
        boolean[] accepted = new boolean[1];
        Thread producer = Thread.ofPlatform().start(() -> accepted[0] = facade.submit("lost-task"));
        Thread.sleep(30);
        producer.interrupt();
        producer.join();
        System.out.println("queued=" + facade.queue.size());
        System.out.println("accepted=" + accepted[0]);
        System.out.println("producerInterrupted=" + producer.isInterrupted());
    }
}

queued=1
accepted=false
producerInterrupted=true

이번에는 생산자가 submit의 반환값 false를 받아 제출이 실패했다는 것을 압니다. catch에서 Thread.currentThread().interrupt()로 상태를 되살렸으므로 producerInterrupted도 true입니다. 이 스레드의 바깥쪽 코드는 반환값을 보고 작업을 다른 곳에 기록하거나 다시 시도할 수 있고, 인터럽트 상태를 보고 하던 일을 정리한 뒤 끝낼 수 있습니다. main이 accepted[0]을 안전하게 읽을 수 있는 것은 join이 끝난 스레드가 한 일을 볼 수 있게 해 주기 때문입니다. 인터럽트 상태를 확인하고 취소를 처리하는 일반적인 방법은 interrupt 플래그와 취소를 다룬 글에서 설명했습니다.

더 깊이: 끝난 스레드의 인터럽트 상태

두 예제는 생산자가 끝난 뒤, 즉 join 다음에 isInterrupted()를 읽습니다. 그런데 Thread.interrupt()의 명세는 살아 있지 않은 스레드에 대해서는 아무 효과가 없어도 된다고만 적고 있어서, 끝난 스레드의 인터럽트 상태가 남는지는 JDK 구현에 달려 있습니다. OpenJDK는 JDK 14에서 인터럽트 상태를 JVM 내부 자료구조 대신 Thread 객체의 필드에 저장하도록 바꿨고(JDK-8229516), 그 뒤로는 끝난 스레드도 인터럽트 상태를 그대로 보고합니다. JDK 21 문서에도 이 동작이 구현 설명(Implementation Note)으로 적혀 있습니다. 그 전의 JDK에서는 스레드가 끝나면 isInterrupted()가 언제나 false였습니다. 이처럼 구현에 기대는 동작이므로, 실제 코드에서 취소 여부를 알아야 한다면 반환값이나 결과 객체에 담아 명시적으로 전달하는 편이 확실합니다.

종료 메시지로 소비자 멈추기

앞에서 만든 큐에는 close()가 있었지만 BlockingQueue에는 큐를 닫는 기능이 없습니다. 공식 문서는 그 대신 생산자가 '데이터가 끝났다'는 뜻의 특별한 원소를 큐에 넣는 방법을 흔히 쓴다고 안내합니다. 이런 원소를 포이즌 필(poison pill)이라고도 부릅니다. 소비자는 이 원소를 꺼내면 반복을 멈추고 끝납니다.

종료 메시지로 가장 먼저 떠오르는 것은 "STOP" 같은 문자열이지만, 실제 데이터에 같은 문자열이 들어오면 소비자가 엉뚱한 때에 멈춥니다. null은 앞에서 본 것처럼 넣을 수조차 없습니다. 그래서 다음 예제는 데이터와 종료를 타입으로 구별합니다. Message 인터페이스 아래에 값을 나르는 Work 레코드와 종료를 뜻하는 Stop 열거형을 두면, 어떤 값을 가진 Work도 종료 메시지로 오해받을 수 없습니다. Stop은 상수가 INSTANCE 하나뿐인 열거형이라 ==로 비교해도 안전합니다.

프로그램은 소비자 스레드 세 개를 먼저 띄웁니다. 그다음 main이 1부터 100까지의 Work를 넣고, 이어서 Stop을 소비자 수만큼 넣습니다. 각 소비자는 꺼낸 값을 지역 변수 local에 더하다가 Stop을 만나면 그 합을 partial 배열의 자기 칸에 적고 끝납니다. main은 세 소비자를 모두 join한 뒤 부분합과 그 합계, 큐에 남은 원소 수를 출력합니다.

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public final class BlockingQueuePipeline {
    interface Message {}
    record Work(int value) implements Message {}
    enum Stop implements Message { INSTANCE }

    public static void main(String[] args) throws InterruptedException {
        int consumerCount = 3;
        BlockingQueue<Message> queue = new ArrayBlockingQueue<>(8);
        long[] partial = new long[consumerCount];
        List<Thread> consumers = new ArrayList<>();

        for (int id = 0; id < consumerCount; id++) {
            int slot = id;
            consumers.add(Thread.ofPlatform().name("consumer-" + id).start(() -> {
                try {
                    long local = 0;
                    while (true) {
                        Message message = queue.take();
                        if (message == Stop.INSTANCE) {
                            partial[slot] = local;
                            return;
                        }
                        local += ((Work) message).value();
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            }));
        }

        for (int value = 1; value <= 100; value++) {
            queue.put(new Work(value));
        }
        for (int i = 0; i < consumerCount; i++) {
            queue.put(Stop.INSTANCE);
        }
        for (Thread consumer : consumers) {
            consumer.join();
        }
        long sum = 0;
        for (long value : partial) sum += value;
        System.out.println("partial=" + Arrays.toString(partial));
        System.out.println("sum=" + sum);
        System.out.println("remaining=" + queue.size());
    }
}

partial=[775, 2721, 1554]
sum=5050
remaining=0

partial 줄은 실행할 때마다 달라질 수 있습니다. 같은 프로그램을 여러 번 실행하자 [238, 1557, 3255], [439, 4611, 0]처럼 매번 다른 값이 나왔고, 마지막 예처럼 한 소비자가 일을 하나도 받지 못한 적도 있습니다. 큐는 넣은 순서대로 원소를 내주지만, 세 소비자 가운데 누가 다음 원소를 가져갈지는 스레드 스케줄에 달려 있기 때문입니다. 반면 sum=5050과 remaining=0은 언제나 같습니다.

합계가 항상 맞는 이유는 넣는 순서에 있습니다. 생산자가 하나뿐이고 Work를 모두 넣은 다음에야 Stop을 넣으므로, 큐에서는 모든 Work가 어느 Stop보다도 먼저 나옵니다. 그래서 세 Stop이 모두 꺼내졌다면 그 앞에 있던 Work 100개도 이미 누군가 꺼낸 것입니다. 또 소비자는 꺼낸 원소를 처리한 뒤에야 다음 원소를 꺼내므로, 각 소비자는 자기 Stop을 꺼내기 전에 앞서 가져간 Work를 모두 더했습니다. 따라서 세 소비자가 모두 끝나면 합계는 언제나 5050이고 큐도 비어 있습니다.

도표에서 보듯 이 프로그램에서 '끝났다'는 말은 세 시점에서 각각 다른 뜻입니다. main이 Work를 다 넣은 시점에는 제출만 끝났을 뿐, 소비자들은 아직 더하는 중일 수 있습니다. 소비자 하나가 Stop을 꺼낸 시점에는 그 소비자만 끝났고, 다른 소비자의 처리가 끝났다는 뜻은 아닙니다. 전체 합계는 세 번의 join이 모두 돌아온 뒤에만 읽을 수 있습니다. join과 완료 가시성을 다룬 글에서 보았듯이 join은 끝난 스레드가 쓴 값을 main이 볼 수 있게 해 주므로, partial 배열을 따로 동기화할 필요도 없습니다.

Stop을 소비자 수만큼 넣는 이유는 Stop 하나가 소비자 하나만 멈추기 때문입니다. take로 꺼낸 원소는 큐에서 사라지므로, 한 소비자가 가져간 Stop을 다른 소비자가 다시 볼 수는 없습니다. 실제로 Stop을 하나 더 넣으면 아무도 꺼내지 않은 Stop이 남아 remaining=1이 출력되고, 하나 덜 넣으면 프로그램이 끝나지 않습니다. 왜 끝나지 않는지는 연습 문제 1에서 따져 봅니다.

생산자가 여러 명이면 한 가지를 더 조심해야 합니다. 생산자마다 자기 일을 마치고 각자 Stop을 넣으면, 한 생산자의 Stop 뒤에 아직 일하는 다른 생산자의 Work가 들어갈 수 있습니다. 소비자들이 그 Stop을 먼저 꺼내 모두 끝나 버리면 뒤의 Work는 아무도 처리하지 않습니다. 그래서 Stop은 한 곳에서만 넣어야 합니다. 예를 들어 main이 모든 생산자 스레드를 join해 제출이 끝났음을 확인한 다음 소비자 수만큼 Stop을 넣으면 됩니다.

인터럽트 처리도 살펴봅시다. main은 throws InterruptedException을 선언했으므로 put이나 join에서 인터럽트되면 예외가 그대로 main 밖으로 전달됩니다. 소비자는 take에서 인터럽트되면 catch에서 인터럽트 상태를 되살린 뒤 람다를 빠져나가 끝납니다. 앞 절의 잘못된 예와 달리 취소 요청을 받은 스레드가 실제로 멈추고, 그 사실도 상태에 남습니다. 다만 이렇게 끝난 소비자는 partial에 자기 합을 적지 않으므로, 실제 서비스라면 정상 종료와 취소를 구별해 기록해야 합니다. 이 예제는 소비자를 인터럽트하지 않으므로 이 경로를 실행하지 않습니다.

더 깊이: 큐를 거친 객체가 안전하게 보이는 이유

소비자는 생산자 스레드가 만든 Work 객체를 따로 동기화하지 않고 읽습니다. 이것이 안전한 이유는 BlockingQueue의 메모리 일관성 규칙에 있습니다. 공식 문서에 따르면 한 스레드가 원소를 큐에 넣기 전에 한 동작은, 다른 스레드가 그 원소를 큐에서 꺼낸 뒤에 하는 동작보다 먼저 일어난(happens-before) 것으로 취급됩니다. ArrayBlockingQueue에서는 생산자와 소비자가 같은 내부 락을 잡았다 놓는 과정이 이 관계를 만듭니다. 하지만 큐에 넣은 뒤에도 생산자가 그 객체를 계속 고친다면 이 규칙으로는 보호되지 않으므로, 큐로 주고받는 객체는 Work 레코드처럼 불변으로 만드는 것이 좋습니다.

큐 용량 정하기

ArrayBlockingQueue의 용량은 '메모리가 허락하는 만큼 크게' 정하면 되는 값이 아닙니다. 큐가 크면 순간적인 폭주를 더 많이 받아 줄 수 있지만, 그만큼 뒤쪽에 들어간 작업이 오래 기다립니다. 큐의 맨 끝에 들어간 작업은 앞의 작업이 모두 처리될 때까지 기다려야 하므로, 소비자가 초당 200건을 처리하는데 큐에 400건이 차 있다면 마지막 작업은 약 2초를 기다립니다. 거꾸로 말하면 큐에서 기다리는 시간을 2초 안으로 묶고 싶을 때 용량의 출발점은 처리량 × 허용 대기 시간, 즉 200 × 2 = 400건입니다.

다음 코드는 이 계산에 순간적인 폭주를 받아 줄 여유 계수(burst factor)를 곱해 용량을 추정합니다.

import java.time.Duration;

public final class QueueCapacityEstimator {
    static int estimate(double consumedPerSecond, Duration allowedQueueDelay, double burstFactor) {
        if (!(consumedPerSecond > 0)) throw new IllegalArgumentException("consumedPerSecond");
        if (allowedQueueDelay.isNegative() || allowedQueueDelay.isZero()) {
            throw new IllegalArgumentException("allowedQueueDelay");
        }
        if (!(burstFactor >= 1.0)) throw new IllegalArgumentException("burstFactor");
        double seconds = allowedQueueDelay.toNanos() / 1_000_000_000.0;
        long estimated = (long) Math.ceil(consumedPerSecond * seconds * burstFactor);
        if (estimated > Integer.MAX_VALUE) throw new IllegalArgumentException("capacity overflow");
        return Math.toIntExact(estimated);
    }

    public static void main(String[] args) {
        int capacity = estimate(200.0, Duration.ofSeconds(2), 1.25);
        System.out.println("capacity=" + capacity);
    }
}

capacity=500

200건/초 × 2초 = 400건에 여유 계수 1.25를 곱해 500건이 나왔습니다. 여기서 여유 계수가 무엇과 무엇을 맞바꾸는지 알아 둬야 합니다. 큐가 500건으로 가득 차면 마지막 작업은 500 ÷ 200 = 2.5초를 기다리므로 2초라는 목표를 넘습니다. 즉 여유 계수는 '폭주가 오면 거절하는 대신 조금 더 기다리게 하겠다'는 선택이고, 그만큼 최악의 대기 시간이 길어집니다.

이 계산은 소비 속도가 일정하다는 가정 위에 있습니다. 실제로는 작업마다 처리 시간이 다르고 도착도 고르지 않으므로 이 값은 출발점일 뿐입니다. 부하 시험에서 작업이 큐에서 기다린 시간의 분포와 거절 비율을 재 보며 조정하고, 운영 중에도 이 두 값을 함께 지켜봅니다. 또 생산 속도가 길게 보아 소비 속도보다 빠르다면 어떤 용량도 결국 찹니다. 큐는 일시적인 폭주를 흡수하는 완충 장치일 뿐이므로, 그때는 생산을 제한하거나 소비자를 늘리거나 작업량 자체를 줄여야 합니다.

입력 검사 코드에도 눈여겨볼 점이 있습니다. 첫 번째 검사를 consumedPerSecond <= 0이 아니라 !(consumedPerSecond > 0)으로 쓴 것은 NaN(Not a Number) 때문입니다. double의 NaN은 <, <=, >, >=로 어떤 값과 비교해도 결과가 false이므로 NaN <= 0도 false가 되어 검사를 통과해 버리지만, !(NaN > 0)은 true가 되어 예외를 던집니다. burstFactor도 같은 이유로 !(burstFactor >= 1.0)으로 검사합니다. burstFactor < 1.0이라고 쓰면 NaN이 통과하고, Math.ceil(NaN)을 long으로 바꾼 값 0이 용량으로 반환됩니다.

확인 문제

소비량이 초당 120건, 허용 대기 시간이 0.5초, 여유 계수가 1.5라면 추정 용량은 얼마일까요? 또 그 용량만큼 가득 찬 큐를 비우는 데는 얼마나 걸릴까요?

상황에 맞는 BlockingQueue 구현 고르기

ArrayBlockingQueue 말고도 BlockingQueue의 구현은 여러 가지입니다. 요구 사항에 따라 다음과 같이 출발점을 고를 수 있습니다.

표에서 눈여겨볼 것은 크기 제한입니다. LinkedBlockingQueue를 용량 없이 만들거나 PriorityBlockingQueue를 쓰면 큐가 가득 차는 일이 사실상 없으므로 앞에서 말한 역압이 생기지 않습니다. 소비자가 느리면 원소가 메모리에 계속 쌓일 뿐입니다. 큐의 크기를 제한하는 것이 목적이라면 용량을 명시하거나 ArrayBlockingQueue를 고릅니다.

큐를 담는 필드나 변수는 BlockingQueue<E> queue = new ArrayBlockingQueue<>(100);처럼 인터페이스 타입으로 선언해 두면, 나중에 구현을 바꾸거나 테스트에서 다른 구현을 넣기 쉽습니다. 큐를 다시 감싸는 클래스는 도메인 타입으로 바꾸기, 지표 수집, 시작과 종료 관리처럼 실제로 하는 일이 있을 때만 둡니다. 앞에서 본 SwallowingQueueFacade처럼 메서드 이름만 바꾸고 예외를 숨기는 계층은 공식 문서에 적힌 동작을 가리고 실수할 여지만 늘립니다.

더 깊이: ArrayBlockingQueue와 LinkedBlockingQueue의 락 구조

두 클래스는 락을 쓰는 방식이 다릅니다. ArrayBlockingQueue는 락 하나로 put과 take를 모두 보호하므로, 생산자와 소비자가 동시에 큐 안의 코드를 실행할 수 없습니다. LinkedBlockingQueue는 넣는 쪽의 putLock과 꺼내는 쪽의 takeLock을 따로 두고 원소 수를 AtomicInteger로 관리하므로, 생산자 하나와 소비자 하나가 동시에 진행할 수 있습니다. 대신 원소를 넣을 때마다 노드 객체를 새로 만들어 가비지 컬렉션 부담이 생기고, ArrayBlockingQueue는 처음에 배열을 한 번에 만들어 두고 다시 씁니다. 어느 쪽이 빠른지는 스레드 수와 작업 크기에 따라 달라지므로, 성능이 중요하다면 실제 부하로 재 보고 고릅니다.

정리

• ArrayBlockingQueue는 앞에서 직접 만든 큐와 같은 구조(배열, 락 하나, notEmpty·notFull)를 검증된 코드로 제공합니다. 용량은 생성할 때 정해지고 바뀌지 않으며, 그 덕분에 메모리 상한과 역압이 생깁니다.
• put의 InterruptedException을 삼키면 작업이 조용히 사라지고 취소 요청도 지워집니다. throws로 넘기거나, 인터럽트 상태를 되살리고 실패를 반환값으로 알립니다.
• BlockingQueue에는 닫는 기능이 없으므로 종료 메시지를 데이터와 다른 타입으로 만들고, 소비자 수만큼 한 곳에서 넣습니다.
• 용량의 출발점은 처리량 × 허용 대기 시간입니다. 여유를 늘리면 최악의 대기 시간도 늘어나고, 생산이 소비보다 계속 빠르면 어떤 용량도 결국 찹니다.

연습 문제

1. 본문의 BlockingQueuePipeline에서 consumerCount만 4로 바꾸고, Stop을 넣는 반복은 i < 3으로 고정하면 어떻게 될지 예측하세요.
2. 감사(audit) 이벤트를 큐에 넣는 submit 메서드를 만드세요. 큐에 빈칸이 생기기를 최대 100밀리초까지만 기다리고, 그 안에 넣지 못하면 스레드를 더 붙잡지 말고 이벤트와 거절 시각을 동시성 컬렉션에 따로 기록합니다. 인터럽트는 거절로 바꾸지 말고 호출한 쪽에 그대로 전달하며, main에서는 성공 건수와 거절 건수를 출력합니다.

import java.time.Duration;
import java.time.Instant;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

public final class TimedAuditSubmissionSolution {
    record AuditEvent(String action) {}
    record Rejected(AuditEvent event, Instant at) {}

    private final BlockingQueue<AuditEvent> accepted = new ArrayBlockingQueue<>(1);
    private final ConcurrentLinkedQueue<Rejected> fallback = new ConcurrentLinkedQueue<>();

    boolean submit(AuditEvent event, Duration timeout) throws InterruptedException {
        boolean stored = accepted.offer(event, timeout.toNanos(), TimeUnit.NANOSECONDS);
        if (!stored) fallback.add(new Rejected(event, Instant.now()));
        return stored;
    }

    public static void main(String[] args) throws InterruptedException {
        TimedAuditSubmissionSolution service = new TimedAuditSubmissionSolution();
        int success = 0;
        if (service.submit(new AuditEvent("login"), Duration.ofMillis(100))) success++;
        if (service.submit(new AuditEvent("logout"), Duration.ofMillis(1))) success++;
        System.out.println("success=" + success);
        System.out.println("fallback=" + service.fallback.size());
    }
}

success=1
fallback=1

다음 글

다음 글에서는 put·take와 함께 add·offer·remove·poll과 시간 제한 버전을 비교하고, 큐가 가득 찼거나 비었을 때 어떤 메서드를 골라야 하는지 살펴봅니다.