RabbitMQ 기본 개념 및 실습

RabbitMQ란

RabbitMQ는 메시지 브로커다.

메시지 브로커는 데이터(메시지)를 송신자(프로듀서)로부터 수신자(컨슈머)에게 전달하는 중간 매개체 역할을 한다.

RabbitMQ는 메시지를 큐(queue)에 저장하고, 필요할 때 적절한 수신자(컨슈머)에게 전달한다.

 

RabbitMQ의 역할

• 비동기 처리: 데이터를 비동기적으로 처리하여 시스템의 응답성을 높인다.
• 부하 분산: 여러 소비자에게 메시지를 분산시켜 시스템의 부하를 균형 있게 분산한다.
• 내결함성: 메시지를 안전하게 저장하여 시스템 장애 시 데이터 손실을 방지한다.

 

RabbitMQ 장단점

---장점---

신뢰성

• 메시지 지속성: RabbitMQ는 메시지를 디스크에 저장하여 시스템 장애 발생 시에도 메시지가 손실되지 않도록 한다.
• 확인 메커니즘: 메시지가 성공적으로 소비자에게 전달되었는지 확인하는 ACK 메커니즘을 지원한다.

유연성

• 다양한 메시지 패턴: RabbitMQ는 여러 가지 메시지 전달 패턴(단일 소비자, 다중 소비자, 라운드 로빈, 팬아웃, Topic 기반 등)을 지원한다.
• 프로토콜 지원: 기본적으로 AMQP(Advanced Message Queuing Protocol)를 사용하지만, STOMP, MQTT 등 다양한 프로토콜도 지원한다.

확장성

• 클러스터링: RabbitMQ는 클러스터링을 통해 여러 노드로 구성된 환경에서 높은 가용성과 부하 분산을 제공한다.
• 분산 아키텍처: 페더레이션(federation)을 통해 분산된 메시징 시스템을 구축할 수 있다.

관리 및 모니터링

• 관리 인터페이스: 웹 기반 관리 인터페이스를 통해 queue, exchange, binding 등을 쉽게 관리할 수 있다.
• 플러그인 시스템: 다양한 플러그인을 통해 기능을 확장할 수 있다(예: 관리 플러그인, 모니터링 플러그인 등).

성능

• 높은 처리량: 적절히 구성된 RabbitMQ는 높은 메시지 처리량을 제공하여 대규모 애플리케이션에서도 효과적이다.

---단점---

설정 및 운영 복잡성

• 복잡한 설정: RabbitMQ의 초기 설정이 다소 복잡할 수 있으며, 클러스터링 및 분산 환경에서는 더욱 많은 설정이 필요하다.
• 운영 관리: 대규모 환경에서 RabbitMQ를 운영하고 관리하는 것은 복잡하다.

성능 문제

• 메시지 브로커 오버헤드: RabbitMQ는 모든 메시지를 중앙 브로커를 통해 전달하기 때문에, 높은 트래픽 상황에서는 브로커의 오버헤드가 발생할 수 있다.
• 대규모 메시지 처리: 대규모 메시지를 처리할 때 성능 저하가 발생할 수 있으며, 이러한 경우에는 적절한 클러스터링 및 최적화가 필요하다.

운영 비용

• 리소스 소비: RabbitMQ는 메모리와 CPU 자원을 많이 소비할 수 있어, 충분한 리소스를 제공해야 한다.
• 모니터링 및 유지보수: 지속적인 모니터링과 유지보수가 필요하며, 이를 위해 추가적인 인력과 비용이 발생할 수 있다.

제한된 메시지 크기 (한계)

• 메시지 크기 제한: RabbitMQ는 용량이 큰 메시지 처리에 제한이 있을 수 있으며, 대용량 파일 전송에는 적합하지 않다.

장단점이 명확하므로 적절한 trade-off 타협점을 찾아 사용할지 말지를 결정하자.

 

RabbitMQ의 기본 구성 요소

메시지(Message)

메시지는 RabbitMQ를 통해 전달되는 데이터 단위로 사용자 등록 정보나 주문 내역이 메시지가 될 수 있다.

 

프로듀서(Producer)

메시지를 생성하고 RabbitMQ에 보내는 역할을 한다.

예를 들어, 웹 애플리케이션이 사용자 등록 정보를 RabbitMQ에 보내는 경우 프로듀서가 된다.

 

큐(Queue)

메시지를 저장하는 장소로 메시지는 큐에 저장되었다가 컨슈머에게 전달된다.

큐는 FIFO(First In, First Out) 방식으로 메시지를 처리한다.

 

컨슈머(Consumer)

큐에서 메시지를 가져와 처리하는 역할을 한다.

 

익스체인지(Exchange)

메시지를 적절한 큐로 라우팅하는 역할을 한다.

프로듀서는 메시지를 직접 큐에 보내지 않고, 익스체인지에 보내며, 익스체인지는 메시지를 적절한 큐로 전달한다.

 

RabbitMQ와 AMQP

RabbitMQ는 AMQP(Advanced Message Queuing Protocol)를 사용한다.

AMQP는 메시지 브로커를 위한 프로토콜로, 메시지의 생성, 전송, 큐잉, 라우팅 등을 표준화하여 메시지 브로커가 상호 운용될 수 있게 한다.

AMQP의 주요 개념

• 메시지(Message): 전송되는 데이터 단위
• 큐(Queue): 메시지를 저장하고 전달하는 자료 구조
• 익스체인지(Exchange): 메시지를 큐로 라우팅하는 역할
• 바인딩(Binding): 익스체인지와 큐를 연결하는 설정. 바인딩을 통해 메시지가 어느 큐로 전달될지 정한다.

 

익스체인지 유형

