[Cloudwave] CGV 대용량 트래픽 예매 서비스 구현 - 3
서론
이전 글에서 비관적 락을 통한 동시성 제어를 했다.
이번 회차에선 대규모 트래픽의 제어를 위해 Redis의 SortedSet을 통한 대기열을 구현해보려 한다.
우선, 왜 이런 구조가 필요한지부터 생각해보자.
한번에 트래픽이 몰리면 서버의 부하가 심해진다.
위의 부하테스트는 동시에 3000명의 사용자가 한 자리를 동시에 20번을 예매 시도해 총 60000번의 요청을 하는 시나리오이다.
하지만 다 처리하지 못하고 약 16000번의 처리만 한 뒤, 처리를 하지 못했고 다음과 같이 cpu 사용량이 나타났다.
이렇게 되면 어플리케이션이 요청을 처리하지 못한다.
cpu 사용량이 일정 수준을 넘으면 오토스케일링을 통해 서버를 늘려 대응할 수 있긴 하지만, 실시간이 중요한 예매 시스템에선 서버가 만들어지는 시간도 치명적이다.
따라서 앞에 대기열을 만들어 점진적으로 부하를 주어 오토스케일링을 통해 트래픽을 대응할 시간을 주는 것이다.
물론 대기열에 입장하는 것 자체도 Redis와 서버간 통신으로 CPU를 잡아먹지만, 비즈니스 로직을 실행하고 DB와 통신하는 것이 더 부하가 크다.
또한, 이번 프로젝트는 AWS상에서 인프라를 구성한다. 그 중에서 DB는 Aurora RDS를 사용한다.
Aurora RDS는 DB 클러스터를 제공해 고가용성을 제공하지만 다음과 같은 한계가 있다.
AWS Aurora의 한계
Aurora를 사용하는 경우, 기본적으로 데이터베이스 인스턴스는 다음과 같이 구성된다.
- Writer (Primary) 인스턴스: 쓰기(INSERT, UPDATE, DELETE)를 담당하는 인스턴스. 무조건 1개만 존재한다.
- Reader (Replica) 인스턴스들: 읽기(SELECT)를 담당하며 여러 개로 확장할 수 있다.
즉, 읽기 부하는 리더 인스턴스 여러 개로 쉽게 분산할 수 있지만, 쓰기 부하는 오로지 하나의 Writer 인스턴스에 몰리게 된다.
아무리 많은 서버가 붙어 있어도, 결국 쓰기 요청은 단 하나의 Writer가 모두 처리해야 하니 병목이 생길 수밖에 없다.
따라서 쓰기 요청 자체를 물리적으로 제한하여 병목을 줄이고, TPS도 일정 수준 이상 나올 수 있도록 해야한다.
Redis로 대기열을 만드는 이유
- 싱글 스레드 기반의 빠른 처리
- Redis는 명령어를 싱글 스레드로 처리하기 때문에, 각 명령어는 직렬로 처리되어 원자성(atomicity)을 보장한다.
- 멀티 인스턴스 환경에서도 안전한 처리
- Redis 자체가 단일 프로세스로 동작하므로, 여러 서버 인스턴스에서 동시에 Redis에 접근해도 명령은 순차적으로 처리된다.
왜 Sorted Set을 사용하는가?
- 자동 정렬
ZADD 명령어로 점수(score)를 넣으면 자동으로 정렬된다.
➔ 별도 추가 로직 없이, 선착순/우선순위 큐를 쉽게 구성할 수 있다. - 초고 성능
Sorted Set의 명령어는 평균 시간복잡도가 O(log N)이므로 굉장히 빠른 성능이다.
➔ 수십만, 수백만 데이터가 쌓여도 빠르고 안정적으로 동작할 수 있다.
Flow
이제 대기열 시스템의 구현을 해볼텐데, 어디에 대기열이 생기는 지 한번 보자.
좌석 예매 페이지는 다음과 같다.
예매 요청이 가능한 페이지에 모든 트래픽을 허용해버리면 DB에 과부하가 걸릴 것이다. 따라서,
좌석 선택하기 전 페이지에서 좌석 선택을 눌러 다음과 같은 대기열 화면에 입장한다.
(흐름도)
[영화 시간 선택 페이지] -> [대기열] -> [예매 가능 큐] -> [좌석 선택 페이지]
대기열 구현
@Service
@RequiredArgsConstructor
public class WaitingQueueService {
private final StringRedisTemplate redisTemplate;
// 유저를 대기열에 추가
public void enterWaitingQueue(String username, Long scheduleId) {
String queueKey = getWaitingQueueKey(scheduleId);
double timestamp = System.currentTimeMillis();
redisTemplate.opsForZSet().add(queueKey, username, timestamp);
}
// 유저의 현재 대기 순서 확인
public QueueRes getUserOrder(String username, Long scheduleId) {
String queueKey = getWaitingQueueKey(scheduleId);
Long rank = redisTemplate.opsForZSet().rank(queueKey, username);
if(rank==null)
throw new CustomException(StatusCode.QUEUE_USER_NOT_EXIST);
return new QueueRes(false,rank + 1);// 0-based index → 1-based 순서
}
public void exitQueue(String username, Long scheduleId) {
String queueKey = getWaitingQueueKey(scheduleId);
redisTemplate.opsForZSet().remove(queueKey, username);
}
public Set<String> getTopUsers(Long scheduleId, int limit) {
String queueKey = getWaitingQueueKey(scheduleId);
return redisTemplate.opsForZSet().range(queueKey, 0, limit - 1);
}
public void removeUsers(Long scheduleId, Set<String> usernames) {
String queueKey = getWaitingQueueKey(scheduleId);
redisTemplate.opsForZSet().remove(queueKey, usernames.toArray());
}
// 모든 스케줄에 대한 scheduleId 반환
public Set<Long> getAllScheduleIds() {
Set<String> keys = redisTemplate.keys("waiting_queue:schedule:*");
return keys.stream()
.map(key -> Long.parseLong(key.replace("waiting_queue:schedule:", "")))
.collect(Collectors.toSet());
}
public void deleteQueueKey(Long scheduleId) {
redisTemplate.delete(getWaitingQueueKey(scheduleId));
}
public Long getQueueSize(Long scheduleId) {
String queueKey = getWaitingQueueKey(scheduleId);
return redisTemplate.opsForZSet().size(queueKey);
}
// sortedSet에 넣을 키 반환
private String getWaitingQueueKey(Long scheduleId) {
return "waiting_queue:schedule:" + scheduleId;
}
}
대기열에 유저 추가
redisTemplate.opsForZSet().add(queueKey, username, timestamp);
사용자가 대기열에 입장할 때 유저의 이름과 영화 일정 ID(key)를 받아서 현재 시간을 기준으로 대기열에 추가한다. ZADD 명령어를 이용해 대기열에 유저를 timestamp를 기준으로 추가하고, 자동으로 정렬된 상태로 관리한다.
각 유저는 timestamp를 점수(score)로 하여 Sorted Set에 추가된다. 이렇게 하면 대기열의 순서가 자연스럽게 시간순으로 정렬된다.
대기 중인 유저의 순서 확인
redisTemplate.opsForZSet().rank(queueKey, username);
대기열에 추가된 유저의 순서는 ZRANK 명령어로 확인할 수 있다. 이를 통해 사용자가 대기열에서 몇 번째에 있는지를 알 수 있다.
예매 가능 큐 구현
@Service
@RequiredArgsConstructor
public class AvailableQueueService {
private final StringRedisTemplate redisTemplate;
private final WaitingQueueService waitingQueueService;
private static final long ALLOW_DURATION_MILLIS = 5 * 1 * 1000; // 5초
// 유저를 available queue에 추가 (입장 허용)
public void allowUser(String username, Long scheduleId) {
String key = getAvailableQueueKey(scheduleId);
long expireAt = System.currentTimeMillis() + ALLOW_DURATION_MILLIS;
redisTemplate.opsForZSet().add(key, username, expireAt);
}
// 유저가 입장 가능한지 확인
public QueueRes isUserAllowed(String username, Long scheduleId) {
String key = getAvailableQueueKey(scheduleId);
Double expireAt = redisTemplate.opsForZSet().score(key, username);
QueueRes queueRes;
if (expireAt != null && expireAt > System.currentTimeMillis()) {
queueRes = new QueueRes(true, null); // 입장 가능
} else {
queueRes = waitingQueueService.getUserOrder(username, scheduleId); // 대기열 순서 반환
}
return queueRes;
}
// 유저를 available queue에서 제거
public void exitQueue(String username, Long scheduleId) {
String key = getAvailableQueueKey(scheduleId);
redisTemplate.opsForZSet().remove(key, username);
}
// 현재 시간 기준으로 만료된 유저 가져오기
public Set<String> getExpiredUsers(Long scheduleId) {
String key = getAvailableQueueKey(scheduleId);
long now = System.currentTimeMillis();
return redisTemplate.opsForZSet().rangeByScore(key, 0, now);
}
public void removeUsers(Long scheduleId, Set<String> usernames) {
String key = getAvailableQueueKey(scheduleId);
redisTemplate.opsForZSet().remove(key, usernames.toArray());
}
// 모든 스케줄에 대한 scheduleId 반환
public Set<Long> getAllScheduleIds() {
Set<String> keys = redisTemplate.keys("available_queue:schedule:*");
return keys.stream()
.map(key -> Long.parseLong(key.replace("available_queue:schedule:", "")))
.collect(Collectors.toSet());
}
public void deleteAvailableQueueKey(Long scheduleId) {
redisTemplate.delete(getAvailableQueueKey(scheduleId));
}
// 전체 유저 가져오기 (스케쥴러에서 사용)
public Set<String> getAllUsers(Long scheduleId) {
String key = getAvailableQueueKey(scheduleId);
return redisTemplate.opsForZSet().range(key, 0, -1);
}
private String getAvailableQueueKey(Long scheduleId) {
return "available_queue:schedule:" + scheduleId;
}
}
입장 가능한 대기열은 대기열에서 예매를 할 수 있는 상태로 이동한 사용자들을 관리하는 큐다.
처음에 대기열에서 나온 사용자를 어떻게 관리해야 할까 고민을 했었다.
가장 먼저 생각한 방법은 Key-Value 방식이었다.
각 유저를 키로 따로 관리하는 방식인데, 예를 들면 available_user:{scheduleId}:{username} 같은 형태로 키를 만들고, 값으로는 예매 가능 만료 시간(expireAt)을 저장하는 식이다.
이렇게 하면 사용자마다 개별 TTL(만료시간)을 설정할 수도 있고, 만료되면 자동 삭제도 가능하다.
단점으로는 사용자 수 만큼 Key가 생성되기 때문에 지나치게 많이 만들어지고 관리하기가 힘들다
대신 Sorted Set을 사용해 영화 일정 id마다 사용자를 관리할 수 있게 하였다.
available_queue:schedule:{scheduleId}라는 이름으로 하나의 Sorted Set을 만들고,
여기에 유저 이름(username)을 멤버로 추가하고, 점수(score)에는 입장 가능 만료시간(expireAt)을 넣었다.
하지만 Redis에서 TTL은 Key에만 적용가능하기 때문에,
스케쥴러에서 일정 시간마다 만료시간이 지난 사용자들을 큐에서 제거하는 방식으로 구현하였다.
예매 스케쥴러 구현
@Component
@RequiredArgsConstructor
@Slf4j
public class ReservationQueueScheduler {
private final WaitingQueueService waitingQueueService;
private final AvailableQueueService availableQueueService;
// 대기열에서 일정 인원 꺼내 입장 허용
@Scheduled(fixedRate = 5 * 1000)
public void moveUsersFromWaitingToAvailable() {
Set<Long> allScheduleIds = waitingQueueService.getAllScheduleIds();
int moveCount = 100; // 한 번에 100명씩 입장 허용
for (Long scheduleId : allScheduleIds) {
Set<String> topUsers = waitingQueueService.getTopUsers(scheduleId, moveCount);
if (topUsers == null || topUsers.isEmpty()) {
return;
}
// 입장 허용 → available queue로 이동
for (String username : topUsers) {
availableQueueService.allowUser(username, scheduleId);
log.info("입장 허용: {} (scheduleId={})", username, scheduleId);
}
// waiting queue에서 제거
waitingQueueService.removeUsers(scheduleId, topUsers);
}
}
// 입장가능열에서 만료된 유저 제거
@Scheduled(fixedRate = 5 * 1000)
public void removeExpiredUsersFromAvailableQueue() {
Set<Long> allScheduleIds = availableQueueService.getAllScheduleIds();
for (Long scheduleId : allScheduleIds) {
Set<String> expiredUsers = availableQueueService.getExpiredUsers(scheduleId);
if (expiredUsers != null && !expiredUsers.isEmpty()) {
availableQueueService.removeUsers(scheduleId, expiredUsers);
log.info("만료 제거: {}명 (scheduleId={})", expiredUsers.size(), scheduleId);
}
}
}
}
대기열에서 입장 가능한 대기열로 유저 이동
스케줄러는 일정 시간마다 대기열에서 일정 수의 유저를 예매 가능 큐로 이동시키는 작업을 한다.
입장 가능 큐에서 만료된 유저 제거
일정 시간마다 입장 가능 큐에서 만료된 유저를 제거하는 작업을 한다.
파라미터 설정
지금은 예매 가능 큐의 만료 시간, 스케쥴러의 입장 허용 수와 같은 파라미터를 임의로 테스트하기 쉽도록 설정했다.
프로젝트가 끝난 뒤, 트래픽 수와 실제 인스턴스와,db의 스펙에 맞게 파라미터를 설정할 예정이다.
테스트
@SpringBootTest
@ActiveProfiles("test")
@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
public class RedisQueueTest {
@Autowired
private ReservationQueueScheduler reservationQueueScheduler;
@Autowired
private WaitingQueueService waitingQueueService;
@Autowired
private AvailableQueueService availableQueueService;
private static final Long SCHEDULE_ID = 1L;
private static final int USER_COUNT = 1000;
@BeforeEach
public void setUp() {
// 대기열에 1000명 추가
IntStream.rangeClosed(1, USER_COUNT)
.forEach(i -> waitingQueueService.enterWaitingQueue("user" + i, SCHEDULE_ID));
}
@AfterEach
public void tearDown() {
// 대기열 & 입장열 키 삭제로 초기화
waitingQueueService.deleteQueueKey(SCHEDULE_ID);
availableQueueService.deleteAvailableQueueKey(SCHEDULE_ID);
}
@Test
@Order(1)
public void testUserEnterWaitingQueue() {
Long queueSize = waitingQueueService.getQueueSize(SCHEDULE_ID);
assertThat(queueSize).isEqualTo(USER_COUNT);
}
@Test
@Order(2)
public void testMoveUsersToAvailableQueue() {
reservationQueueScheduler.moveUsersFromWaitingToAvailable();
// available queue에 100명, waiting queue에 900명 남아야 함
Set<String> availableUsers = availableQueueService.getAllUsers(SCHEDULE_ID);
Long remainingWaiting = waitingQueueService.getQueueSize(SCHEDULE_ID);
assertThat(availableUsers.size()).isEqualTo(100);
assertThat(remainingWaiting).isEqualTo(900);
}
@Test
@Order(3)
public void testIsUserAllowed() {
reservationQueueScheduler.moveUsersFromWaitingToAvailable();
QueueRes allowed = availableQueueService.isUserAllowed("user1", SCHEDULE_ID);
QueueRes waiting = availableQueueService.isUserAllowed("user150", SCHEDULE_ID);
assertThat(allowed.getIsAvailable()).isTrue();
assertThat(waiting.getIsAvailable()).isFalse();
}
@Test
@Order(4)
public void testRemoveExpiredUsers() throws InterruptedException {
reservationQueueScheduler.moveUsersFromWaitingToAvailable();
// 입장 허용 후 5초 대기 (테스트용으로 5초뒤 만료로 설정함)
Thread.sleep(5000);
// 지금은 그냥 스케쥴러 실행 확인
reservationQueueScheduler.removeExpiredUsersFromAvailableQueue();
// 큐에 없어야함.
Set<String> stillAllowed = availableQueueService.getAllUsers(SCHEDULE_ID);
assertThat(stillAllowed.size()).isEqualTo(0);
}
}
순서대로 Redis를 이용한 기능이 잘 동작하는지에 대한 테스트 코드이다.
1. testUserEnterWaitingQueue()
setUp()을 통해 유저 1000명이 큐에 들어왔는지 확인하는 테스트이다.
2. testMoveUsersToAvailableQueue()
대기열에서 일부 사용자를 입장 가능 큐로 이동시키는 로직을 검증한다.
3. testIsUserAllowed()
입장 가능 큐에 있는 유저가 입장이 가능한지 확인하는 테스트이다.
4. testRemoveExpiredUsers()
입장 가능한 대기열에서 만료된 유저가 제거되었는지, 큐에서 더 이상 만료된 유저가 없는지 확인한다.
마무리
대기열을 통해 순차적으로 입장 가능한 사용자들만 예매를 진행하게 함으로써, 서버와 DB의 과부하를 방지하고 고가용성을 보장하는 시스템을 구현하였다.
이렇게 보면 끝난 것 같지만, 서버와 db의 부하를 막는 방파제역할을 하는 Redis 대기열에 모든 트래픽이 몰리게된다.
따라서 이를 또 수용할 수 있는 시스템이 필요하다.
다음 글엔 kafka를 통해 Redis 대기열에 몰리는 트래픽을 순차적으로 보낼 수 있는 아키텍처를 만들어 보도록 하겠다.