25.03.10 RabbitMQ 란?

개요

RabbitMQ 에 대해 강의로 처음 접하게되었다. 개념와 사용법에 대해 정리해본다.

 

 

---

 

RabbitMQ 란?

AMQP ( Advanced Message Queuing Protocol ) 기반의 오픈 소스 메시지 브로커 ( Message Broker ) 이다.

애플리케이션 간의 비동기 메시지 처리를 지원하여, 높은 확장성과 안정성을 가진 분산 시스템을 구축할 때 유용하다.

 

RabbitMQ의 핵심 개념

RabbitMQ는 메시지를 모내는 Producer( 발행자 ) 와 메시지를 받는  Consumer ( 구독자 ) 사이에서 Broker 역할을 수행한다. 즉, Producer가 메시지를 Queue에 전달하면 Consumer가 메시지를 가져가는 방식으로 동작한다.

 

출처 : https://www.cloudamqp.com/img/blog/exchanges-topic-fanout-direct.png

 

 

1. Message ( 메시지 )

• Producer가 보내는 데이터 단위 (예 : JSON, String, Binary 데이터 등 )
• 메시지는 Exchange -> Queue -> Consumer 순서로 이동한다.

2. Queue ( 큐 )

• 메시지를 보관하는 버퍼 역할
• Consumer가 메시지를 가져갈 때 까지 메시지는 큐에 쌓인다.
• FIFO (First In, First Out) 방식으로 동작

3. Exchange ( 교환기 )

• Producer가 보낸 메시지를 적절한 Queue로 라우팅하는 역할
• Exchange 유형에 따라 라우팅 방식이 달라짐

4. Binding ( 바인딩 )

• Exchange와 Queue를 연결하는 설정
• Binding Key를 설정하면 특정 메시지만 특정 Queue로 전달 가능

5. Producer ( 생산자 )

• 메시지를 생성해 Exchange에 전송

6. Consumer ( 소비자 )

• Queue 에서 메시지를 가져와 처리
• 하나의 Queue에 여러 Consumer가 존재할 수도 있음

---

RabbitMQ의 주요 동작 방식

1. Producer가 메시지를 생성하여 Exchange로 보냄
2. Exchange가 메시지를 적잘한 Queue로 라우팅
3. Queue에 메시지가 저장됨
4. Consumer가 Queue에서 메시지르 가져와 처리
5. Consumer가 처리를 완료하면 ACK ( 확인 응답 )을 보냄

---

Exchange 유형

 

Direct Exchange ( 직접 교환기 )

• 특정 라우팅 키 ( Routing Key )와 정확히 일치하는 메시지만 전달
• 하나의 메시지가 하나의 Queue로 향함
• 예시 ) routing.key = "order.created"

Fanout Exchange ( 팬아웃 교환기 )

• 라우팅 키와 상관없이 모든 바인딩된 Queue로 메시지를 전송
• 브로드캐스트 방식
• 예시 ) SNS 알림, 로그 시스템

Topic Exchange ( 토픽 교환기 )

• 라우팅 키 패턴을 활용하여 메시지를 전달
• .(점)으로 구분된 여러개의 단어 조합 가능
• 예시
  •  error.log → error.# 와 매칭됨
  • system.info.log → *.info.* 와 매칭됨

Headers Exchange ( 헤더 교환기 )

• 라우팅 키 대신 헤더 값으로 메시지를 라우팅
• 정확한 헤더 값이 일치해야함

---

RabbitMQ의 특징 및 장점

• 비동기 메시징 지원
  
  • 시스템의 유연성과 확장성 증가
• 분산 환경에서 강력한 메시지 전달 보장
  • 데이터 유실 방지 ( Persistent 메시지 )
• 여러 개의 Consumer를 활용한 로드 밸런싱 가능
• 다양한 프로토콜 및 언어 지원 ( Java, Python, JavaScript 등 )
• 고성능, 경량 메시지 브로커

---

 

그럼 RabbitMQ는 왜 사용하는가?

 

애플리케이션 간 결합도를 낮추기 위해

• 직접적인 서비스 호출 대신 메시지를 주고받는 방식 채용
• 서비스 간 독립성을 유지하며 유연한 아키텍처 구현 

비동기 처리 및 성능 향상

• 메시지 Queue에 넣고 비동기적으로 처리
• 요청이 물려도 빠른 응답 가능

로드 밸런싱

• 여러 Consumer가 하나의 Queue를 처리하여 부하 분산

안전적인 메시지 전달 보장

• RabbitMQ는 메시지를 디스크에 저장 가능하여 장애 발생 시 복구 가능

 

---

RabbitMQ 적용 사례

• 이메일 전송 시스템 ( 비동기 이메일 전송 )
• 주문 처리 시스템 ( 결제 완료 후 주문 처리 )
• 로그 수집 시스템 ( 여러 애플리케이션에서 로그를 중앙 서버로 전송 )
• 채팅 서비스 ( 실시간 메시지 전달 )
• MSA 간 통신 

