0. 들어가며
Spring WebSocket에서 기본으로 제공하는 Simple Broker는 이름 그대로 “Simple”한 구현체다. 별도의 외부 브로커 없이도 STOMP 기반 메시징을 빠르게 구성할 수 있도록, 각 서버 인스턴스의 메모리 안에서만 동작한다.
WebSocket 세션은 서버 인스턴스에 직접 연결되며, SimpMessagingTemplate을 통해 전송되는 메시지는 해당 인스턴스가 관리하고 있는 세션들에게만 전달된다. 이는 성능과 단순성을 위해 의도된 설계다.
문제는 서버가 하나일 때는 전혀 드러나지 않는다. 하지만 인스턴스가 두 개 이상이 되는 순간, 각 서버는 서로의 세션 존재를 알 수 없게 되며, 아래와 같은 문제가 발생한다.
- 첫 번째, 메시지 전파 문제
사용자 A는 인스턴스 1에, 사용자 B는 인스턴스 2에 접속해 있다고 가정해보자. A가 메시지를 보내면 인스턴스 1의 메모리에 연결된 세션들에는 메시지가 전달되지만, 인스턴스 2에 있는 B에게는 메시지가 도달하지 않는다. 인스턴스 간에 서로 "메시지가 왔으니 니가 가진 세션들에게도 뿌려줘"라고 말해줄 중간 매개체(Message Broker)가 없기 때문이다. - 두 번째, DB 저장과 메시지 발행의 원자성(Automicity)
브로커를 도입하더라도 더 큰 고민이 남는다. 채팅 내용을 DB에 저장하는 것과 메시지 브로커에 전달하는 것, 이 두 작업은 서로 다른 인프라에서 일어난다. 만약 DB 저장은 성공했는데 갑자기 네트워크 오류로 Message Broker로의 발행이 실패한다면? 혹은 그 반대의 상황이라면? 사용자는 채팅을 보냈다고 생각하지만 상대방은 영원히 메시지를 받지 못하거나, DB에는 기록이 없는 유령 메시지가 떠다니는 데이터 불일치 현상이 발생한다.
Simple Broker는 “모든 클라이언트에게 메시지를 보내는 브로커”가 아니라, “이 인스턴스에 연결된 클라이언트들에게만 메시지를 전달하는 로컬 브로커”에 가깝다. 몰론 Simple Broker 외에도 STOMP Broker Relay라는 선택지가 있지만, 이는 WebSocket 메시지 전파를 외부 브로커에 위임하여 인스턴스 스케일 아웃 시 세션 전파 문제만 해결해줄 뿐 DB 트랜잭션과 메시지 발행의 원자성은 해결하지 못한다.
따라서 멀티 인스턴스 환경에서 WebSocket 메시지를 전체 사용자에게 일관되게 전달하려면, 인스턴스 간 메시지를 중계해줄 외부 Message Broker가 반드시 필요하다. 그리고 이때 DB 저장과 메시지 발행의 원자성 보장도 함께 고려해야 한다.
결론: Message Broker와 Transactional Outbox 패턴을 통한 해결
이 문제를 해결하기 위해 외부 브로커인 RabbitMQ(CloudAMQP)를 도입하여 인스턴스 간 통신을 연결하고, Outbox 패턴을 통해 DB 저장과 메시지 발행을 하나의 트랜잭션으로 묶어 데이터의 정합성을 보장하기로 하였다.
1. 왜 RabbitMQ인가
동일한 문제를 해결하기 위해 Redis나 Kafka를 사용하는 사례도 많다. 하지만 현재 프로젝트의 요구사항과 운영 환경을 기준으로 보면, 두 기술 모두 아쉬운 지점이 있었다.
Redis Pub/Sub
Redis Pub/Sub은 구독자가 연결되어 있는 순간에만 메시지를 전달한다. 즉, 메시지를 저장하지 않는다. 인스턴스가 재시작 중이거나 네트워크 순단이 발생하면 그 사이에 발행된 메시지는 영구적으로 유실되는 문제가 있다.
Outbox 패턴으로 "발행 보장"을 구현해놓고, 브로커 단에서 메시지가 유실되면 의미가 없다. Redis Pub/Sub에는 메시지 ACK 메커니즘도 없어서, 브로커가 "전달했다"고 확인해줄 방법도 없다.
Redis Streams를 쓰면 영속성 문제는 해결되지만, Consumer Group 관리와 오프셋 추적을 직접 구현해야 하는 부담이 생긴다. 이 시점에서 이미 RabbitMQ가 제공하는 것을 재구현하는 셈이다.
Kafka
Kafka는 대용량 이벤트 스트리밍에 설계된 시스템이다. 메시지를 디스크에 순서대로 저장하고, 컨슈머가 오프셋을 관리하며, 파티션 기반으로 수평 확장한다.
Kafka에서 Fanout 브로드캐스트를 구현하려면 각 인스턴스가 서로 다른 Consumer Group을 가져야 한다. 가능은 하지만, 이 정도 규모의 채팅에 Kafka의 파티션/오프셋/리밸런싱 관리를 도입하는 것은 복잡도 대비 이점이 없다고 생각했다.
RabbitMQ
RabbitMQ는 다음 요구사항을 균형 있게 만족시켰다.
1. Fanout Exchange가 요구사항에 정확히 대응한다
"모든 인스턴스에 같은 메시지를 전달"이라는 요구사항을 Exchange 타입 하나로 해결한다. 별도의 Consumer Group 트릭이나 토픽 복제 설정 없이, Fanout Exchange에 큐를 바인딩하면 끝이다.
2. 메시지 ACK로 전달을 보장한다
Consumer가 메시지를 정상 처리하면 ACK를 보내고, 큐에서 제거된다. ACK 전에 Consumer가 죽으면 메시지는 큐에 남아서 다른 Consumer(또는 재시작된 Consumer)가 다시 받는다. Redis Pub/Sub에는 없는 메커니즘이다.
3. CloudAMQP 무료 플랜
CloudAMQP의 Little Lemur 플랜(무료)으로 managed RabbitMQ를 사용할 수 있다. 사이드 프로젝트에서 메시지 브로커 서버를 직접 운영할 필요가 없다.
4. 큐의 생명주기가 인스턴스와 일치한다
exclusive=true, autoDelete=true 큐를 사용하면, 인스턴스가 종료될 때 큐가 자동 삭제된다. 죽은 인스턴스의 큐에 메시지가 쌓이는 문제가 없다. Kafka에서는 비활성 Consumer Group의 오프셋 관리를 별도로 처리해야 한다.
결론적으로 메시지 유실 방지가 목표이므로 Redis Pub/Sub은 탈락, 처리량 수십 건 수준의 채팅에 Kafka는 과하므로 탈락. RabbitMQ의 Fanout Exchange + 메시지 ACK가 요구사항에 가장 정확히 부합했다.
2. 아키텍처 설계
최종적으로 설계한 메시지 흐름은 다음과 같다.
Client (STOMP)
│
▼
WebSocketController
│
▼
WebSocketChatService (@Transactional)
├── ChatMessage DB 저장
└── OutboxEvent DB 저장 (같은 트랜잭션)
│
▼ (트랜잭션 커밋 후)
TransactionalEventListener
│
▼ (즉시 발행 시도)
RabbitMQ (CloudAMQP)
Fanout Exchange
┌─────┴─────┐
▼ ▼
Queue(inst-1) Queue(inst-2) ← 인스턴스별 전용 큐
│ │
▼ ▼
Consumer Consumer
│ │
▼ ▼
STOMP 전송 STOMP 전송
│ │
▼ ▼
Client A Client B핵심 원칙
- DB 저장과 이벤트 발행을 같은 트랜잭션에서 -> 원자성 보장
- RabbitMQ Fanout Exchange로 모든 인스턴스에 브로드캐스트 -> 다중 인스턴스 문제 해결
- 즉시 발행 실패 시 스케줄러로 재시도 -> 메시지 유실 방지
3. 구현 상세: Outbox 패턴
Outbox 패턴은 마이크로서비스 아키텍처에서 자주 사용되는 패턴이다. 핵심 아이디어는 간단하다. 외부 시스템(여기서는 RabbitMQ)에 보낼 메시지를 DB 테이블에 먼저 저장하고, 비즈니스 로직과 같은 트랜잭션으로 묶는 것이다.
테이블 설계
CREATE TABLE outbox_events (
id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
aggregate_type VARCHAR(100) NOT NULL,
aggregate_id VARCHAR(255) NOT NULL,
event_type VARCHAR(100) NOT NULL,
payload JSONB NOT NULL,
status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
retry_count INT NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
published_at TIMESTAMPTZ
);
CREATE INDEX idx_outbox_events_pending
ON outbox_events (status, created_at) WHERE status = 'PENDING';payload를 JSONB로 저장한 이유는 브로드캐스트 대상(topic/user)과 메시지 내용을 하나의 구조화된 데이터로 담기 위해서다.
엔티티
enum class OutboxEventStatus {
PENDING,
PUBLISHED
}
@Entity
@Table(name = "outbox_events")
class OutboxEvent(
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
val id: Long = 0L,
@Column(name = "aggregate_type", nullable = false, updatable = false)
val aggregateType: String,
@Column(name = "aggregate_id", nullable = false, updatable = false)
val aggregateId: String,
@Column(name = "event_type", nullable = false, updatable = false)
val eventType: String,
@JdbcTypeCode(SqlTypes.JSON)
@Column(nullable = false, updatable = false, columnDefinition = "jsonb")
val payload: String,
@Enumerated(EnumType.STRING)
@Column(nullable = false)
var status: OutboxEventStatus = OutboxEventStatus.PENDING,
@Column(name = "retry_count", nullable = false)
var retryCount: Int = 0,
@Column(name = "created_at", nullable = false, updatable = false)
val createdAt: Instant = Instant.now(),
@Column(name = "published_at")
var publishedAt: Instant? = null
) {
fun markPublished() {
status = OutboxEventStatus.PUBLISHED
publishedAt = Instant.now()
}
}트랜잭션 내에서 함께 저장
WebSocketChatService에서 채팅 메시지와 Outbox 이벤트를 같은 @Transactional 안에서 저장한다.
@Transactional
fun sendMessage(userId: Long, chatRoomId: Long, content: String) {
// 1. 채팅 메시지 DB 저장
val chatMessage = chatRoomService.saveChatMessage(userId, chatRoomId, content)
// 2. 브로드캐스트 메시지 구성
val response = ChatMessageResponse(
id = chatMessage.id,
chatRoomId = chatRoomId,
senderId = userId,
content = content,
type = MessageType.CHAT,
timestamp = chatMessage.createdAt
)
val broadcasts = mutableListOf<ChatBroadcastMessage>()
broadcasts.add(
ChatBroadcastMessage(
destinations = listOf(
BroadcastDestination(
type = DestinationType.TOPIC,
path = "/topic/chat/$chatRoomId"
)
),
payload = response
)
)
// 3. Outbox에 저장
saveOutboxEvents(chatRoomId.toString(), "CHAT_MESSAGE", broadcasts)
}
private fun saveOutboxEvents(aggregateId: String, eventType: String, broadcasts: List<ChatBroadcastMessage>) {
val payload = objectMapper.writeValueAsString(broadcasts)
val outboxEvent = outboxEventRepository.save(
OutboxEvent(
aggregateType = "ChatRoom",
aggregateId = aggregateId,
eventType = eventType,
payload = payload
)
)
// Spring ApplicationEvent 발행 -> 트랜잭션 커밋 후 즉시 발행 시도
applicationEventPublisher.publishEvent(OutboxPublishEvent(outboxEvent.id))
}여기서 중요한 점은 ChatMessage와 OutboxEvent가 같은 트랜잭션에서 저장되므로, 둘 다 성공하거나 둘 다 실패한다는 것이다. 이를 통해 DB에 메시지는 있는데 이벤트가 없는 상황, 혹은 그 반대 상황을 원천적으로 방지할 수 있다.
4. 구현 상세: 즉시 발행 + Fallback 폴링
Outbox 패턴의 이벤트 발행 전략은 두 가지를 조합했다.
즉시 발행: @TransactionalEventListener
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
@Transactional(propagation = Propagation.REQUIRES_NEW)
fun handleOutboxPublishEvent(event: OutboxPublishEvent) {
try {
val outboxEvent = outboxEventRepository.findById(event.outboxEventId).orElse(null) ?: return
if (outboxEvent.status != OutboxEventStatus.PENDING) return
rabbitTemplate.convertAndSend(exchangeName, "", outboxEvent.payload)
outboxEvent.markPublished()
outboxEventRepository.save(outboxEvent)
} catch (e: Exception) {
logger.warn(e) { "즉시 발행 실패 (outboxEventId=${event.outboxEventId}), 스케줄러에서 재시도 예정" }
}
}TransactionPhase.AFTER_COMMIT을 사용하여 트랜잭션이 커밋된 후에만 실행되도록 함으로써, DB에 데이터가 확실히 저장된 상태에서만 RabbitMQ로 발행을 시도하도록 했다.
Propagation.REQUIRES_NEW로 새 트랜잭션을 열어서 status 업데이트를 처리한다. 어차피 Fallback 폴링이 있기 때문에, 즉시 발행에 실패해도 예외를 삼키고 로그만 남기도록 했다.
Fallback 폴링: @Scheduled
@Scheduled(fixedDelayString = $$"${outbox.poll-interval-ms:5000}")
@SchedulerLock(name = "outbox-poll", lockAtMostFor = "PT25S", lockAtLeastFor = "PT4S")
@Transactional
fun pollPendingEvents() {
val pendingEvents = outboxEventRepository.findByStatus(OutboxEventStatus.PENDING)
for (event in pendingEvents) {
if (event.retryCount >= maxRetryCount) {
logger.error { "최대 재시도 횟수 초과 (outboxEventId=${event.id}, retryCount=${event.retryCount})" }
continue
}
try {
rabbitTemplate.convertAndSend(exchangeName, "", event.payload)
event.markPublished()
} catch (e: Exception) {
event.retryCount++
logger.warn(e) { "Outbox 폴링 발행 실패 (outboxEventId=${event.id}, retryCount=${event.retryCount})" }
}
outboxEventRepository.save(event)
}
}5초마다 PENDING 상태인 이벤트를 조회해서 재시도한다. 무한 재시도를 방지하기 위해 maxRetryCount(기본 5회)를 두었다.
정리(Cleanup): 오래된 이벤트 삭제
@Scheduled(fixedRate = 3600000)
@SchedulerLock(name = "outbox-cleanup", lockAtMostFor = "PT5M", lockAtLeastFor = "PT30S")
@Transactional
fun cleanupPublishedEvents() {
val cutoff = Instant.now().minus(cleanupOlderThanHours, ChronoUnit.HOURS)
outboxEventRepository.deleteByStatusAndPublishedAtBefore(OutboxEventStatus.PUBLISHED, cutoff)
}PUBLISHED 상태로 1시간 이상 경과한 이벤트를 삭제하여, Outbox 테이블이 무한정 커지는 것을 방지하였다.
5. 구현 상세: RabbitMQ Fanout Exchange
왜 CloudAMQP인가
필자는 RabbitMQ를 직접 운영하는 대신 CloudAMQP(managed RabbitMQ 서비스)를 사용했다. Render를 사용하여 서버를 배포하고 있기 때문에 서버 내부에서 띄우기보다는 안정성을 위해 외부 관리형 서비스를 사용하는 방식을 채택했다.
Fanout Exchange + 인스턴스별 Queue
@Configuration
class RabbitMQConfig(
@Value($$"${chat.exchange-name}")
private val exchangeName: String
) {
private val instanceId: String = UUID.randomUUID().toString()
@Bean
fun chatBroadcastExchange(): FanoutExchange =
FanoutExchange(exchangeName, true, false)
@Bean
fun chatInstanceQueue(): Queue =
Queue("$exchangeName.$instanceId", false, true, true)
@Bean
fun chatQueueBinding(chatBroadcastExchange: FanoutExchange, chatInstanceQueue: Queue): Binding =
BindingBuilder.bind(chatInstanceQueue).to(chatBroadcastExchange)
}핵심 설계
- Fanout Exchange: 바인딩된 모든 큐에 메시지를 복제해서 전달한다. 라우팅 키를 무시하고 무조건 브로드캐스트
- 인스턴스별 UUID 큐: 각 서버 인스턴스가 시작될 때 고유한 큐를 생성한다. exclusive=true, autoDelete=true로 설정해서 인스턴스가 죽으면 큐도 자동으로 삭제되도록 한다.
이 구조 덕분에 인스턴스가 3개든 10개든, 모든 인스턴스가 동일한 채팅 메시지를 수신할 수 있다.
Consumer: RabbitMQ + WebSocket 클라이언트
@Component
class ChatMessageRabbitMQConsumer(
private val messagingTemplate: SimpMessagingTemplate,
private val objectMapper: ObjectMapper
) {
@RabbitListener(queues = ["#{chatInstanceQueue.name}"])
fun handleMessage(payload: String) {
try {
val broadcasts: List<ChatBroadcastMessage> = objectMapper.readValue(
payload,
object : TypeReference<List<ChatBroadcastMessage>>() {}
)
for (broadcast in broadcasts) {
for (destination in broadcast.destinations) {
val serializedPayload = objectMapper.writeValueAsString(broadcast.payload)
when (destination.type) {
DestinationType.TOPIC -> {
messagingTemplate.convertAndSend(destination.path, serializedPayload)
}
DestinationType.USER -> {
messagingTemplate.convertAndSendToUser(
destination.userId!!,
destination.path,
serializedPayload
)
}
}
}
}
} catch (e: Exception) {
logger.error(e) { "RabbitMQ 메시지 처리 실패" }
}
}
}RabbitMQ에서 메시지를 받으면, 목적지 타입(TOPIC/USER)에 따라 SimpMessagingTemplate으로 해당 인스턴스에 연결된 WebSocket 클라이언트에게 전달한다. 모든 인스턴스가 같은 메시지를 받지만, 각 인스턴스는 자신에게 연결된 클라이언트에게만 전송하므로 중복 수신은 발생하지 않는다.
6. 구현 상세: ShedLock으로 분산 스케줄링
@Scheduled 폴링은 모든 인스턴스에서 동시에 실행된다. 따라서 인스턴스가 3개면 같은 PENDING 이벤트를 3개 인스턴스가 동시에 처리하려고 시도할 수 있다. 결과적으로 같은 메시지가 3번 발행될 수도 있다.
이를 해결하기 위해 ShedLock을 사용하였다. 분산 스케줄링에서 "하나의 인스턴스만 실행"을 보장하는 방법은 여러 가지가 있겠지만, ShedLock을 선택한 데는 몇 가지 이유가 있다.
- Outbox 이벤트는 PostgreSQL에 저장되어 있으므로, 락도 같은 DB에 두면 "PostgreSQL만 살아있으면 폴링이 동작한다"는 단순한 의존 관계를 만들 수 있다.
- 추가 인프라가 필요 없고, 코드 변경을 최소화할 수 있다.
- ShedLock은 "여러 인스턴스에서 같은 스케줄러가 동시에 실행되지 않도록 보장"한다는 딱 하나의 문제만 풀기 때문에, 별다른 기능이 없는 대신, 그만큼 가볍고 이해하기 쉽다.
@Configuration
@EnableScheduling
@EnableSchedulerLock(defaultLockAtMostFor = "PT30S")
class ShedLockConfig {
@Bean
fun lockProvider(dataSource: DataSource): LockProvider =
JdbcTemplateLockProvider(
JdbcTemplateLockProvider.Configuration.builder()
.withJdbcTemplate(org.springframework.jdbc.core.JdbcTemplate(dataSource))
.usingDbTime()
.build()
)
}ShedLock은 DB 테이블을 이용한 분산 락 라이브러리다. shedlock 테이블에 락 정보를 기록해서, 여러 인스턴스 중 하나만 스케줄러를 실행하도록 보장한다.
CREATE TABLE shedlock (
name VARCHAR(64) PRIMARY KEY,
lock_until TIMESTAMPTZ NOT NULL,
locked_at TIMESTAMPTZ NOT NULL DEFAULT now(),
locked_by VARCHAR(255) NOT NULL
);스케줄러에 적용한 락 설정은 다음과 같다.
@SchedulerLock(name = "outbox-poll", lockAtMostFor = "PT25S", lockAtLeastFor = "PT4S")- lockAtMostFor = "PT25S": 락을 최대 25초까지 유지한다. 인스턴스가 비정상 종료되어도 25초 후에는 다른 인스턴스가 락을 획득할 수 있다.
- lockAtLeastFor = "PT4S": 락을 최소 4초 동안 유지한다. 작업이 너무 빨리 끝나도 다음 인스턴스가 바로 실행하지 않도록 방지할 수 있다.
7. 결과
달성한 것
- 메시지 유실 방지: Outbox 패턴으로 DB 트랜잭션과 메시지 발행의 원자성을 보장했다. 즉시 발행이 실패해도 폴링으로 재시도된다.
- 다중 인스턴스 지원: RabbitMQ Fanout Exchange로 모든 인스턴스에 메시지를 브로드캐스트한다.
- 장애 복구: 서버가 재시작되어도 PENDING 상태의 이벤트를 폴링으로 복구한다.
- 중복 처리 방지: ShedLock으로 스케줄러의 분산 실행을 제어한다.
트레이드오프
- 복잡도 증가: 단순 messagingTemplate.send() 한 줄이면 될 것을 Outbox 테이블, 이벤트 리스너, 폴링 스케줄러, RabbitMQ Consumer 까지 만들어야 한다.
- 지연: 즉시 발행이 성공하면 거의 실시간이지만, 실패 시 최대 5초(폴링 간격)까지 지연될 수 있다.
- Outbox 테이블 관리: 정리 스케줄러를 돌리지 않으면 테이블이 계속 커진다.
추후 개선 가능한 점
- CDC(Change Data Capture) 방식: Debezium 같은 도구로 Outbox 테이블의 변경을 감지해서 폴링 없이 이벤트를 발행할 수 있다. 폴링보다 실시간성이 좋고 DB 부하도 줄어든다.
- Dead Letter Queue: 최대 재시도 횟수를 초과한 이벤트를 별도 큐로 이동시켜 모니터링할 수 있다.
'개발일지' 카테고리의 다른 글
| jOOQ 코드 생성 프로세스 개선하기 (0) | 2026.01.10 |
|---|