메시지 브로커가 메시지를 교환기에서 큐로 라우팅하는 방식이다.

익스체인지는 다양한 방식으로 메시지를 라우팅할 수 있으며, 주로 메시지의 라우팅 키와 바인딩 키 또는 패턴을 기반으로 작동한다.

1. Direct Exchange (가장 많이 쓰인다고 함)
   • 라우팅 키가 정확히 일치하는 큐로 메시지를 전달
   • 예를 들어, 라우팅 키가 error인 메시지는 error라는 바인딩 키를 가진 큐로 전달된다.
2. Topic Exchange
   • 라우팅 키의 패턴을 사용하여 메시지를 라우팅한다.
   • 패턴에는 와일드카드 * (0개 이상의 단어)와 #(단어 하나)가 사용된다.
   • 예를 들어, 라우팅 키가 queue.message.rabbit인 메시지는 바인딩 키가 .message.인 큐로 전달된다.
3. Fanout Exchange
   • 라우팅 키를 무시하고 교환기에 바인딩된 모든 큐로 메시지를 브로드캐스트한다.
   • 모든 바인딩된 큐로 메시지가 전달된다.
4. Headers Exchange
   • 라우팅 키 대신 메시지의 헤더를 기반으로 메시지를 라우팅한다.
   • 헤더 값과 바인딩된 헤더 값이 일치하는 큐로 메시지를 전달한다.

 

RabbitMQ 사용해보기

도커를 사용하여 RabbitMQ를 설치

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 --restart=unless-stopped rabbitmq:management

 

클라이언트 연결은 5672 포트, 관리 콘솔 연결은 15672 포트

localhost:15672에 접속하면 로그인 페이지가 나타난다.

http://localhost:15672

 

 

guest/guest (default)를 입력하여 접속하면 대시보드를 볼 수 있다.

RabbitMQ 어드민 페이지

 

프로듀서 만들기

Spring for RabbitMQ dependency 추가 후 스프링 프로젝트 생성

 

application.yml

spring:
  application:
    name: order
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
message:
  exchange: market
  queue:
    product: market.product
    payment: market.payment

 

exchange의 이름은 market, 큐의 이름은 market.product와 market.payment가 있다.

 

 

QueueConfig

@Configuration
public class QueueConfig {

    @Value("${message.exchange}")
    private String exchange;

    @Value("${message.queue.product}")
    private String queueProduct;

    @Value("${message.queue.payment}")
    private String queuePayment;

// exchange 생성
    @Bean
    public TopicExchange exchange() {
        return new TopicExchange(exchange);
    }

// 큐 생성
    @Bean
    public Queue queueProduct() {
        return new Queue(queueProduct);
    }

    @Bean
    public Queue queuePayment() {
        return new Queue(queuePayment);
    }

// 바인딩 생성
    @Bean
    public Binding bindingProduct() {
        return BindingBuilder.bind(queueProduct())
                .to(exchange())
                .with(queueProduct);
    }

    @Bean
    public Binding bindingPayment() {
        return BindingBuilder.bind(queuePayment())
                .to(exchange())
                .with(queuePayment);
    }
}

 

Controller와 Service는 간단하게만 작성한다.

 

// controller
@GetMapping("/order/{id}")
public String order(@PathVariable String id) {
    orderService.createOrder(id);
    return "Order complete";
}

// service
private final RabbitTemplate rabbitTemplate;

public void createOrder(String orderId) {
    rabbitTemplate.convertAndSend(productQueue, orderId);
    rabbitTemplate.convertAndSend(paymentQueue, orderId);
}

 

요청을 보내보자. localhost:8080/order/1

 

두 개의 큐가 생성되었고, 메시지는 1이 담겨 있다.

 

market.payment 큐에 들어가서 bindings를 확인해보자.

 

 

라우팅 키를 큐의 이름과 동일시 했기 때문에 위와 같은 결과가 출력되었다.

 

메시지를 확인해보자.

 

url 파라미터로 전송한 메시지 1이 담겨있다.

 

 

컨슈머 만들기(Product, Payment)

스프링 이니셜라이저에서 프로듀서를 만들었던 것처럼 똑같이 생성하고 yml 파일을 수정한다.

Payment > application.yml > 컨슈머 입장에서 다른 큐와 exchange를 알 필요 없으므로 제거한다.

spring:
  application:
    name: payment
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
message:
  queue:
    payment: market.payment

 

@RabbitListener를 통해 어떤 큐를 통해 메시지를 받을지 정한다.

@Slf4j
@Component
public class PaymentEndpoint {

    @Value("${spring.application.name}")
    private String appName;

    @RabbitListener(queues = "${message.queue.payment}")
    public void receiveMessage(String orderId) {
        log.info("receive orderId:{}, appName : {}", orderId, appName);
    }

}

 

Order 프로젝트에서 발행한 메시지가 큐에 담겨 있다가 Payment Consumer가 실행되자마자 소모된 것을 확인할 수 있다.

 

관리자 페이지에서도 market.payment 큐의 메시지가 소모되었다.

마찬가지로 Product 애플리케이션의 application.yml 파일도 작성해준다.

spring:
  application:
    name: product
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
message:
  queue:
    payment: market.product

 

똑같이 @RabbitListener(queues = "${message.queue.payment}") 를 만들어주고,

구성 편집에 들어가서 다음과 같이 설정한다.

 

 

OrderApplication을 실행하고 ProductApplication-1, -2를 실행한 뒤, order controller로 요청을 여러 번 보낸다.

 

라운드 로빈 형식으로 메시지가 번갈아 가며 소모되는 것을 확인할 수 있다.

 

ProductApplication-1

 

 

ProductApplication-2