[TIL] RabbitMQ 이해하기

 

8/13 진행사항

✅ 알고리즘 1문제

✅ 대규모 스트림 처리 1-1 ~ 1-8

 

 

RabbitMQ란?

 

메시지 브로커라는 소프트웨어로, 두 시스템 사이에서 메시지를 전달하는 역할을 한다.

👉 데이터(메시지)를 송신자(프로듀서)로부터 수신자(컨슈머)에게 전달하는 중단 매개체.

 

기본 구성 요소

1. Message - 전달하고자 하는 실제 데이터
2. Producer - message를 생성하고 이를 RabbitMQ로 전송하는 역할
3. Queue - message를 임시로 저장하는 장소
4. Consumer - Queue에 있는 메시지를 받아 처리하는 역할
5. Exchange - 프로듀서가 보낸 메시지를 적절한 Queue로 라우팅 하는 역할

RabbitMQ는 AMQP 프로토콜을 사용하여 메시지를 관리하는 역할을 한다.

AMQP) Advanced Message Queuing Protocol

 

Exchage 유형

 

1. Direct Exchange - 라우팅 키가 정확히 일치하는 큐로 메시지를 전달
2. Topic Exchange - 라우팅 키의 패턴을 기반으로 메시지를 전달
3. Fanout Exchange - 라우팅 키와 상관없이 모든 연결된 큐에 메시지를 전달
4. Headers Exchange - 라우팅 키 대신 메시지 헤더의 속성을 기반으로 메시지를 전달

현업에서는 주로 Direct Exchange를 사용한다.

 

AMQP

 

메시지를 안전하고 신뢰성 있게 전달하는 방식을 정의한 프로토콜로,

다양한 시스템 간에 메시지를 주고받을 때 일관된 방법을 제공하는 것이 목적이다.

프로토콜) 컴퓨터와 네트워크에서 데이터를 주고받기 위한 일련의 규칙과 절차

 

 

 

RabbitMQ 실습

producer : order (메세지 생성 후 두 개의 큐에 전달)

consumer : product-1, product-2, payment

queue : market.product, market.payment

 

도커를 사용하여 RabbitMQ 설치

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

 

http://localhost:15672 접속 후 guest/guest로 로그인

 

Order Application

 

build.gradle

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	implementation 'org.springframework.boot:spring-boot-starter-web'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

 

application.properties

spring.application.name=order

message.exchange=market
message.queue.product=market.product
message.queue.payment=market.payment

spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest

 

OrderApplicationQueueConfig.java

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

TopicExchange : 라우팅 키에 따라 메시지를 특정 큐로 라우팅 하는 역할

여기서 exchange 필드 값을 사용하여 익스체인지 이름을 설정함

 

Binding: 바인딩은 큐와 익스체인지를 연결하는 역할

queueProduct()로 생성된 큐를 익스체인지와 연결하고 라우팅 키로 queueProduct 값을 사용하여 바인딩

 

OrderController.java

@RestController
@RequiredArgsConstructor
public class OrderController {

    private final OrderService orderService;

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

 

OrderService.java

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;

    private final RabbitTemplate rabbitTemplate;

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

RabbitTemplate : Spring에서 RabbitMQ와 상호작용하기 위한 기본 템플릿 클래스

convertAndSend : 메시지를 특정 큐나 익스체인지로 보내는 데 사용

👉productQueue와 paymentQueue에 orderId를 보냄

 

 

localhost:8080/order/1 실행

market이라는 이름으로 생성 된 exchage 확인

 

market.payment와 market.product라는 이름으로 생성된 queue 확인

 

PaymentApplication

 

build.gradle

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

 

application.properties

spring.application.name=payment

message.queue.payment=market.payment

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

 

PaymentEndpoint.java

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: 이 메소드가 RabbitMQ의 메시지를 수신하는 리스너 메소드임을 나타냄

• queues 속성: 메시지를 수신할 큐의 이름을 지정

market.payment의 이름을 가진 큐에 orderId 메시지가 도착하면

@RabbitListener가 붙은 메소드를 호출하고 매개변수로 메시지를 받는다.

market.payment message가 소모된 것 확인

 

 

Product Application

 

build.gradle

dependencies {
	implementation 'org.springframework.boot:spring-boot-starter-amqp'
	compileOnly 'org.projectlombok:lombok'
	annotationProcessor 'org.projectlombok:lombok'
	testImplementation 'org.springframework.boot:spring-boot-starter-test'
	testImplementation 'org.springframework.amqp:spring-rabbit-test'
	testRuntimeOnly 'org.junit.platform:junit-platform-launcher'
}

 

application.properties

spring.application.name=product

message.queue.product=market.product

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

 

PaymentEndpoint.java

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 ProductEndpoint {

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

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

 

컨슈머가 라운드로빈으로 메시지 전달 받는 것 확인하기