RabbitMQ와 Spring Boot로 구현하는 Auction Server

• Auction Server는 입찰, 낙찰, 즉시 구매 등의 다양한 이벤트를 처리하는 시스템으로, 안정적인 동작을 위해 주로 비동기적 방식으로 처리됩니다. 예를 들어, 사용자가 입찰을 진행할 때마다 이를 즉시 처리하는 대신, RabbitMQ와 같은 메시지 큐 시스템에 메시지를 추가하고, 서버의 부하가 적은 시점에 해당 메시지를 처리합니다. 이를 통해 시스템 과부하를 방지하고, 사용자는 대기 시간 없이 다른 작업을 진행할 수 있습니다.
• 특히, 인기 상품의 경매와 같은 이벤트에서는 급격한 트래픽 증가가 예상되므로, RabbitMQ는 메시지를 안정적으로 큐에 보관하고, 서버의 부하가 적을 때 후처리하여 부하를 분산시킵니다. 이로 인해 트래픽 폭주 시에도 시스템이 안정적으로 작동할 수 있습니다.
• Spring Boot는 이와 같은 비동기적 메시지 처리 시스템을 구현하는 데 이상적인 프레임워크로, RabbitMQ와의 통합을 통해 효율적인 메시지 큐 처리 및 유효성 검사 로직을 구현할 수 있습니다. 예를 들어, 입찰 시에는 유효성 검사(경매 최소 금액, 입찰 금액 비교 등)를 통해 메시지가 큐에 Enqueue되며, Dequeue 시점에서 재차 유효성 검사를 수행하여 처리의 신뢰성을 높입니다.

---

메시지 큐(Message Queue)

• 프로세스 또는 프로그램 간에 데이터를 교환할 때 사용하는 통신 방법 중에 하나
• 메시지 지향 미들웨어(Message Oriented Middleware:MOM)를 구현한 시스템을 의미
  • MOM은 비동기 메세지를 사용하는 응용 프로그램 간의 데이터 송수신을 의미합니다.

• 송신자(Sender)와 수신자(Receiver) 사이에 비동기적으로 메시지를 교환하는 솔루션
• 메세지 큐를 사용하는 예시로는 이메일 전송, 구독 서비스등이 있습니다. 

메시지 큐의 특징

• 비동기 통신
  • 메시지 큐는 송신자와 수신자 간에 독립적으로 메시지를 교환하므로 비동기적인 통신을 제공합니다.
  • 이는 송신자와 수신자가 실시간으로 상호작용하지 않고도 메시지를 보낼 수 있음을 의미합니다.
• 메시지 보관
  • 메시지 큐는 메시지를 안전하게 보관합니다.
  • 송신자가 메시지를 보내면 메시지 큐가 해당 메시지를 저장하고, 수신자가 메시지를 받을 준비가 되면 메시지를 전달합니다. 이로 인해 메시지 손실을 방지하고, 송신자와 수신자 간의 비동기 통신을 지원합니다.
• 대용량 및 확장성
  • 메시지 큐는 대용량의 메시지를 처리하고 처리량을 확장할 수 있는 능력을 갖추고 있습니다.
  • 이는 분산 아키텍처와 클러스터링을 통해 구현될 수 있습니다.

---

주요한 메시지 큐 솔루션에는 다음과 같은 종류가 있습니다.

RabbitMQ

• AMQP(Advanced Message Queuing Protocol)를 지원하는 오픈 소스 메시지 브로커입니다.
• 다양한 메시징 패턴을 지원하며, 유연성과 확장성이 높습니다.
• 장점
  • AMQP 프로토콜을 지원
  • 다양한 메시징 패턴과 프로토콜 변환을 지원
  • 유연성과 확장성이 뛰어남
  • 다양한 클라이언트 라이브러리를 제공
• 단점
  
  • 메시지를 디스크에 저장하는 것이 아니라 메모리에서 처리하기 때문에 대용량 데이터 처리에는 제한이 있을 수 있습니다.
  • 높은 처리량이 요구되는 경우 성능에 제약이 있을 수 있습니다.

Amazon Simple Queue Service (SQS)

• 아마존 웹 서비스(AWS)의 완전 관리형 메시지 큐 서비스입니다.
• 신뢰성, 확장성, 내구성 및 쉬운 운영을 제공합니다.
• 장점
  • 완전 관리형 서비스로서 운영 및 관리가 용이합니다.
  • 고가용성, 확장성 및 내구성이 뛰어나며, AWS 생태계와의 통합이 원활합니다.
