컨슈머 스레드(Consumer Thread)와 프로듀서 스레드(Producer Thread)

 

Contents

---

 

 

 

컨슈머 스레드(Consumer Thread)와 프로듀서 스레드(Producer Thread)란

 

멀티스레딩 환경에서 자주 사용되는 개념입니다. 이 개념들은 주로 생산자-소비자 문제(Producer-Consumer Problem)라는 고전적인 동시성 문제를 해결하는 데 사용됩니다.

이유와 필요성

자원 효율적인 사용:

프로듀서 스레드와 컨슈머 스레드를 통해 작업을 분리함으로써 자원을 보다 효율적으로 사용할 수 있습니다. 예를 들어, 데이터 생성과 데이터 처리를 동시에 수행함으로써 시스템의 자원 활용도를 높일 수 있습니다.

병렬 처리:

멀티스레딩을 통해 여러 작업을 동시에 수행할 수 있습니다. 프로듀서 스레드가 데이터를 생성하는 동안 컨슈머 스레드는 데이터를 처리할 수 있어 전체 시스템의 처리량이 증가합니다.

대기 시간 감소:

프로듀서가 데이터를 생성할 때 컨슈머가 대기하지 않고 바로 데이터를 처리할 수 있어 전체 대기 시간이 줄어듭니다. 이는 특히 실시간 시스템에서 중요한 요소입니다.

응답성 향상:

사용자와 상호작용하는 애플리케이션에서는 백그라운드 작업을 별도의 스레드에서 처리함으로써 메인 스레드의 응답성을 높일 수 있습니다.

실생활 예제

웹 서버:

• 프로듀서: 클라이언트의 요청을 수신하는 스레드.
• 컨슈머: 요청을 처리하고 응답을 생성하는 스레드.
• 웹 서버는 클라이언트의 HTTP 요청을 빠르게 수신하여 대기열에 넣고, 별도의 스레드가 이 요청들을 처리하여 응답을 반환합니다.

동영상 스트리밍:

• 프로듀서: 서버로부터 데이터를 다운로드하는 스레드.
• 컨슈머: 다운로드한 데이터를 디코딩하고 화면에 출력하는 스레드.
• 스트리밍 애플리케이션은 동영상을 끊김 없이 재생하기 위해 버퍼를 사용합니다. 프로듀서 스레드는 네트워크를 통해 데이터를 지속적으로 받아오고, 컨슈머 스레드는 받아온 데이터를 처리하여 사용자에게 끊김 없는 재생을 제공합니다.

로그 처리 시스템:

• 프로듀서: 애플리케이션의 로그를 생성하는 여러 스레드.
• 컨슈머: 로그를 저장하거나 외부 시스템으로 전송하는 스레드.
• 애플리케이션에서 로그가 발생하면 이를 즉시 파일에 기록하거나 외부 서버로 전송하면 성능에 영향을 줄 수 있습니다. 따라서 로그를 생성하는 스레드와 이를 처리하는 스레드를 분리하여 성능을 최적화합니다.

 

예제 코드

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

public class ProducerConsumerExample {

    // 공유 큐 (BlockingQueue)
    private static final int QUEUE_CAPACITY = 10;
    private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(QUEUE_CAPACITY);

    public static void main(String[] args) {
        Thread producerThread = new Thread(new Producer());
        Thread consumerThread = new Thread(new Consumer());

        producerThread.start();
        consumerThread.start();
    }

    // 프로듀서 클래스
    static class Producer implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    int item = (int) (Math.random() * 100);
                    queue.put(item);
                    System.out.println("Produced: " + item);
                    Thread.sleep((int) (Math.random() * 1000)); // 임의의 시간 대기
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }

    // 컨슈머 클래스
    static class Consumer implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    int item = queue.take();
                    System.out.println("Consumed: " + item);
                    Thread.sleep((int) (Math.random() * 1000)); // 임의의 시간 대기
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

 

1. BlockingQueue 사용: ArrayBlockingQueue는 큐가 가득 차면 put 메서드가 블로킹되고, 큐가 비어 있으면 take 메서드가 블로킹됩니다. 이로써 스레드 간의 동기화 문제를 쉽게 해결할 수 있습니다.
2. Producer 클래스:
   • run 메서드에서 무작위로 정수를 생성하고 이를 큐에 넣습니다.
   • put 메서드는 큐가 가득 차 있을 경우 블로킹되므로, 데이터가 안전하게 추가됩니다.
3. Consumer 클래스:
   • run 메서드에서 큐에서 데이터를 꺼내 처리합니다.
   • take 메서드는 큐가 비어 있을 경우 블로킹되므로, 안전하게 데이터를 가져올 수 있습니다.
4. 스레드 생성 및 시작:
   • Producer와 Consumer 각각의 인스턴스를 생성하여 스레드로 실행합니다.