📨 Spring Boot 3은 MQ를 통합하여 채팅 메시지 저장 시스템 구축

실시간 채팅 서비스를 구축할 때는 메시지가 즉시 전달되고, 쉽게 과거 기록을 조회할 수 있도록 지속적으로 저장되어야 합니다.

그러나 데이터베이스에 동기적으로 데이터를 직접 쓰면 동시성이 높은 시나리오에서 쉽게 성능 병목 현상이 발생하여 메시지의 실시간성에 영향을 미칠 수 있습니다.

"계층을 추가하면 해결할 수 없는 문제는 없다"는 개념을 고수하여 비동기 저장을 위한 메시지 큐(MQ)를 도입하는 것은 우아한 솔루션입니다.

메시지는 즉각적인 전달을 보장하기 위해 먼저 MQ에 빠르게 기록된 다음, 전담 소비자 서비스를 통해 대기열에서 꺼내져 원활하게 데이터베이스에 기록됩니다.

이 글에서는 Spring Boot 3와 메시지 큐 기술을 결합하여 효율적이고 안정적인 채팅 메시지 저장 시스템을 구축하는 방법을 자세히 살펴보겠습니다.

MQ 소개

여기서 MQ의 주요 역할은 채팅 기능을 채팅 콘텐츠의 저장 프로세스로부터 분리하고 분리하는 것입니다.
이 메커니즘은 공장과 도매업체 간의 주문 관계를 최적화하는 것과 매우 유사합니다.
기존 모델에서는 공장에서 제품을 출하할 때마다 각 도매업체에 개별적으로 통지해야 합니다.

MQ가 도입된 후, 이 프로세스는 우아하고 효율적으로 바뀌었습니다.
마치 공장에서 카카오톡 단체톡 메시지만 게시하면 모든 도매업체가 일대일 알림 없이 동시에 정보를 얻을 수 있는 것과 같습니다.
공장은 생산에 집중하고, 도매업체는 수요에 따라 생산을 처리하며, 각 업체는 자체적인 의무를 수행합니다.

서비스 간 통신을 위한 중개자로서 메시지 큐는 분산 시스템에서 중요한 역할을 합니다.
일반적인 솔루션으로는 전문적인 메시지 큐 시스템(예: RabbitMQ, Kafka, RocketMQ 등), 분산 조정 서비스 Zookeeper, Redis 기반의 경량 큐 등이 있습니다.

MQ 선택

많은 메시지 큐 제품마다 고유한 특성과 적용 가능한 시나리오가 있습니다.

메시지 큐	개발 언어	특징	적용 가능한 시나리오
래빗 MQ	고랭	성숙하고 안정적이며 배포가 쉽고 라우팅 기능이 풍부하며 커뮤니티가 활성화되어 있음	복잡한 라우팅 요구 사항, 중소 규모 메시지 볼륨 및 안정성 보장이 필요할 때
액티브 MQ	자바	구식 MQ 및 JMS 구현, 높은 리소스 소모	Java 생태계와 긴밀하게 통합된 기존 엔터프라이즈 애플리케이션
로켓 MQ	자바	높은 처리량, 낮은 지연 시간, 재정적 수준의 안정성 및 대용량 스토리지 지원	대규모 인터넷 애플리케이션 및 금융 결제 시나리오
카프카	스칼라 / 자바	초고처리량, 지속성, 파티션 설계, 스트림 처리에 능숙	로그 수집, 빅데이터 실시간 처리, 스트리밍 데이터 분석
제로 MQ	C++	가볍고 분산화된 내장 라이브러리	성능, 지연 간 통신이 매우 민감한 시나리오
Redis 큐	C	가볍고 간단하며 메모리 기반, 낮은 대기 시간	간단한 시나리오, 임시 대기열 및 낮은 지속성 요구 상황

채팅 메시지 저장 시나리오의 경우, 우리는 주로 다음과 같은 고려 사항을 기반으로 RabbitMQ를 최종적으로 선택했습니다.

1. 성숙하고 안정적입니다: RabbitMQ는 오랜 역사를 가지고 있으며, 실제 운영 환경에서 완벽하게 검증되었으며 안정성이 보장됩니다.

2. 유연한 라우팅: 다양한 유형의 메시지에 대한 정교한 라우팅을 달성하기 위해 다양한 스위치 유형과 바인딩 메커니즘을 제공합니다.

3. 통합이 쉽습니다: Spring 생태계와 긴밀하게 통합된 Spring Boot는 포괄적인 스타터 지원을 제공합니다.

4. 운영 및 유지 관리가 편리함: 간단한 배포, 내장 관리 인터페이스, 모니터링 및 관리 용이

5. 커뮤니티 지원: 활발한 커뮤니티와 풍부한 문서 리소스를 통해 문제가 발생할 때 쉽게 해결책을 찾을 수 있습니다.

