[Spring] WebSocket 메시지 처리 효율화: Redis 기반 비동기 저장

✅ 개요

Spring WebSocket과 STOMP, 그리고 Redis Pub/Sub 기반의 1대1 채팅 기능을 구현하였습니다. 그리고 기존에는 다음과 같이 동기적으로 `@MessageMapping`에서 메시지를 받자마자 RDBMS에 저장한 다음 Redis에 발행하는 구조를 가졌습니다.

동기적으로 메시지를 DB에 저장

이러한 구조는 `1 메시지 + 1 DB 커넥션`으로 인해 전체적인 메시징 응답 속도를 저하시켜 병목 현상을 야기할 수 있겠다는 생각이 들었습니다. 이를 해결하기 위해 Redis의 List 자료구조를 메시지 큐로 사용해, 메시지 저장을 비동기로 분리하여 개선하는 과정을 정하고자 합니다.

 

✅ 메시지 큐란?

본 포스팅은 메시지 큐의 자세한 설명을 위한 글이 아니므로 간단하게 개념만 정리해보겠습니다.

 

메시지 큐(Message Queue, MQ)는 프로세스 또는 프로그램 간에 데이터를 교환할 때 사용하는 통신 방법 중 하나로, 메시지를 파이프라인에 임시로 저장해두었다가 나중에 처리할 수 있게 하는 비동기 통신 메커니즘을 의미합니다.

 

메시지 큐의 핵심 개념

• 생산자(Producer) : 데이터를 생성하여 큐에 보내는 주체
• 큐(Queue) : 데이터가 저장되는 임시 저장소
• 소비자(Consumer) : 큐에서 데이터를 가져와 실제 처리를 담당하는 주체

비유) 우체통 : 우체통(큐)에 편지(데이터)를 넣어두면(생산), 나중에 우체부가 수거하여 배달(소비)하는 방식

 

✔️ 메시지 큐 종류와 특징

메시지 큐의 종류는 다양하게 있는데, 그 중 많이 쓰이는 것 같은 RabbitMQ, Kafka, Redis에 대해 정리해 보겠습니다.

	RabbitMQ	Apache Kafka	Redis (List/Stream)
처리 성능	중	최상	최상
주요 특징	- 데이터 전달 보장 - 관리 UI 제공	- 고처리량 - 파티셔닝 지원 - 데이터 영속성	- 인메모리 기반 초고속 처리 - 메모리 제한 및 데이터 유실 가능성
사용 사례	- 적당한 규모 - 복잡한 메시징 시나리오	- 대규모 실시간 데이터 파이프라인 - 로그 수집	- 가벼운 메시징 - 영속성보다는 속도가 중요

 

✔️ Redis를 선택한 이유

우선 RabbitMQ나 Kafka의 경우 높은 학습 곡선과 추가적인 인프라 구축 비용을 필요로 하기 때문에, 개인 프로젝트 규모를 생각해보면 오버 엔지니어링이 될 것 같았습니다.

Redis는 설치나 사용 방법이 간단하고 이미 프로젝트에서 Pub/Sub이나 캐싱 처리에 사용하고 있기 때문에 적절하다고 생각했습니다. 하지만 Redis는 데이터 유실 가능성이 존재하기 때문에 추후 RabbitMQ나 Kafka를 학습하며 적용해 보는 것도 좋을 것 같습니다.

 

✅ Redis를 메시지 큐로 활용하는 방법

Redis의 자료 구조 중 하나인 list는 큐로 사용하기 적절한 자료 구조입니다. 큐의 tail과 head에서 데이터를 넣고 뺄 수 있는 `LPUSH`, `LPOP`, `RPUSH`, `RPOP` 커맨드를 사용해 메시지 큐를 직접 구현할 수 있습니다.

 

Redis 기반 메시지 큐를 사용하여 최종 구현하고자 하는 구조는 다음과 같습니다.

메시지 큐를 이용한 비동기 처리

주요 특징은 서버는 메시지를 받고 나서 메시지 큐에 임시 저장과 메시지 발행 후 곧바로 응답을 한다는 것입니다. DB INSERT 작업은 주기적으로 실행되는 배치 작업에 의해 일괄 저장되기 때문에 `1 메시지 + 1 DB 커넥션` 구조를 크게 개선할 수 있을 것입니다.

 

핵심 로직은 다음과 같습니다.

public void addMessageToQueueAndCache(Long roomId, ChatMessageResDto message) {
	//메시지 큐 저장(RPUSH)
    redisTemplate.opsForList().rightPush("chat:queue", message);

	//캐시 저장
    String cacheKey = "chat:cache:" + roomId;
    redisTemplate.opsForList().leftPush(cacheKey, message);
    redisTemplate.opsForList().trim(cacheKey, 0, 99);
}

 

