[MQ] RabbitMQ를 활용한 메시지 큐 시스템 구축과 라운드 로빈 방식 컨슈머 동작 확인

 

 

프로듀서(Producer)

 

 

Docker로 RabbitMQ 컨테이너 실행

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

 

• docker run
  • Docker 컨테이너를 새로 생성하고 실행하는 기본 명령어입니다.
• -d
  • Detached 모드로 컨테이너를 실행합니다.
  • 즉, 백그라운드에서 컨테이너가 실행되며, 명령어를 입력한 터미널에서 분리됩니다.
• --name rabbitmq
  • 컨테이너에 rabbitmq라는 이름을 지정합니다. 이 이름으로 컨테이너를 참조하고 관리할 수 있습니다.
• -p 5672:5672
  • 포트 매핑을 설정합니다.
  • 호스트의 포트 5672를 컨테이너의 포트 5672에 매핑합니다. RabbitMQ의 기본 AMQP 포트가 5672이므로, 이 설정으로 호스트에서 RabbitMQ에 접근할 수 있습니다.
• -p 15672:15672
  • 호스트의 포트 15672를 컨테이너의 포트 15672에 매핑합니다. 이 포트는 RabbitMQ의 관리 콘솔 웹 인터페이스가 사용하는 포트입니다.
• --restart=unless-stopped
  • 이 옵션은 컨테이너가 자동으로 재시작되도록 설정합니다:
    • unless-stopped: 컨테이너가 수동으로 중지되지 않는 한, Docker 데몬이 재시작될 때마다 컨테이너를 자동으로 재시작합니다. 이는 시스템 재부팅 후에도 컨테이너가 자동으로 시작되도록 보장합니다.
• rabbitmq:management
  • 사용할 Docker 이미지와 태그를 지정합니다.
  • rabbitmq:management는 RabbitMQ의 관리 콘솔이 포함된 공식 이미지입니다. 관리 콘솔을 통해 RabbitMQ의 상태를 모니터링하고 설정을 조정할 수 있습니다.

 

 

실행 후 "docker ps" 명령어로 현재 실행 중인 Docker 컨테이너의 목록을 보면 위와 같이 정상적으로 rabbitmq:management 이미지의 컨테이너가 실행 중입니다.

 

 

RabbitMQ 관리자 웹

 

localhost:15671에 접속하면 다음과 같이 관리자 화면에 접속할 수 있습니다.

---

RabbitMQ 관리자 웹 인터페이스의 초기 아이디와 비밀번호는 기본적으로 다음과 같습니다.

• 아이디 (Username): guest
• 비밀번호 (Password): guest

 

 

 

목표

 

동작 확인의 목표

 

• 프로듀서가 메시지를 큐에 전송할 때, 컨슈머가 없더라도 메시지가 큐에 적재되는지 확인합니다.
• 컨슈머가 메시지를 처리할 때, 큐에서 메시지가 사라지고 제대로 처리되는지 확인합니다.
• 같은 메시지 큐에 바인딩된 컨슈머가 여러 개 있을 때, 라운드 로빈 방식으로 메시지가 분배되는지 확인합니다.

 

---

Order Application(Producer)

application.properties

 

위 아키텍처의 그림처럼 exchange, product, payment를 설정해 주었습니다.

 

• message.exchange
  • RabbitMQ에서 메시지를 라우팅 할 교환기(Exchange)의 이름을 market으로 설정합니다.
• message.queue.product
  • 제품 관련 메시지를 전송할 큐의 이름을 market.product로 설정합니다.
• message.queue.payment
  • 결제 관련 메시지를 전송할 큐의 이름을 market.payment로 설정합니다.

 

 

Queue Config(RabbitMQ 설정 클래스)

 

Queue Config를 통해 RabbitMQ의 큐와 교환기(Exchange)를 생성하고, 이들 간의 연결을 설정합니다.

 

• @Configuration
  • 이 어노테이션은 해당 클래스가 Spring의 설정 클래스로, Bean 정의를 포함하고 있음을 나타냅니다.
• @Value
  • 이 어노테이션은 application.properties 또는 application.yml 파일에서 설정 값을 주입받습니다.
• TopicExchange exchange():
  • TopicExchange를 생성하고, message.exchange의 값으로 교환기 이름을 설정합니다.
    • 본 글에서는 message.exchange는 market.exchange입니다.
• Queue queueProduct()
  • Queue를 생성하고, message.queue.product의 값으로 큐 이름을 설정합니다.
    • 본 글에서는 message.queue.product는 market.product 큐입니다.