Kafka나 RocketMQ가 극도로 높은 동시성 시나리오에서 더 나은 처리량 성능을 보일 수 있지만, 여기서의 초점이 시스템 분리라는 점을 고려하면 RabbitMQ는 이미 개발 및 유지 관리 비용을 줄이는 동시에 요구 사항을 잘 충족할 수 있습니다.

프로그램 시나리오

메시지 큐는 시스템 아키텍처에서 많은 고전적인 응용 프로그램 시나리오를 갖습니다.

• 비동기 처리: 시간이 많이 소요되는 작업(예: 이메일 전송 및 로그 처리)은 메시지 큐를 통해 비동기적으로 처리되어 사용자 요청에 빠르게 응답하고 사용자 경험을 개선합니다.

• 성능 향상: 비동기 분리를 통해 시스템 응답 시간이 단축되고 처리량이 향상되어 특히 I/O 집약적 작업에 적합합니다.

• 시스템 분리: 서비스 간의 직접적인 종속성을 줄이고, 시스템의 탄력성과 유지관리성을 개선하고, 독립적인 확장과 업그레이드를 용이하게 합니다.

• 피크 쉐이빙 및 밸리 필링: 트래픽 피크 동안 메시지 큐는 요청을 캐시하고 처리 용량에 따라 점진적으로 사용하여 시스템 과부하 및 충돌을 방지할 수 있습니다.

채팅 메시지 저장 시나리오에서는 주로 RabbitMQ를 사용하여 비동기 메시지 저장을 구현합니다. 이를 통해 채팅 기능의 응답 속도를 보장할 뿐만 아니라 메시지를 데이터베이스에 안정적으로 보관하고 시스템이 메시지 피크에 대처할 수 있는 기능을 제공합니다.

RabbitMQ 소개

RabbitMQ에서 메시지의 전체 수명 주기는 다음과 같습니다.

1. 프로듀서가 메시지를 생성합니다: 채팅 애플리케이션에서 사용자가 채팅 콘텐츠를 보내고 애플리케이션은 이를 MQ 메시지로 캡슐화합니다.

2. 교환소로 전달: 생산자는 지정된 교환소로 메시지를 보내고 라우팅 키를 지정합니다.

3. 스위치 라우팅 전달: Exchange는 메시지 라우팅 키와 바인딩 규칙에 따라 메시지를 전달할 대기열을 결정합니다.

   • 직접 스위치인 경우 라우팅 키는 정확히 일치합니다.
   • Fanout 스위치인 경우 모든 바운드 큐에 브로드캐스트됩니다.
   • 토픽 스위치인 경우 경로는 패턴에 따라 일치됩니다.

4. 대기열에 저장: 적격 대기열은 메시지를 수신하고 저장하며 소비자가 처리할 때까지 기다립니다.

5. 소비자가 메시지를 받습니다: 스토리지 서비스는 큐에서 메시지를 가져오는 소비자 역할을 하며, 이는 푸시 모드나 풀 모드일 수 있습니다.

6. 확인 처리: 소비자가 메시지를 성공적으로 처리한 후(예: 데이터베이스에 채팅 내용 저장) RabbitMQ에 확인(ACK)을 보냅니다.

7. 메시지 삭제: 확인을 받은 후 RabbitMQ는 큐에서 메시지를 삭제합니다.

RabbitMQ 설치

RabbitMQ는 다양한 방법으로 설치할 수 있으며, Docker는 가장 편리한 배포 솔루션을 제공합니다.
Docker를 사용하여 RabbitMQ를 빠르게 배포하는 단계는 다음과 같습니다.

1. 이미지를 끌어오세요

먼저 Docker Hub에서 RabbitMQ 공식 이미지를 가져옵니다.
management이후의 시각적 작업 및 모니터링을 위한 웹 관리 인터페이스가 포함된 태그가 있는 버전을 선택하는 것이 좋습니다.

docker pull rabbitmq:4.1-management

팁: 독자는 Docker Hub를 방문하여 연습할 때 최신 버전을 보고 사용할 수 있습니다.

2. 컨테이너를 시작하세요

이미지를 가져온 후 다음 명령으로 RabbitMQ 컨테이너를 시작합니다.

docker run --name rabbitmq -p 5681:5671 -p 5682:5672 -p 4379:4369 -p 15681:15671 -p 15682:15672 -p 25682:25672 --restart always -d rabbitmq:4.1-management

여기서 우리는 다음과 같은 매핑과 구성을 만들었습니다.

• AMQP 포트(5672)와 관리 인터페이스 포트(15672)를 노출합니다.
• RabbitMQ가 서버가 다시 시작된 후 자동으로 시작될 수 있도록 컨테이너를 자동으로 다시 시작하도록 구성합니다(--restart always).
• 컨테이너를 백그라운드에서 실행합니다(-d)

