MMMQ는 메시지 큐 시스템의 핵심 개념과 동작 원리를 학습하기 위한 프로젝트입니다.
- 단순히 도구를 사용하는 것을 넘어, 메시지 브로커의 존재 이유와 동작 메커니즘을 근본적으로 이해합니다.
- 큐 자료구조부터 메시지 영속화, 동시성 제어까지 바닥부터 직접 쌓아 올립니다.
- 오픈 소스 방식의 협업과 피드백을 통해 '함께 성장하는 소프트웨어'를 지향합니다.
Producer는 메시지를 생성하여 Broker로 전송합니다. Broker는 메시지를 큐에 저장하고, Consumer가 메시지를 요청할 때 이를 전달합니다. 만약 메시지 처리에 실패하면, 메시지는 Dead Letter Queue로 이동하여 별도로 관리됩니다. Dead Letter Queue에 저장된 메시지는 디스크에 나중에 재처리하거나 분석할 수 있습니다.
Producer는 메시지를 생성하고 브로커에게 전송합니다. 토픽과 함께 메시지를 발행할 수 있으며, 전송 실패 시 설정된 횟수만큼 재시도를 수행합니다.
Broker는 Producer로부터 메시지를 받아 Consumer에게 전달합니다. 메시지 수신을 위한 엔드포인트를 제공하며, FrontDispatcher를 통해 메시지를 적절한 Dispatcher에게 전달합니다.
FrontDispatcher는 Broker로 들어온 메시지의 토픽을 분석하여 해당 토픽을 구독 중인 모든 Dispatcher에게 메시지를 분배합니다.
Dispatcher는 메시지 큐를 관리하고 Consumer에게 메시지를 전달합니다. 각 Consumer마다 별도의 Dispatcher가 할당되며, 내부적으로 BlockingQueue를 사용하여 메시지를 관리합니다. 각 Dispatcher는 독립된 스레드 풀을 소유하여 특정 Consumer의 느린 메시지 소비가 다른 Consumer에게 영향을 미치는 느린 소비자 문제를 해결합니다.
DeadLetterQueue는 메시지 전송에 최종적으로 실패했을 때 해당 메시지를 처리합니다. 전송 실패 원인과 함께 메시지를 DeadLetter로 포장하여 저장함으로써, 추후 분석 및 복구를 지원합니다. 재시도 횟수 초과, 처리 불가 예외 발생 등 다양한 실패 상황에 대응합니다.
FrontHandler는 수신한 메시지의 토픽을 분석하여 해당 토픽을 구독 중인 모든 HandlerExecution들을 찾아 실행합니다.
HandlerExecution은 어노테이션 기반(@MMMQListener)과 인터페이스 구현(MMMQHandler) 방식을 모두 동일한 방식으로 처리할 수 있도록 추상화된 메시지 핸들러 실행부입니다. 이를 통해 다양한 방식의 메시지 핸들러 실행부의 일관성을 유지하고 확장성을 높입니다.
@MMMQListener 어노테이션을 사용하면 메서드에 @MMMQListener(topic = "...")를 붙여 메시지 핸들러를 지정할 수 있습니다. Spring의 @EventListener처럼, 선언적으로 특정 토픽의 메시지를 처리할 메서드를 간편하게 등록할 수 있습니다.
MMMQListener 인터페이스를 구현하여 메시지 핸들러를 구현할 수도 있습니다. 컴파일 타임에 타입 안정성을 제공하며, 복잡한 메시지 처리 로직을 구현할 때 유용합니다.
- Java 17 이상
- Spring Boot 3.2.0 이상
- Spring Web (spring-boot-starter-web 의존성 포함)
repositories {
maven { url "https://jitpack.io" }
mavenCentral()
}
dependencies {
...
implementation 'com.github.moko-meringue.mmmq:{모듈}:{버전}'
// Broker 모듈 의존성 추가
// implementation 'com.github.moko-meringue.mmmq:broker:{버전}'
// Consumer 모듈 의존성 추가
// implementation 'com.github.moko-meringue.mmmq:consumer:{버전}'
// Producer 모듈 의존성 추가
//implementation 'com.github.moko-meringue.mmmq:producer:{버전}'
}
Producer는 Broker에 메시지를 전송합니다. Producer Bean을 등록하고 Broker의 주소를 Host 객체로 전달해야 합니다.
@Configuration
public class ProducerConfig {
@Bean
public Producer orderProducer() {
Host brokerHost = new Host("http", "ip", 8080);
return new Producer(brokerHost);
}
}Consumer는 Broker로부터 메시지를 수신하여 처리합니다. @MMMQListener 어노테이션 또는 MMMQHandler 인터페이스를 사용하여 메시지 핸들러를 등록할 수 있습니다.
메서드에 어노테이션을 붙여 간편하게 특정 토픽을 구독하는 핸들러를 만들 수 있습니다.
@Service
public class OrderService {
// ...
@MMMQListener(topic = "new-orders")
public void handleNewOrder(Order order) {
// Handle new order
}
}MMMQHandler 인터페이스를 구현하여 클래스 단위로 핸들러를 정의할 수 있습니다.
@Service
public class OrderService implements MMMQListener<Order>{
// ...
@Override
public String listens() {
return "new-orders";
}
@Override
public void handle(Order order) {
// Handle new order
}
}Broker는 어떤 Consumer에게 메시지를 전달할지 알아야 합니다. Dispatcher Bean을 등록하여 Consumer의 주소와 구독할 토픽을 설정합니다.
@Configuration
public class DispatcherConfig {
@Bean
public Dispatcher orderDispatcher() {
return new Dispatcher(
"order-dispatcher",
new Host("http", "ip", 8080),
Set.of(new Topic("new-orders")),
new TimerDeadLetterQueue(
"order-dead-letter-queue",
new DeadLetterFileWriter(
Path.of("/home/ubuntu/dead-letters"),
"order-dead-letter-writer"
),
60_000
)
);
}
}모든 코드는 모코와 머랭의 페어 프로그래밍으로 작성되었습니다.
이 프로젝트는 교육 목적으로 제작되었습니다.