• Queue queuePayment():
  • Queue를 생성하고, message.queue.payment의 값으로 큐 이름을 설정합니다.
    • 본 글에서는 message.queue.payment는 market.payment 큐입니다.

---

 

• Binding productBinding()
  • queueProduct 큐를 exchange 교환기에 바인딩합니다.
    • 바인딩할 때 queueProduct 라우팅 키를 사용합니다. 본 글에서는 큐의 이름과 동일(market.product)하게 지정했습니다.
• Binding paymentBinding():
  • queuePayment 큐를 exchange 교환기에 바인딩합니다.
    • 바인딩할 때 queuePayment 라우팅 키를 사용합니다. 본 글에서는 큐의 이름과 동일(market.payment)하게 지정했습니다.

 

Order Controller

 

• @RestController
  • 이 어노테이션은 이 클래스가 RESTful 웹 서비스의 컨트롤러임을 나타냅니다.
  • HTTP 요청을 처리하고 JSON 또는 문자열 형식의 응답을 반환합니다.
• @RequiredArgsConstructor
  • Lombok의 어노테이션으로, 클래스의 final 필드에 대해 자동으로 생성자를 생성합니다. 이를 통해 의존성 주입을 간편하게 처리할 수 있습니다.
• private final OrderService orderService
  • OrderService를 의존성 주입을 통해 주입받아 사용합니다. 이 필드는 final로 선언되어, 객체가 불변임을 보장합니다.
• @GetMapping("/order/{id}"):
  • GET HTTP 요청을 처리하는 간단한 메서드를 정의합니다. URL 경로에 {id} 변수를 포함하여, 이를 메서드의 파라미터로 받습니다.
• public String order(@PathVariable("id") String id)
  • id라는 URL 경로 변수를 메서드의 파라미터로 받습니다.
  • 이 메서드는 orderService.createOrder(id)를 호출하여 주문을 생성하고, 문자열 "Order complete!!"를 응답으로 반환합니다.

Order Service

 

 

 

• @Service
  • 이 어노테이션은 해당 클래스가 Spring의 서비스 컴포넌트임을 나타냅니다. 비즈니스 로직을 처리하는 클래스입니다.
• @RequiredArgsConstructor
  • Lombok의 어노테이션으로, 클래스의 final 필드에 대해 자동으로 생성자를 생성합니다. 이를 통해 의존성 주입을 간편하게 처리할 수 있습니다.
• @Value("${message.queue.product}")
  • application.properties 또는 application.yml 파일에서 message.queue.product 속성의 값을 주입받습니다.
    • 이 값은 RabbitMQ의 제품 큐 이름을 나타냅니다.
• @Value("${message.queue.payment}")
  • application.properties 또는 application.yml 파일에서 message.queue.payment 속성의 값을 주입받습니다. 이 값은 RabbitMQ의 결제 큐 이름을 나타냅니다.
• private final RabbitTemplate rabbitTemplate
  • RabbitTemplate는 Spring AMQP에서 RabbitMQ와의 메시지 전송을 담당하는 클래스입니다. 
• public void createOrder(String orderId):
  • 이 메서드는 주문 ID를 받아 두 개의 큐(productQueue와 paymentQueue)로 주문 ID를 메시지로 전송합니다.
  • Order Service에서는 메시지를 큐로 전송하는 역할을 하는 convertAndSend() 메서드를 사용하여 두 개의 큐(productQueue와 paymentQueue)에 메시지를 보내는 createOrder메서드입니다.

 

 

RabbitTemplate이란❓

RabbitTemplate은 Spring AMQP에서 RabbitMQ와 상호작용하기 위한 핵심 클래스입니다.

 

이 클래스는 메시지를 생성하고 전송하며, RabbitMQ와의 통신을 단순화하고 편리하게 만들어주는 기능을 제공합니다. RabbitTemplate을 사용하면 메시지를 큐에 송신 큐에서 메시지를 수신하는 작업을 쉽게 처리할 수 있습니다.

 

주요 기능

 

• 메시지 전송
  • RabbitTemplate은 메시지를 큐나 교환기로 전송할 수 있습니다. 이를 통해 프로듀서 애플리케이션이 RabbitMQ에 메시지를 보내는 작업을 간편하게 수행할 수 있습니다.
• 메시지 수신
  • 큐에서 메시지를 수신할 수 있는 기능을 제공합니다. 이를 통해 컨슈머 애플리케이션이 RabbitMQ에서 메시지를 읽어오는 작업을 손쉽게 처리할 수 있습니다.
