[내일배움캠프 75일차] 메일 발송 MessageQueue 적용하기

MessageQueue란?

메시지 지향 미들웨어로 임시로 데이터를 저장하는 큐 형태의 버퍼 역할을 한다.
프로세스 또는 프로그램 간에 데이터를 교환할 때 사용하는 통신 방법이다.

Queue에 데이터를 입력하는 주체를 Producer, 소비하는 주체를 Consumer라고 한다.

메일 발송에 MessageQueue를 도입해야 하는 이유

지난 글에서 잠시 언급했던 것에 이어서 좀 더 설명을 보충하고자 한다.

💡프로젝트에 적합하다.

Message Queue는 바로 처리되지 않더라도 언젠가 처리되어야 할 작업들에 적용할 수 있다.

문제풀이 링크 메일 발송은 정해진 시간에 딱 맞춰 빠르게 보내지는 것보다는
몇 분 늦더라도 정확하게 도착하는 것이 중요하기 때문에 적합하다고 판단하였다.

 

💡신뢰도 보완

현재까지 구현된 알고리즘은 중간에 메일 전송에 실패하거나 서버가 다운될 때
별도의 대처 방안이 없어 메일 발송에 대한 신뢰도가 떨어진다. 

1000개의 메일을 발송할 때, 문제 출제 -> 메일 발송 이 과정을 순차적으로 반복하기 때문에
526번째 메일에서 서버가 다운된다면, 이후 메일 발송에 대한 정보는 손실되어 복구가 어렵다.

어디까지 처리되었고 어디부터 다시 처리해야 하는지를 추적할 수 없다.

MessageQueue 도입 시,
서버가 다운되더라도 Redis 서버가 정상 동작하는 한, Queue에 저장된 메시지는 유지된다.
따라서, 발생한 문제를 해결한 후 Queue에 남아있는 데이터에 대해 처리할 수 있다.

 

💡속도 개선의 가능성

앞서 몇 분 늦더라도 정확하게 도착하는 것이 더 중요하다고 언급하였지만
속도가 개선된다면, 사용자 경험 측면에서도 프로젝트 성능 측면에서도 긍정적일 것이라고 생각한다.

MessageQueue를 도입하면 비동기 처리를 도입할 수 있으므로 속도를 향상할 가능성이 높아진다.

 

단점

• 메시지 큐 운영 및 관리를 위한 관리 비용 증가
• 시스템 구조의 복잡성 증가
• 오버헤드 발생 가능성

 

트레이드오프

단점과의 트레이드오프를 고려하더라도 프로젝트 특성상, 문제를 풀 수 있는 수단을 메일로 받는다는 점에서
메일 발송은 중요한 수단이기 때문에 이를 감수하더라도 MessageQueue를 도입할 가치가 있다고 판단하였다.

---

Redis Streams 사용

MessageQueue 기능을 제공하는 프로그램들은 Kafka, RabbitMQ 등 다양하게 있다.

그중에서 Redis Streams를 사용하여 구현하고자 한 이유는 다음과 같다.

• 이미 Redis를 사용 중이었기 때문에 접근성이 매우 높다.
• docker랑 CI/CD 설정을 굳이 변경하고 싶지 않다.

어떤 기능을 위한 프로젝트를 추가하기 위해서 docker와 CI/CD 설정을 변경할 때마다 자꾸 뭐가 안 돼서 함께 고생을 했었다.
따라서, 굳이 설정을 추가 및 변경하지 않고도 기능을 도입할 수 있는 Redis를 가급적이면 사용하고 싶었다.

또한, 예상되는 프로젝트 규모가 Redis가 감당할 수 있는 정도라고 생각했기 때문에 Redis Streams를 사용하였다.

---

MessageQueue 구현

기존에 구현된 기능들을 기반으로 MessageQueue를 적용했을 때, 구조는 위의 이미지와 같다.

Producer

Spring Batch의 MailJob이 Producer의 역할을 맡아
데이터를 Redis Streams의 Message Queue에 넣는다.

public void enqueueQuizEmail(Subscription subscription, Quiz quiz) {
    Map<String, String> data = new HashMap<>();
    data.put("email", subscription.getEmail());
    data.put("subscriptionId", subscription.getId().toString());
    data.put("quizId", quiz.getId().toString());

    redisTemplate.opsForStream().add("quiz-email-stream", data);
}

이 때, 메일 내용 구성 및 메일 발송 로그 생성에 필요한 Subscriptio, Quiz 엔티티는 
Redis의 데이터 특성상 Id가 문자열 형태로 직렬화 되어 저장된다.

 

Consumer

Consumer 역시 Spring Batch가 맡는다.
Spring Batch는 다른 글에서 자세히 다룰 것이기 때문에 간단한 구조만 설명하자면

https://terasoluna-batch.github.io/guideline/5.0.0.RELEASE/en/Ch02_SpringBatchArchitecture.html

Step은 ItemReader → ItemProcessor → ItemWriter 구조로 동작한다.

📖 ItemReader

ItemReader는 데이터를 읽어온다.
여기서는 Message Queue에 저장된 데이터를 꺼내서 읽는 역할을 수행하도록 Custom하였다.

public Map<String, String> read() {Add commentMore actions
    List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream()
        .read(StreamOffset.fromStart("quiz-email-stream"));

    MapRecord<String, Object, Object> msg = records.get(0);
    redisTemplate.opsForStream().delete("quiz-email-stream", msg.getId());

 

🔨 ItemProcessor

ItemProcessor에서는 메일 내용 구성과 메일 발송 로그 생성에 필요한 데이터를 실제로 조회한 후 Dto를 생성하여 넘긴다.

@Override
public MailDto process(Map<String, String> message) throws Exception {
    Long subscriptionId = Long.valueOf(message.get("subscriptionId"));
    Long quizId = Long.valueOf(message.get("quizId"));

    Subscription subscription = subscriptionRepository.findByIdOrElseThrow(subscriptionId);
    Quiz quiz = quizRepository.findById(quizId).orElseThrow(() -> new QuizException(QuizExceptionCode.NOT_FOUND_ERROR));

    return new MailDto(subscription, quiz);
}

 

✏️ ItemWriter

ItemProcessor에서 만든 MailDto로 실제 메일을 발송한다.

@Override
public void write(Chunk<? extends MailDto> items) throws Exception {
    for (MailDto mail : items) {
        try {
            mailService.sendQuizEmail(mail.subscription(), mail.quiz());
        } catch (Exception e) {
            // 에러 로깅 또는 알림 처리
            System.err.println("메일 발송 실패: " + e.getMessage());
        }
    }
}