먼저 메시지를 list에 저장합니다. 여기서 저장한 메시지는 배치 처리 로직에서 순서대로 꺼내서 저장하기 때문에 `RPUSH` 후 `LPOP` 또는 `LPUSH` 후 `RPOP` 순서로 가면 될 것 같습니다.

 

추가로 캐시에 메시지를 저장하는 이유는 데이터 정합성을 위함입니다. 채팅방 입장 후 DB에 저장된 전체 메시지들을 화면에 뿌려주기 위해 DB 조회를 할 텐데, 비동기 배치 작업이 이루어지기 전 새로고침이나 나갔다 오는 경우 메시지를 즉시 조회하지 못하는 경우가 발생할 수 있습니다. 따라서 DB + 캐시 조회를 통해 완전한 전체 메시지를 응답합니다. 다음은 그 로직입니다.

/**
 * 메시지 기록 조회(커서 기반)
 *
 * @param lastMessageId 마지막 조회 메시지
 * @param roomId        채팅방
 * @param pageSize      조회 개수
 */
public ChatMessagesResDto getMessageHistories(Long lastMessageId, Long roomId, int pageSize) {
	//DB 저장된 메시지 목록
    List<ChatMessageResDto> fetchedMessages = chatRepositoryFacade.getMessages(lastMessageId, roomId, pageSize);
    //DB + 캐시 = 최종 반환할 메시지 목록
    List<ChatMessageResDto> resultMessages = new ArrayList<>(fetchedMessages);

    if (lastMessageId == null) {
    	//캐시에 저장된 메시지 목록
        List<Object> cachedRaw = chatRedisService.getCachedRaw(roomId);

        if (cachedRaw != null && !cachedRaw.isEmpty()) {
            List<ChatMessageResDto> cachedData =
                    cachedRaw.stream()
                             .map(chatRedisService::convert)
                             .filter(m -> fetchedMessages.isEmpty() || m.getTimestamp().isAfter(fetchedMessages.get(0).getTimestamp()))
                             .toList();
                             
			//캐시가 더 최근 메시지이기 때문에 앞에 저장
            resultMessages.addAll(0, cachedData);
        }
    }

    boolean hasMore = resultMessages.size() > pageSize;

    if (hasMore) {
        resultMessages.remove(resultMessages.size() - 1);
    }

    return new ChatMessagesResDto(resultMessages, hasMore);
}

다음 핵심 로직은 배치 작업 입니다. 정기 배치 작업을 통해 사용자의 메시지를 잊지 않고 DB에 저장해줍니다.

@Scheduled(fixedDelay = 30000)
@Transactional
public void flushMessagesToDB() {
    String queueKey = "chat:queue";
    String backupKey = "chat:queue:backup";

    if (!redisTemplate.hasKey(queueKey)) return;

    redisTemplate.rename(queueKey, backupKey);

    List<Object> rawMessages = redisTemplate.opsForList().range(backupKey, 0, -1);
    if (rawMessages == null || rawMessages.isEmpty()) return;

    List<ChatMessage> entities = convertToEntities(rawMessages);
    
    if (!entities.isEmpty()) {
        chatRepositoryFacade.saveAllChatMessages(entities);
        redisTemplate.delete(backupKey);
    }
}

배치 작업의 주기로 얼마가 적당할 지 잘 모르겠어 일단 30초마다 동작하도록 했습니다.

여기서 주의할 점은 `backupKey`라는 별도 키를 사용해 배치 작업을 수행하는 것입니다.

 

`backupKey`를 사용하지 않는다면?

• `range(0, -1)`은 (왼쪽부터) 모든 데이터를 가져옵니다. (하나씩 `LPOP`하기 보다는 한번에 가져오는 것이 효율적)
• 그리고 배치 작업 도중에도 얼마든지 새로운 데이터가 메시지 큐에 저장될 수 있습니다. 즉, `range(0, -1)` 이후 들어온 메시지는 같이 저장되지 못하고 삭제되는 문제가 발생합니다.
• 따라서 기존 큐를 백업 큐로 아예 이름을 바꿔, 일종의 스냅샷을 만드는 효과를 가집니다. 그동안 새로 들어오는 메시지는 다시 생성된 기존 큐의 이름으로 쌓이게 됩니다.

마지막은 JPA의 `saveAll`로 일괄 저장 하는데, `JdbcTemplate`을 사용해서 `batch insert`로 처리해도 좋을 것 같습니다.

 

✅ 요약

🔴Before : WebSocket 실시간 채팅 메시지를 동기적으로 DB에 바로 저장

🟢After : 메시지를 잠시 큐에 쌓아두었다가 비동기적으로 DB에 일괄 저장

🛠️How : Redis List 자료 구조를 메시지 큐로 사용