• 단점
  • 지연 시간이 상대적으로 길 수 있고, 요청당 비용이 발생할 수 있습니다.
  • 일부 고급 기능은 다른 메시지 큐와 비교하여 제한적일 수 있습니다.

---

경매 서버 프로젝트의 요구 사항을 충족하기 위해, 메시지 보장, 확장성, 안정성, 메시지 순서 보장, 라우팅 기능, 다양한 프로토콜 지원, 풍부한 문서화, 오픈 소스와 같은 다양한 이점을 고려하여 RabbitMQ을 사용하였습니다. 

RabbitMQ을 사용하기 위해서는 다음과 같은 핵심 개념을 알아야 합니다.

출처 : https://yoonbing9.tistory.com/129

Producer(생산자)

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

Exchange(교환기)

• Exchange는 생산자가 메시지를 어떤 큐로 보낼지 결정하며, 다양한 라우팅 알고리즘과 방식을 제공하여 메시지를 적절한 큐로 보내는 역할을 합니다.
• Exchange 종류
  • Direct Exchange
    • Direct Exchange는 메시지의 라우팅 키와 바인딩 키가 정확하게 일치하는 큐로 메시지를 전송합니다.
    • 이 Exchange는 1:1 라우팅에 적합합니다.
    • ex) 서로 다른 로그 유형 ( 애플리케이션 로그, 보안 로그, 성능 로그 등)을 분리하여 처리할 때
  • Fanout  Exchange
    • Fanout Exchange는 바인딩된 모든 큐로 메시지를 브로드캐스트합니다.
    • 메시지를 받는 모든 큐에 복사본이 전달됩니다.
    • 이 Exchange는 1:N 라우팅에 적합합니다.
    • ex) 이벤트 관리 플랫폼에서 모든 사용자에게 알림을 전달할 때
  • Topic Exchange
    • Topic Exchange는 메시지의 라우팅 키와 바인딩 키를 패턴으로 매칭하여 메시지를 전달합니다.
    • 와일드카드(*, #)를 사용하여 유연한 라우팅을 지원합니다.
    • 이 Exchange는 다양한 라우팅 패턴에 적합합니다.
    • ex) 특정 주제에 관한 내용을 구독하는 사용자에게 관련 이벤트 업데이트를 제공할 때

Queue(큐)

• 메시지가 저장되는 장소로, Consumer가 메시지를 가져옵니다.

Binding(바인딩)

• Exchange와 Queue 사이의 연결을 나타냅니다. 이 연결은 어떤 Exchange에서 어떤 Queue로 메시지를 보낼지를 정의합니다.

Consumer(소비자)

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

---

RabbitMQ 설치

현재 환경은 windows에서 진행하고 있습니다. windows는 RabbitMQ 공식 홈페이지를 통해 다운로드 받을수 있습니다. 

설치 프로그램을 실행하여 과정을 거치면 사전 준비는 완료되었습니다.

설치 확인 방법은 http://localhost:15672/#/ 을 통해 확인할 수 있으며, 기본 ID/Password는 guest입니다.

 

경매 서버에서는 입찰이 발생할 때 순서를 보장하여 처리해야 합니다. 예를 들어, 사용자가 먼저 입찰을 하였는데 후에 들어오는 입찰 금액이 먼저 처리되면, 먼저 입찰한 금액이 제대로 처리되지 않을 수 있습니다. 이런 문제를 해결하기 위해 RabbitMQ의 큐에서는 순서를 보장하는 방법으로 단일 큐 사용, 우선순위 큐 사용, 컨슈머를 통한 메시지 시퀀스 번호로 정렬 등이 있습니다. 이 중 단일 큐와 우선순위 큐를 선택하여 순서를 보장했습니다. 단일 큐를 사용하면 한 곳에서 모든 작업을 관리할 수 있고, 선입선출(FIFO) 순서로 처리되어 순서가 보장됩니다. 여러 큐를 사용해도 순서를 보장할 수 있지만, 현재는 단일 큐를 우선 사용하고 추후 성능테스트를 통해 큐의 개수를 늘려가며 비교할 계획입니다.

 

우선순위 큐는 애플리케이션에서 받는 시간을 토대로 우선순위를 설정하였으며, 이는 밀리세컨드까지 계산하여 과거 순으로 정렬하여 메시지를 처리 하도록 설정하였습니다.

 