---

Spring Boot에서 RabbitMQ 사용 예시

 

의존성 추가

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-amqp'
}

 

 

application.yml 설정

# message value
message.exchange=market # exchange 명
message.queue.product=market.product 
message.queue.payment=market.payment

# rabbit mq config
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest

 

 

message를 전달한 애플리케이션에서 설정

package com.market.order;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class OrderApplicationQueueConfig {

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

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

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

    @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);}
}

 

이 클래스는 Exchange, Queue, Binding을 설정하는 설정 클래스이다.

 

설정된 yml 파일에서 프로퍼티 값을 가져와 exchange, queueProduct, queuePayment 변수에 저장한다.

그다음, Exchange 설정을 진행한다.
RabbitMQ의 TopicExchange를 생성하며, exchange 변수 값을 가져와 "market"이라는 이름의 Exchange를 등록한다

 

이후, Queue 객체를 생성하고 스프링 빈으로 등록한다

• "market.product" 이름의 큐를 생성
• "market.payment" 이름의 큐를 생성

마지막으로, Queue와 Exchange를 바인딩한다.
BindingBuilder.bind() 메서드를 사용하여 큐와 Exchange를 연결한다.

바인딩의 이름은 큐에 이름과 동일하게 사용하는것이 좋다.

• queueProduct()를 exchange()에 "market.product" 라우팅 키로 바인딩
• queuePayment()를 exchange()에 "market.payment" 라우팅 키로 바인딩

이렇게 설정하면, 특정 라우팅 키를 가진 메시지가 해당 큐로 전달되도록 구성된다.

 

message 전송

package com.market.order;

import lombok.RequiredArgsConstructor;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

@Service
@RequiredArgsConstructor
public class OrderService {

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

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

    // RabbitMQ 로 요청을 보낼때 사용하는 템플릿
    private final RabbitTemplate rabbitTemplate;

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

 

전송할 큐의 이름을 프로퍼티 값으로 주입받는다.

 

RabbitTemplate

spring에서 제공하는 RabbitMQ 메시지 전송을 위한 템플릿 클래스이다.

템플릿을 통해 큐로 메시지를 전송할 수 있다.

 

rabbitTemplate.convertAndSend 를 통해 market.product, market.payment 큐에 매개변수 값을 메시지로 전송한다.

 

RabbitMQ는 각 큐에 메시지를 저장하고, 해당 큐의 메시지를 consumer가 받아 처리한다.

 

 

Consumer 가 받아서 메시지 처리

package com.market.payment;


import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@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);
    }
}

 

앱의 이름을 가져오는 이유는 로드 밸런싱이 정상적으로 동작하는지 확인하기 위해서이다.

@RabbitListener 어노테이션을 사용하여 queues에 선언된 RabbitMQ 큐를 구독(Consume)하는 메서드로 지정한다.

메시지의 내용을 String 타입의 orderId로 받아와 테스트를 진행한다.

 

 

---

 

RabbitMQ의 메시지 분배 방식

RabbitMQ는 기본적으로 메시지를 라운드 로빈 ( Round Robin ) 방식으로 분배해 로드 밸런싱을 수행한다.

 

즉 동일한 애플리케이션이 여러 개의 인스턴스로 실행되고 ( 예 : 서로 다른 포트에서 실행 ) 동일한 큐를 구독하면,

RabbitMQ는 메시지를 각 인스턴스에 순차적으로 할당한다. 이를 통해 하나의 애플리케이션 인스턴스가 모든 메시지를 독점하는 것이 아니라, 여러 개의 인스턴스가 번갈아 가며 메시지를 처리하게 된다.

 

라운드 로빈 방식이 정상적으로 동적하는 조건은 다음 과 같다.

• 여러 개의 애플리케이션 인스턴스가 동일한 큐를 구독해야 한다.
  • @RabbitListener 설정이 모든 인스턴스가 동일하게 설정되어 있어야함
• 공유 큐 ( Shared Queue ) 방식이여아한다.
  • 메시지는 큐 단위 로 분배되기 때문에, 하나의 큐를 여러 인스턴스가 구독할 때만 라운드 로빈이 적용
• 메시지가 자동으로 ACK ( 확인 응답 ) 처리 되어야한다.
  • 기본적으로 Spring AMQP는 메시지를 처리하면 자동으로 ACK를 보내지만, ackMode를 MANUAL로 설정하면 수동으로 처리할 수도 있음.

---

요약

RabbitMQ는 비동기 메시지 처리, 마이크로서비스 간 통신, 성능 최적화 등의 목적에서 사용되며, 높은 확장성과 안정성을 제공하는 메시지 브로커이다.
Spring Boot와 쉽게 통합할 수 있어 마이크로서비스 아키텍처에서 많이 활용된다.

RabbitMQ를 활용하면 애플리케이션 간 결합도를 낮추고, 시스템의 성능을 개선하며, 안정적인 메시지 전달을 보장한다.