Blocking Queue를 이용해서 Producer-Consumer 패턴 구현하기

public class CounterBroker {
    // 최대 1000개의 이벤트를 저장할 수 있다.
    private final BlockingQueue<Function<Integer, Integer>> queue = new LinkedBlockingQueue<>(1000);

    public void addEvent(int value){
        try{
            // 이 이벤트를 컨슈머가 처리할 당시 count와 value를 더한 값을 출력한다.
            queue.put((c) -> c+value);
        }
        catch(InterruptedException e){
            Thread.currentThread().interrupt();
        }
    }
    public Function<Integer, Integer> take(){
        try {
            return queue.take();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}

public class CounterProducer {
    private final CounterBroker counterBroker;

    public CounterProducer(CounterBroker counterBroker) {
        this.counterBroker = counterBroker;
    }

    public void add(int value){
        counterBroker.addEvent(value);
    }
}

public class CounterCustomer {
    private final CounterBroker counterBroker;
    private AtomicInteger count = new AtomicInteger(100); // AtomicInteger로 변경

    public CounterCustomer(CounterBroker counterBroker) {
        this.counterBroker = counterBroker;
    }

    public void consumEvent(){
        while(true){
            Function<Integer,Integer> event = counterBroker.take();
            IntUnaryOperator operator = event::apply;
            count.updateAndGet(operator); // 원자적인 연산 보장하기 위함
        }
    }
    public int show(){ // 스레드 안전하게 카운트 결과를 출력한다.
        // AtomicInteger의 get 메서드를 사용하여 값을 가져옴
        return count.get();
    }
}

public class QueueCounterTest {
    private final int counteNumber = 1;
    private final int totalCount = 10000;
    private static final Logger logger = LoggerFactory.getLogger(SynchronizedCounterTest.class);
    @Test
    @DisplayName("producer consumer 패턴을 이용해서 더하기 이벤트 발생 스레드와 더하기 이벤트 처리 스레드를 분리하자")
    public void 프로듀서_컨슈며_더하기_멀티_프로듀서_멀티_컨슈머() throws InterruptedException {
        CounterBroker counterBroker = new CounterBroker();
        CounterCustomer customer = new CounterCustomer(counterBroker);
        CounterProducer producer = new CounterProducer(counterBroker);
        LocalTime lt1 = LocalTime.now();

        int initalCount = customer.show();
        ExecutorService service = Executors.newFixedThreadPool(15);
        CountDownLatch latch = new CountDownLatch(totalCount);
        // CounterCustomer 스레드 생성 및 비동기로 처리 시작
        for(int i=0; i<3; i++){
            CompletableFuture.runAsync(customer::consumEvent);
        }
        // 프로듀서 스레드 생성
        for (int i = 0; i < totalCount; i++) {
            service.submit(() -> {
                producer.add(counteNumber);
                latch.countDown();
            });
        }
        latch.await();
        int finalCount = customer.show();

        LocalTime lt2 = LocalTime.now();
        long dif = Duration.between(lt1, lt2).getNano();
        logger.info("프로듀서_컨슈며_더하기_멀티_프로듀서_단일_컨슈머 테스트가 걸린 시간 : " + dif / 1000 + "ms");
        Assertions.assertEquals(initalCount + totalCount*counteNumber, finalCount);
    }
}

expected: <10100> but was: <10096>
...
at com.thread.concurrency.counter.QueueCounterTest.프로듀서_컨슈며_더하기_멀티_프로듀서_멀티_컨슈머(QueueCounterTest.java:53)

public class CounterConsumer {
    ...

    public void consumeEvent() throws InterruptedException {

        while (!queue.isEmpty()) {
            System.out.println(Thread.currentThread().getName()+"은 현재 큐 사이즈 : "+queue.size());
            Long value = queue.take();
//            count.addAndGet(value);
            System.out.println("결과 카운트 : "+count.addAndGet(value));
        }
    }
    public Long show(){
        while(true){
            if(queue.isEmpty()){
                return count.get();
            }
        }
    }
}

public class CounterConsumer implements Consumer {
    private final BlockingQueue<Long> queue;
    private final AtomicLong count = new AtomicLong(0);

    public CounterConsumer(BlockingQueue<Long> queue) {
        this.queue = queue;
    }

    public void consumeEvent(long timeout, TimeUnit unit) throws InterruptedException {
        while (true) {
            Long value = queue.poll(timeout, unit);
            if(value == null){
                break;
            }
            count.addAndGet(value);
        }
    }
    public Long show(){
        while(true){
            if(queue.isEmpty()){
                return count.get();
            }
        }
    }
}

...
// 컨슈머 스레드 생성
consumerService.submit(() -> {
    try {
        consumer.consumeEvent(timeout, unit);
        consumerLatch.countDown();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
});
consumerLatch.await(); // 모든 컨슈머 스레드가 작업을 끝낼 때까지 대기
...