Exchange는 1:1 라우팅에 적합한 Direct Exchange로 설정하였습니다. 이는 추후 여러 큐를 사용할 수 있으며, 큐 이름과 정확히 일치하는 Exchange를 사용하여  확장성을 고려하였습니다. Routing key는 auction.key로 설정하였으며, 이 Exchange에서는 Binding key를 따로 설정할 필요가 없으며, 정확한 이름으로 일치하여 사용해야하므로 Routing key로 대체할 수 있습니다.

---

build.gradle

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

RabbitMQ을 사용하기 위해서 의존성을 추가 해줍니다.

Spring AMQP 모듈은 AMQP와의 통합을 지원하는 Spring 프로젝트로 RabbitMQ 클라이언트 라이브러리를 포함하고 있어 RabbitMQ와의 통합을 용이하게 합니다.

# application.properties RabbitMQ 연결 설정
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

rabbitmq.queue.name =  auction.queue
rabbitmq.exchange.name = auction.exchange
rabbitmq.routing.key = auction.key

Exchange 이름은 커스텀한 이름으로 설정되며, Routing key는 Direct Exchange를 사용할 때 Exchange 이름과 일치하도록 설정해야 합니다. 만약 일치하지 않을 경우, 메시지가 큐로 전달되지 않을 수 있습니다.

 

 

Config

RabbitMQ의 구성 파일을 작성했습니다. 설정은 미리 작성한 application.properties 파일의 정보를 기반으로 설정하였습니다.

 

RabbitMQ는 JSON 형식으로 전송되는 메시지의 특성상 직렬화와 역직렬화 시 LocalDateTime 변환으로 인한 문제가 발생하였습니다. 이를 해결하기 위해 Jackson2JsonMessageConverter를 메시지 변환기로 추가하였으며, ObjectMapper.registerModule에 JavaTimeModule을 추가 하여 이러한 문제를 해결했습니다.

---

RabbitMQServiceImpl

@RabbitListener 어노테이션은 Spring AMQP 프로젝트에서 제공하는 메시지 리스닝을 쉽게 구현할 수 있도록 지원하는 어노테이션입니다. 이 어노테이션을 사용하면 메시지 큐에서 메시지를 받아들이고 처리하는 메서드를 간단하게 작성할 수 있습니다.

 

입찰 가격의 유효성 검사를 통해 메시지 처리를 수행했습니다. 이는 메시지가 수신될 때와 처리될 때 해당 상품의 최고 입찰 가격이 다를 수 있기 때문에 유효성 검사를 두 번 진행했습니다. 만약 가격에 대한 유효성 검사가 실패하면 커스텀 예외(InputMismatchException)가 발생합니다. RabbitMQ는 기본적으로 예외가 발생하면 메시지를 재처리하도록 설정되어 있어 해당 예외가 발생한 메시지가 재처리되어 메시지 처리시 무한 루프가 발생하는 문제가 있었습니다.이 문제를 해결하기 위해 해당 예외가 발생할 때 channel.basicReject(tag, false); 구문을 통해 메시지를 재처리하지 않도록 flase로 설정하여 문제를 해결했습니다. 해당 구문은 channel은 큐와 통신하기 위한 통로로 메시지 송수신, 큐 및 교환기 생성 및 삭제 등의 작업을 처리하는 용도로 사용됩니다. 또한, tag는 @Header(AmqpHeaders.DELIVERY_TAG)을 통해 헤더의 정보 중 DELIVERY_TAG으로 메시지에는 고유한 delivery tag가 할당되며 이를 가져와 사용하여 메시지를 확인하거나 거부할 수 있습니다.

 

결과

• RabbitMQ에게 JSON형태로 메시지를 보내게 됩니다. 연결하여 보내고 받는 과정에서 문제가 발생했습니다. LocalDateTime의 날짜 문제가 발생하여 이를 해결하기 위해 자료조사를 통해 "objectMapper.registerModule(new JavaTimeModule());" 구문을 추가 하여 해결하였습니다.
• 메세지를 처리 하는 과정에서 예외가 발생하며 무한 반복하는 문제가 발생하였습니다. 디버깅으로 확인한 결과 유효성 검사로 인한 예외 발생시 큐에 메세지가 남아 처리 되지 않은 문제였습니다. 이를 유효성 검사에서 발생하는 커스텀 예외로 catch을 하여 "channel.basicReject(tag, false);" 구문을 처리 하여 큐에 메시지를 재전송되지 않도록 수정하였습니다.

---

참고

GitHub - gamsayeon/Auction-Server

• https://github.com/gamsayeon/Auction-Server
• https://chamggae.tistory.com/164
• https://yoonbing9.tistory.com/129
• https://www.rabbitmq.com/