[Java] LinkedBlockingQueue 클래스

 

1. 정의

An optionally-bounded blocking queue based on linked nodes. This queue orders elements FIFO (first-in-first-out). The head of the queue is that element that has been on the queue the longest time. The tail of the queue is that element that has been on the queue the shortest time. New elements are inserted at the tail of the queue, and the queue retrieval operations obtain elements at the head of the queue. Linked queues typically have higher throughput than array-based queues but less predictable performance in most concurrent applications.

The optional capacity bound constructor argument serves as a way to prevent excessive queue expansion. The capacity, if unspecified, is equal to Integer.MAX_VALUE. Linked nodes are dynamically created upon each insertion unless this would bring the queue above capacity.

This class and its iterator implement all of the optional methods of the Collection and Iterator interfaces.

This class is a member of the Java Collections Framework.

 

linked node를 기반으로한 선택적으로 제한된 차단 큐

큐의 요소들은 FIFO(First-In-First-Out) 순서를 선다

큐의 head는 큐에 가장 오래 있던 요소, 큐의 tail은 가장 짧게있던 요소

새로운 요소는 큐의 tail에 삽입되고 대기열 검색 작업은 큐 헤드에 있는 요소를 가져옴

연결된 큐는 일반적으로 배열기반 큐보다 처리량이 높지만 대부분의 동시 애플리케이션에서 예측 가능성이 낮다

선택적 용량 제한 생성자 인수는 과도한 큐 확장을 방지하는 방법을로 사용된다

지정되지 않은 경우 용량은 Integer.MAX_VALUE와 같다.

연결된 노드는 대기열이 용량을 초과하지 않는 한 삽입할 때마다 동적으로 생성된다.

 

이 클래스와 해당 iterator는 Collection 및 Iterator 인터페이스의 모든 선택적 메서드를 구현한다.

 

이 클래스는 자바 컬렉션 프레임워크의 멤버이다.

 

 

2. 큐에 요소 추가하기

1) offer() 메서드

큐의 용량을 초과하지 않고 요소를 추가할 수 있는 경우 요소를 추가함

성공하면 true, 큐가 가득차서 요소를 추가하지 못한 경우 false 반환

 

 

2) put() 메서드

큐의 끝에 지정된 요소를 삽입하고 필요한 경우 공간을 사용할 수 있을 때 까지 대기

 

3. 큐에 요소 제거하기

1) take() 메서드

큐의 헤드를 검색하고 필요한 경우 요소를 사용할 수 있을 때 까지 대기

 

2) poll() 메서드

큐의 헤드를 검색하고 제거하거나, 큐가 비어 있는 경우 null 반환

 

3. 예제

package com.example.queueproject.service;

import com.example.queueproject.dto.CsBotRequestDto;
import com.example.queueproject.dto.CsBotResponseDto;
import lombok.RequiredArgsConstructor;
import org.springframework.http.*;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Service
@RequiredArgsConstructor
public class CsBotService {
    private final LinkedBlockingQueue<CsBotRequestDto> requestQueue = new LinkedBlockingQueue<>(3);
    private final RestTemplate restTemplate;

    public CsBotResponseDto getCsBot(CsBotRequestDto csBotRequestDto) {
        /*
            큐에 요청 추가
            put() 메서드는 큐가 가득 찬 경우 호출되면 block 됨
            새로운 요소를 추가하기 위해서 큐에서 요소가 제거되길 기다림
            offer() 메서드는 큐가 가득 찬 경우 false를 반환하며 요소를 추가하지 않음
        */
        try {

            // 큐에 요청 추가 (일정 시간 동안 기다림)
            if (!requestQueue.offer(csBotRequestDto, 1, TimeUnit.SECONDS)) {
                System.out.println("큐가 가득 차서 요청을 추가하지 못했습니다.");
                return null;
            }

            // 가정: 요청 처리를 시뮬레이션하기 위해 6초 동안 대기
            Thread.sleep(6000);

            // 요청 처리 로직
            System.out.println("Processing request: " + csBotRequestDto);
            int queueSize = getQueueSize();
            System.out.println("offer 현재 큐 사이즈:" + queueSize);

            // 요청 헤더 설정
            HttpHeaders headers = new HttpHeaders();
            headers.setContentType(MediaType.APPLICATION_JSON);

            // 요청 데이터 설정
            HttpEntity<CsBotRequestDto> requestEntity = new HttpEntity<>(csBotRequestDto, headers);

            // 요청 url
            String url = "http://localhost:8000/api";

            // POST 요청 보내기
            System.out.println(requestEntity);
            ResponseEntity<CsBotResponseDto> responseEntity = restTemplate.exchange(
                    url,
                    HttpMethod.POST,
                    requestEntity,
                    CsBotResponseDto.class
            );
            System.out.println("responseEntity" + responseEntity);



            // 큐에서 요청 제거
                requestQueue.poll();

            System.out.println("take 현재 큐 사이즈:" + queueSize);

            return responseEntity.getBody();

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // 예외 처리
            System.out.println("에러");
        }
        return null;
    }

    public int getQueueSize() {
        return requestQueue.size();
    }
}