• 메시지 변환
  • RabbitTemplate은 메시지를 객체로 변환하거나 객체를 메시지로 변환하는 작업을 처리합니다. 이는 메시지의 직렬화와 역직렬화를 자동으로 처리하여, 데이터 전송과 수신을 간편하게 합니다.
• 템플릿 메서드 패턴
  • RabbitTemplate은 템플릿 메서드 패턴을 적용하여, RabbitMQ와의 상호작용을 위한 기본적인 작업을 정의합니다.
  • 이 패턴을 통해 사용자는 복잡한 RabbitMQ API를 직접 다루지 않고도 간편하게 메시지를 전송하고 수신할 수 있습니다.

 

 

큐 적재 확인하기 

 

localhost:8080/order/1로 요청을 보내면 컨트롤러에서 반환한 "Order complete!!"가 정상적으로 반환되었습니다.

 

RabbitMQ 관리자 화면에 들어가면, Exchanges에서 market이라는 교환기를 확인할 수 있으며, Queue and Streams에서는 market.product와 market.payment 두 개의 큐가 생성된 것을 볼 수 있습니다.

 

Ready 항목을 보면, 각 큐에 메시지가 발행되었다는 것을 의미합니다. 현재 컨슈머가 없기 때문에 메시지는 큐 내부에 대기 중입니다.

Binding을 보면, 설정에서 지정한 대로 각 큐와 동일한 이름의 라우팅 키가 형성된 것을 확인할 수 있습니다.

 

 

컨슈머 만들기 (Product, Payment)

Payment Application(Consumer)

Payment 설정 파일

 

컨슈머는 큐의 이름만 알고 있으면 되기 때문에, 설정 파일에서는 큐 이름과 RabbitMQ 연결 정보만 지정합니다.

 

PaymentEndpoint

 

PaymentEndpoin 클래스는 RabbitMQ에서 수신한 메시지를 처리하는 컨슈머 컴포넌트를 구현하고 있습니다. 이 클래스는 Spring의 메시지 리스너를 사용하여 큐에서 메시지를 수신하고 로그를 출력합니다.

 

@RabbitListener 어노테이션은 Spring AMQP에서 메시지 큐에서 메시지를 수신하고 처리하는 메서드를 정의하는 데 사용하는 어노테이션입니다. 이 어노테이션을 메서드에 붙이면, 해당 메서드는 지정된 큐에서 오는 메시지를 자동으로 수신하고 처리합니다.

 

이전에는 컨슈머가 없었기 때문에 메시지가 큐에 대기 상태로 남아 있었습니다. 그러나 Payment 컨슈머를 생성하면, market.payment 큐에 적재된 메시지를 가져와서 처리하게 됩니다.

 

 

Payment 애플리케이션을 실행하면 로그가 기록되며, RabbitMQ Listener 함수인 receiveMessage가 동작하여 로그에 메시지가 출력된 것을 확인할 수 있습니다.

 

RabbitMQ 관리자 페이지의 Queues and Streams 섹션에서 확인하면, 원래 market.payment 큐에 적재되어 있던 메시지의 수가 1에서 0으로 변경된 것을 볼 수 있습니다.

 

이 변화는 Consumer 애플리케이션의 Listener가 큐에 적재된 메시지를 읽고 처리했음을 의미합니다.

 

Product Application(Consumer)

Product 설정 파일

ProductEndpoint

---

 

마찬가지로 Product 애플리케이션을 실행하면 market.product 큐에 적재되어 있던 메시지를 Product 애플리케이션의 Listener가 읽어 처리합니다. 이로 인해 로그에 메시지가 출력된 것을 확인할 수 있습니다.

 

 

또한, Queues and Streams 섹션에서 확인해 보면, 원래 market.product 큐에 적재되어 있던 메시지의 수가 1에서 0으로 바뀐 것을 볼 수 있습니다.

 

 

컨슈머가 여러 개 있을 때, 라운드 로빈 방식 확인하기

그렇다면 이번에는 처음 아키텍처에서처럼 Product 컨슈머를 두 개 생성하여, market.product 큐에 메시지를 적재했을 때 Product1과 Product2가 메시지를 어떻게 처리하는지 확인해 보겠습니다.

 

 

order에 요청을 보내는 순간, 컨슈머가 메시지를 읽어 처리했기 때문에 큐에 남아 있는 메시지가 없다는 것을 확인할 수 있습니다.

 

이제 order/1부터 order/5까지 차례대로 요청을 보내보겠습니다.

 

 

그러면 위와 같이 라운드 로빈 방식으로 각 컨슈머가 메시지를 가져가는 것을 확인할 수 있습니다.