3. 설치 확인

성공적으로 시작한 후 브라우저에서 http://127.0.0.1:15682 RabbitMQ 관리 콘솔에 액세스하여 엽니다.

기본 사용자 이름과 비밀번호(둘 다 guest)를 사용하여 로그인하세요.

로그인에 성공하면 RabbitMQ 관리 인터페이스가 나타납니다.
여기서 스위치와 대기열을 생성하고, 연결 상태를 보고, 메시지 처리량과 같은 중요한 지표를 모니터링할 수 있습니다.

참고: 기본 게스트 사용자는 로컬호스트에서만 접근할 수 있습니다.
원격 액세스가 필요한 경우 새로운 관리자 사용자를 만들고 적절한 권한을 설정하는 것이 좋습니다.

Spring Boot는 RabbitMQ를 통합합니다

시작하기에 앞서 메시지 테이블을 만들어 보겠습니다.

이 문서의 채팅 서비스는 이전 문서 "Java 엔지니어 고급 필수 사항: Spring Boot 3 + Netty를 이용한 높은 동시성 인스턴트 메시징 서비스 구축"을 기반으로 합니다.

관심 있는 독자는 확인해 보세요.

1. chat_message — 메시지 기본 정보

DROP TABLE IF EXISTS `chat_message`;
CREATE TABLE `chat_message` (
  `id` VARCHAR(32) NOT NULL COMMENT '메시지 고유 ID',
  `sender_id` VARCHAR(32) NOT NULL COMMENT '보낸 사람 사용자 ID',
  `receiver_id` VARCHAR(32) NOT NULL COMMENT '받는 사람 사용자 ID',
  `receiver_type` INT NULL DEFAULT NULL COMMENT '수신자 유형 (예: 사용자, 그룹 등)',
  `msg_type` INT NOT NULL COMMENT '메시지 유형 (예: 텍스트, 이미지, 비디오 등)',
  `msg` VARCHAR(255) NOT NULL COMMENT '텍스트 메시지 내용',
  `chat_time` DATETIME NOT NULL COMMENT '채팅 시간 (보낸 시간과 받은 시간 동일)',
  `show_msg_date_time_flag` INT NULL DEFAULT NULL COMMENT '메시지 시간 표시 여부 플래그 (1분 이상 간격용)',
  `is_read` TINYINT(1) NULL DEFAULT NULL COMMENT '읽음 여부 (1: 읽음, 0: 안 읽음)',
  PRIMARY KEY (`id`)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '기본 채팅 메시지 테이블';

2. chat_message_video — 메시지 타입이 비디오일 때 상세 정보

DROP TABLE IF EXISTS `chat_message_video`;
CREATE TABLE `chat_message_video` (
  `message_id` VARCHAR(32) NOT NULL COMMENT '연결된 메시지 ID',
  `video_path` VARCHAR(128) NULL DEFAULT NULL COMMENT '비디오 파일 경로',
  `video_width` INT NULL DEFAULT NULL COMMENT '비디오 폭(px)',
  `video_height` INT NULL DEFAULT NULL COMMENT '비디오 높이(px)',
  `video_times` INT NULL DEFAULT NULL COMMENT '비디오 재생 시간(초)',
  PRIMARY KEY (`message_id`),
  CONSTRAINT `fk_video_message` FOREIGN KEY (`message_id`) REFERENCES `chat_message`(`id`) ON DELETE CASCADE
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '비디오 메시지 상세 테이블';

3. chat_message_voice — 메시지 타입이 음성일 때 상세 정보

DROP TABLE IF EXISTS `chat_message_voice`;
CREATE TABLE `chat_message_voice` (
  `message_id` VARCHAR(32) NOT NULL COMMENT '연결된 메시지 ID',
  `voice_path` VARCHAR(128) NULL DEFAULT NULL COMMENT '음성 파일 경로',
  `speak_voice_duration` INT NULL DEFAULT NULL COMMENT '음성 재생 시간(초)',
  PRIMARY KEY (`message_id`),
  CONSTRAINT `fk_voice_message` FOREIGN KEY (`message_id`) REFERENCES `chat_message`(`id`) ON DELETE CASCADE
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = '음성 메시지 상세 테이블';

종속성 가져오기

먼저, pom.xml 프로젝트의 저장소 파일에 RabbitMQ 종속성을 추가합니다.

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

구성 추가

spring:  
  rabbitmq:
    host: 127.0.0.1
    port: 5682
    username: guest
    password: guest
    virtual-host: /

작가 프로듀서

RabbitMQ에 메시지를 보내기 위한 메시지 게시자 클래스를 만듭니다.

import com.sleekydz.model.netty.ChatMsg;
import com.sleekydz.utils.JsonUtils;

public class MessagePublisher {

    // 교환기 이름 정의
    public static final String EXCHANGE = "sleekydz86_exchange";

    // 큐 이름 정의
    public static final String QUEUE = "sleekydz86_queue";

    // 메시지를 큐로 전송하고 데이터베이스에 저장하기 위한 라우팅 키
    public static final String ROUTING_KEY_SEND = "sleekydz86.kakao.send";

    // 메시지를 전송하고 저장하는 메서드
    public static void sendMsgToSave(ChatMsg msg) throws Exception {
        RabbitMQConnectUtils connectUtils = new RabbitMQConnectUtils();
        connectUtils.sendMsg(JsonUtils.objectToJson(msg),
                EXCHANGE,
                ROUTING_KEY_SEND);
    }
}

메시지 전송을 위한 도구 클래스 작성

import com.rabbitmq.client.*;

import java.util.ArrayList;
import java.util.List;

public class RabbitMQConnectUtils {

    private final List<Connection> connections = new ArrayList<>();
    private final int maxConnection = 20;

    // 개발환경 dev
    private final String host = "59.45.23.1";
    private final int port = 5682;
    private final String username = "guest";
    private final String password = "guest";
    private final String virtualHost = "/";

    public ConnectionFactory factory;

    public ConnectionFactory getRabbitMqConnection() {
        return getFactory();
    }

    public ConnectionFactory getFactory() {
        initFactory();
        return factory;
    }

    private void initFactory() {
        try {
            if (factory == null) {
                factory = new ConnectionFactory();
                factory.setHost(host);
                factory.setPort(port);
                factory.setUsername(username);
                factory.setPassword(password);
                factory.setVirtualHost(virtualHost);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void sendMsg(String message, String queue) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish("",
                            queue,
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBytes("utf-8"));
        channel.close();
        setConnection(connection);
    }

    public void sendMsg(String message, String exchange, String routingKey) throws Exception {
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.basicPublish(exchange,
                            routingKey,
                            MessageProperties.PERSISTENT_TEXT_PLAIN,
                            message.getBytes("utf-8"));
        channel.close();
        setConnection(connection);
    }

    public GetResponse basicGet(String queue, boolean autoAck) throws Exception {
        GetResponse getResponse = null;
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        getResponse = channel.basicGet(queue, autoAck);
        channel.close();
        setConnection(connection);
        return getResponse;
    }

    public Connection getConnection() throws Exception {
        return getAndSetConnection(true, null);
    }

    public void setConnection(Connection connection) throws Exception {
        getAndSetConnection(false, connection);
    }

    private synchronized Connection getAndSetConnection(boolean isGet, Connection connection) throws Exception {
        getRabbitMqConnection();

        if (isGet) {
            if (connections.isEmpty()) {
                return factory.newConnection();
            }
            Connection newConnection = connections.get(0);
            connections.remove(0);
            if (newConnection.isOpen()) {
                return newConnection;
            } else {
                return factory.newConnection();
            }
        } else {
            if (connections.size() < maxConnection) {
                connections.add(connection);
            }
            return null;
        }
    }

}

소비자 쓰기

메시지를 수신하고 처리하기 위한 메시지 소비자 클래스를 만듭니다.

import com.sleekydz.model.netty.ChatMsg;
import com.sleekydz.service.ChatMessageService;
import com.sleekydz.utils.JsonUtils;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

// 메시지 소비자 클래스
@Component
@Slf4j
public class RabbitMQConsumer {

    @Resource
    private ChatMessageService chatMessageService;

    // 지정된 큐(RabbitMQConfig.QUEUE)에서 메시지를 수신하는 메서드
    @RabbitListener(queues = {RabbitMQConfig.QUEUE})
    public void watchQueue(String payload, Message message) {
        String routingKey = message.getMessageProperties().getReceivedRoutingKey();
        log.info("수신된 routingKey = {}", routingKey);

        // 메시지 발신 라우팅 키에 해당하는 경우 처리
        if (routingKey.equals(RabbitMQConfig.ROUTING_KEY_SEND)) {
            String msg = payload;
            ChatMsg chatMsg = JsonUtils.jsonToPojo(msg, ChatMsg.class);

            // 메시지를 DB에 저장
            chatMessageService.saveMsg(chatMsg);
        }
    }
}

정리

RabbitMQ를 Spring Boot와 통합하여 비동기 메시지 처리 메커니즘을 구현하고, 채팅 메시지의 저장 작업을 분리하고, 시스템의 성능과 확장성을 개선했습니다.

사용자가 메시지를 보내면 해당 메시지를 RabbitMQ로 보내고, 이 메시지는 소비자에 의해 비동기적으로 처리되어 데이터베이스에 저장됩니다.

이를 통해 데이터베이스를 직접 운영함으로써 발생하는 성능 병목 현상을 방지할 수 있습니다.