[Java] BlockingQueue

✅ BlockingQueue

 

 - 큐가 꽉 찼을 때 Object를 삽입하려고 할 때는 공간이 생길 때까지 wait하고,

    큐가 비었을 때 Object를 꺼내려고 할 때는 큐에 새로운 Object가 삽입될 때까지 wait하는 기능을 가지고 있는 큐이다.

 - Thread-safe하다.

 - 처음에는 Producer-Consumer Queue를 구현하기 위해 만들어졌다고 한다.

 

✅ BlockingQueue Methods

 

 - BlockingQueue의 메서드들은 예외가 발생했을 때 어떻게 처리하느냐에 따라 다음과 같이 분류할 수 있다.

 

	Throws Exception	Special value	Blocks	Times out
Insert	add(e)	offer(e)	put(e)	offer(e, time, unit)
Remove	remove()	poll()	take()	poll(time, unit)
Examine	element()	peek()	-	-

 

    - Throws Exception : 예외 발생시 Exception을 던진다

    - Special value : 예외 발생시 특정 값(null이나 false)을 반환한다.

    - Blocks : 예외 발생시 wait한다.

    - Times out : 예외 발생시(작업을 당장 수행할 수 없으면) wait한다. 지정한 time limit 내에 작업을 수행할 수 있게 되면 수행 후 true를 리턴하고, timeout되면 false를 리턴한다.

 

 - 각 메서드에 대한 자세한 설명은 공식 문서 참고

 

✅ Producer-Consumer problem 해결

 

여러 개의 Producer 스레드와, 여러 개의 Consumer 스레드를 실행할 때,

BlockingQueue를 이용하여 생산자-소비자 문제를 쉽게 해결할 수 있다.

 

 

Java

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;

public class BlockingQueueEx {
    public static void main(String[] args) {
    
        BlockingQueue buffer = new LinkedBlockingQueue();
        
        Producer p1 = new Producer(buffer, "p1");
        Producer p2 = new Producer(buffer, "p2");
        Consumer c1 = new Consumer(buffer, "c1");
        Consumer c2 = new Consumer(buffer, "c2");
        new Thread(p1).start();
        new Thread(p2).start();
        new Thread(c1).start();
        new Thread(c2).start();
    }
}

class Producer implements Runnable{

    Logger logger = Logger.getLogger("logger");
    static AtomicInteger i = new AtomicInteger();
    private final BlockingQueue queue;
    String name;

    Producer (BlockingQueue queue, String name){
        this.queue = queue;
        this.name = name;
    }

    public int produce() throws InterruptedException {
        int data = i.incrementAndGet();
        logger.info(String.format("%s 이/가 데이터를 삽입합니다: %d", name, data));
        Thread.sleep(500);
        return data;
    }

    @Override
    public void run() {
        try{
            while(true){
                queue.put(produce());
            }
        } catch (InterruptedException ignore){

        }
    }
}

class Consumer implements Runnable {
    Logger logger = Logger.getLogger("logger");
    private final BlockingQueue queue;
    String name;

    Consumer(BlockingQueue queue, String name){
        this.queue = queue;
        this.name = name;
    }

    public void consume(Object target) throws InterruptedException {
        logger.info(String.format("%s 이/가 데이터를 소비합니다: %d " , name, target));
        Thread.sleep(500);
    }

    @Override
    public void run(){
        try{
            while(true){
                consume(queue.take());
            }
        } catch (InterruptedException ignore){

        }
    }
}

 

 - 실행 결과

 

 

✅ Reference

 

 - https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html