[Java] Producer - Consumer 패턴 구조

[Java] Producer - Consumer 패턴 구조

출처 : https://t1.daumcdn.net/cfile/tistory/24111644521D8CBC35

 

 

Single Thread 상에서 Producer - Consumer

class Producer {
    int id = 0;
    int produce() {
        return nextId();
    }

    int nextId() {
        return id = id + 1;
    }
}

class Consumer {
    void consume(int id) {
        print("ID: " + id);
    }
}

void test() {
    Producer p = new Producer();
    Consumer c = new Consumer();

    result = p.produce();
    c.consume(result);
}

1씩 증가시키는 '생산'을 하는 Producer 클래스가 있고, 그것을 전달받아 사용하는 Consumer가 있다. 즉, 생산하는 역할 따로, 소비하는 역할 따로 있으면 Producer-Consumer 패턴이다.

Single Thread에서의 특징은 '호출하면 바로 반응한다' 이다. Produce를 호출하면 다른 짓을 하지 않고 바로 새로운 id를 리턴해준다. 실시간 반응 (실시간 리턴은 아님)이다. 다만 produce에서 오래 작업을 한다면 Consumer 객체는 오래 기다려야한다.

 

Multi Thread 상에서 Producer-Consumer

1. Producer와 Consumer 가 각각 쓰레드를 가지고 있다. ⇒ 내부에 loop를 가지고 있다.
2. 전달 매개체 (보통 큐로 구현) 가 생겼다. ⇒ 아래 예시에서 Table 클래스

public class Main {
    public static void main(String[] args) {
        Table table = new Table(100);
        new ProducerThread(table).start();
        new ConsumerThread(table).start();
    }
}

public class ProducerThread extends Thread {
    private static int id = 0;
    Table table;

    public ProducerThread(Table table) {
        this.table = table;
    }

    public void run() {
        while(true) {
            Thread.sleep(1000);
            String packet = "NO : " + nextId();
            table.put(packet); // 큐에 추가
        }
    }

    private static synchronized int nextId() {
        return id++;
    }
}

public class ConsumerThread extends Thread {
    private final Table table;
    public ConsumerThread(Table table) {
        this.table = table;
    }

    public void run() {
        while(true) {
            String packet = table.take(); // 큐에서 가져옴
            System.out.println("consumer : " + packet);
        }
    }
}

public class Table {
    private final String[] buffer;
    private int tail;
    private int head;
    private int count;

    public Table(int count) {
        this.buffer = new String[count];
        this.head = 0;
        this.tail = 0;
        this.count = 0;
    }

    public synchronized void put(String packet) {
        while(count >= buffer.length) { //buffer가 가득차면 대기
            wait();
        }

        buffer[tail] = packet; // 후입하라!
        tail = (tail + 1) & buffer.length; // Cicular 큐라서 tail의 위치가 바뀜
        count++;
        notifyAll(); // 버퍼에 무엇인가가 들어 갔으니 take 해도 된다고 이벤트 날림
    }

    public synchronzized String take() {
        while(count <= 0) { //buffer에 아무것도 없으면 대기!
            wait();
        }

        String packet = buffer[head]; // 선출하라!
        head = (head + 1) % buffer.length; // 원형큐에 맞게 header의 위치를 바꿔줌
        count --;
        notifyAll(); // 버퍼에서 하나를 가져갔으니 put 해도 된다고 이벤트 날림
        return packet;
    }
}

• Producer : 1초에 한번씩 table에 id를 집어 넣는다.
• Consumer : table에서 id를 가져와서 출력한다.
• Table
  • String 배열을 가지고 원형큐 구현
  • Producer 와 Consumer 에서 해당 배열에 동시 접근을 막기 위한 장치 존재 ⇒ synchronized, wait()
  • synchronized, wait() 을 직접 조작하는 것은 사실 위험하고 어렵기 때문에 Actor 패턴 이 등장한다.

Actor

• 능동적으로 비동기 message를 처리한다.
  • 능동적 = '쓰레드 하나가 할당되어 있다'
  • 비동기 message를 처리한다 = 'message를 담아둘 수 있는 자신만의 큐를 가지고 있다'
• 즉, 객체 + 루프를 가진 쓰레드 + 큐 로 이루어진 것이다.

---

Reference

• https://hamait.tistory.com/550