생산자(Producer) / 소비자(Consumer) 패턴

Java 를 사용하다보면 multi Thread 를 고려하게 되고,

자연스레 트랜잭션이나 동시성 이슈를 고려하게 된다.

(해당 내용에 대해서는 아래 참고 포스팅 참고 바람)

 

이번 포스팅에서는 
자바에서 생산자/소비자 패턴(Producer/Consumer Pattern)에 대해 설명을 할 것이다.

 

소개

 

멀티스레딩 환경에서 자주 사용되는 디자인 패턴으로, 데이터를 생산하는 생산자 스레드와 소비하는 소비자 스레드 간의 작업을 효율적으로 분배하는 데 유용한 패턴이다. 이 패턴을 통해 작업을 동시에 처리하면서도 공유 자원의 동기화를 통해 데이터의 일관성을 유지할 수 있다. 

 

개요

• 생산자: 데이터를 생성하여 공유 자원(예: 큐)에 넣는다.
• 소비자: 공유 자원에 있는 데이터를 꺼내서 처리힌다.
• 목적: 생산과 소비를 분리하여 서로 독립적으로 작업할 수 있도록 한다. 특히 데이터가 큐에 차면 생산자는 대기하고, 큐가 비면 소비자는 대기하는 구조이다.
  일반적인 큐와 비슷하게 동작한다.

구현 방식

• BlockingQueue: 자바의 java.util.concurrent 패키지에서 제공하는 BlockingQueue는 생산자-소비자 패턴을 구현하기에 매우 유용하다. 생산자가 큐에 데이터를 넣고, 소비자가 큐에서 데이터를 꺼내는 방식으로 동작한다.
• wait() / notify() 메서드: 고전적인 방법으로, 객체에서 wait()와 notify()를 사용하여 스레드 간의 동기화를 구현할 수 있다.

3. 예제 코드

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; class Producer implements Runnable {     private BlockingQueue<Integer> queue;     public Producer(BlockingQueue<Integer> queue) {         this.queue = queue;     }     @Override     public void run() {         try {             for (int i = 1; i <= 10; i++) {                 queue.put(i);                 System.out.println("Produced: " + i);                 Thread.sleep(100); // 생산 속도 조절             }         } catch (InterruptedException e) {             Thread.currentThread().interrupt();         }     } } class Consumer implements Runnable {     private BlockingQueue<Integer> queue;     public Consumer(BlockingQueue<Integer> queue) {         this.queue = queue;     }     @Override     public void run() {         try {             while (true) {                 int value = queue.take();                 System.out.println("Consumed: " + value);                 Thread.sleep(150); // 소비 속도 조절             }         } catch (InterruptedException e) {             Thread.currentThread().interrupt();         }     } } public class ProducerConsumerExample {     public static void main(String[] args) {         BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(5);         new Thread(new Producer(queue)).start();         new Thread(new Consumer(queue)).start();     } }

장점

• 동기화된 데이터 처리: 생산과 소비가 동기화되어 중복 처리나 데이터 손실을 방지
• 유연성: 생산자와 소비자의 수를 독립적으로 조정할 수 있어 시스템의 성능을 극대화
• 효율적 자원 사용: 큐가 찼을 때 생산자가 대기하고, 큐가 비었을 때 소비자가 대기하여 불필요한 리소스 소모를 방지

활용 사례

• 멀티스레딩 환경에서의 데이터 파이프라인 구성
• 비동기 작업 처리 시스템
• 네트워크 요청 처리 시스템 (예: 메시지 큐 서비스)

---

참고 포스팅

https://thenicesj.tistory.com/1026

ConcurrentHashMap 을 활용한 동시성 제어

https://thenicesj.tistory.com/491

@Transactional 사용시 주의 사항1 (checked Exception)