[Spring boot] RabbitMQ

RabbitMQ

RabbitMQ는 시스템 간에 메시지를 안전하게 주고받을 수 있게 해주는 대표적인 오픈소스 메시지 브로커이다.

개요는 다음 페이지 참고
https://nohumbit.tistory.com/42

[Architecture] Event-Driven Archiecture 개요

 

소개

프로토콜

RabbitMQ는 다양한 메세지 패턴(단일 소비자, 다중 소비자, 라운드로빈, 팬아웃, Topic 기반 등)을 지원하며, 여러 프로토콜 또한 지원한다

• AMQP (Advanced Message Queuing Protocol)
  • RabbitMQ에서 기본적으로 사용됨
  • Exchange와 Queue를 이용해 메세지를 라우팅
  • 주로 MSA 간 데이터 전송에 사용되며, 트랜잭션 보장과 메시지 유실 방지가 필요한 신뢰성 위주의 서버 환경에서 사용
• STOMP (Simple Text-Oriented Message Protocol)
  • 가볍고 단순한 텍스트 기반 프로토콜
  • Pub/Sub 커맨드 방식 (CONNECT, SUBSCRIBE, SEND, MESSAGE 등)
  • 실시간 양방향 통신 환경에서 사용
• MQTT (Message Queuing Telemetry Transport)
  • 초경량 바이너리 헤더로, 프로토콜 헤더 크기가 최소 2Byte
  • QoS 3단계 제어 (보내고 말기, 적어도 1회 전달, 정확히 1회 전달)

클러스터링

RabbitMQ는 모든 메시지를 중앙 브로커를 통해 전달하기 때문에 오버헤드가 발생할 수 있다. 따라서 클러스터링을 통해 여러 노드로 구성된 환경에서 높은 가용성과 부하 분산을 제공한다.

• Shared-Nothing Architecture
  • 디스크나 메모리를 공유하지 않는 구조
• Quorum Queue
  • Raft 합의 알고리즘 기반 메시지 복제 및 자동 장애 조치 (Failover)
• Federation
  • 서로 다른 두 클러스터를 연결하여 WAN 환경에서 메시지 교환
• Shovel
  • Federation과 유사하지만, 단순 단방향으로 메시지를 이관/릴레이

주요 개념

기본 구성 요소

RabbitMQ AMQP

RabbitMQ의 기본 구성요소를 간단히 설명한다. 개념적인 부분은 Event-Driven 포스트를 참고하자.

• Message
  • 전달되는 데이터의 단위
• Producer
  • 메세지를 생성하고 브로커에 전달하는 주체
• Queue
  • 메세지를 저장하는 장소 (First In, First Out)
• Consumer
  • Queue의 메세지를 소비하는 주체
• Exchange
  • Message를 적절한 Queue로 라우팅하는 역할
• Binding
  • Exchange와 Queue를 연결하는 설정

Exchange 유형

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

• Direct Exchange
  • 라우팅 키가 정확히 일치하는 Queue로 Message를 전달
  • 대부분의 경우에 사용됨
• Topic Exchange
  • 라우팅 키의 패턴을 사용해 Message를 라우팅
  • 패턴에는 와일드카드 * (단어 1개 대체), #(0개 이상의 단어 대체)이 사용됨
  • ex) *.order.# → create.order (o), order.created (x)
• Fanout Exchange
  • 라우팅 키를 무시하고 바인딩된 모든 큐로 BroadCasting
• Headers Exchange
  • 라우팅 키 대신 Message의 헤더를 기반으로 라우팅

---

실습

실습 구조도

RabbitMQ 환경 구성

• 도커를 통한 RabbitMQ 설치

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

• 대시보드 접속
  • http://localhost:15672 (기본 계정 : guest / guest)
• build.gradle

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

Producer

• application.yml

spring:
    rabbitmq:
        host: localhost
        port: 5672
        username: guest
        password: guest

# 커스텀 매핑 설정
message:
    exchange: market
    queue:
        product: market.product # 라우팅 Key와 Queue
        payment: market.payment

• OrderApplicationQueueConfig.java

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

• OrderService.java

@Service
@RequiredArgsConstructor
public class OrderService {

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

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

    private final RabbitTemplate rabbitTemplate;

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

Consumer (Payment)

• application.yml

spring:
    application:
        name: payment

message:
    queue:
        payment: market.payment

spring:
    rabbitmq:
        host=localhost
        port=5672
        username=guest
        password=guest

• PaymentEndpoint.java

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