Redis로 대기열 구현하기

이번 포스팅에선 Redis로 대기열을 구현한 것을 공유하려고 컴퓨터 앞에 앉았습니다. 

 

이번에 주요한 기능은 WebSocket과 Redis의 Sorted Set 자료구조입니다. Sorted Set은 정렬 알고리즘의 시간 복잡도가 O(log n)이라 굉장히 빠르고 중복을 허용하지 않는 Set 자료구조에다 정렬이 되어있어 선착순으로 사용자를 지워줄 수 있다는 장점이 있습니다. 

 

Redis의 다양한 자료구조들은 기능을 구현할 때 유용하게 쓰이는 것 같네요. 

 

이번 포스팅은 다음과 같은 흐름을 가집니다. 

 

상황)

온라인 쇼핑몰에서 블랙 프라이데이를 기념하여 50퍼센트 할인 쿠폰을 100명에게 쏩니다. 서버는 이를 감당할 수 있는 대기열을 구현해야합니다. 

 

흐름)

1. 사용자가 이벤트 페이지를 조회합니다. 
2. 조회함과 동시에 웹소켓에 연결합니다. 
3. 연결할 때 웹소켓 핸들러에서 사용자를 대기열에 추가합니다. 이때 Redis Sorted Set에 추가합니다. 
4. 10초에 한번씩 Redis range명령어로 상위 10명의 데이터를 가져옵니다. 그와 동시에 데이터베이스에서 수량을 감소시킵니다. 
5. 상위 10명의 데이터를 Sorted Set에서 지워버립니다. 
6. 대기하고 있는 사용자에겐 WebSocket을 이용해 현재 얼마나 대기열이 형성되어있는지 알려줍니다. 

 

바로 시작해보죠. 

 

@Data
@AllArgsConstructor
public class Coupon {

    private String couponName;
    private Item item;
    private int code;
}

 

우선 쿠폰 정보입니다.

 

그리고 제일 중요한 EventService를 만들어야합니다. 

 

@Service
@RequiredArgsConstructor
@Slf4j
public class EventService {

    private final ItemRepository itemRepository;
    private final RedisTemplate<String, Object> redisTemplate;
    private static final long FIRST_ELEMENT = 0;
    private static final long LAST_ELEMENT = 9;

    public void addQueue(String sessionId) {
        final String people = Thread.currentThread().getName();
        final long now = System.currentTimeMillis();

        redisTemplate.opsForZSet().add("치킨", sessionId, 10f);
        log.info("대기열에 추가 - {} ({})초", people, now);
    }

    public Long getOrder(String sessionId) {
        return redisTemplate.opsForZSet().rank("치킨", sessionId);
    }

    @Transactional
    public void publish(Item item) {
        Set<Object> queue = redisTemplate.opsForZSet().range("치킨", FIRST_ELEMENT, LAST_ELEMENT);

        for (Object people : queue) {
            final Coupon coupon = new Coupon("치킨", item, 10000);
            log.info("{}님의 {}기프티콘이 발급되었습니다. ({})", people, "치킨", coupon.getCode());
            redisTemplate.opsForZSet().remove("치킨", people);
            Item decreaseItem = item.decreaseCount();
            itemRepository.save(decreaseItem);
        }
    }

    public boolean isValidEvent(Item item) {
        return item.getCount() == 0;
    }
}

 

FIRST_ELEMENT는 0으로 잡아주시고 LAST_ELEMENT는 몇개의 데이터를 처리할 것인지를 결정하는 것입니다. 저는 0부터 9개 즉, 상위 10개의 데이터를 가져올 것입니다. 

 

@Slf4j
@Component
@RequiredArgsConstructor
public class EventScheduler {

    private final EventService eventService;
    private final Item item;

    @Scheduled(fixedDelay = 10000)
    private void chickenEventScheduler() {
        if (eventService.isValidEvent(item)) {
            log.info("===== 선착순 이벤트가 종료되었습니다. =====");
            return;
        }
        eventService.publish(item);
    }
}

 

EventScheduler에선 10초마다 데이터를 처리해줄겁니다. 

 

@Configuration
@EnableWebSocket
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketConfigurer {

    private final QueueWebSocketHandler queueWebSocketHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(queueWebSocketHandler, "/ws").setAllowedOrigins("*");
    }
}

 

위의 Config는 웹소켓과 관련된 설정입니다. 

@Slf4j
@Component
@RequiredArgsConstructor
public class QueueWebSocketHandler extends TextWebSocketHandler {

    private final EventService eventService;
    private final RedisTemplate<String, Object> redisTemplate;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        eventService.addQueue(session.getId());
        log.info("Connected : " + session.getId());
    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        Long rank = eventService.getOrder(session.getId());
        log.info("session id : " + session.getId());
        long rankMessage = rank == null ? 0 : rank;
        session.sendMessage(new TextMessage(String.valueOf(rankMessage)));
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("Disconnected : " + session.getId());
    }
}

 

QueueWebSocketHandler에선 웹소켓과 관련된 작업을 처리해줄겁니다. 

 

그리고 프론트단을 가볍게 만들어보면?

 

import {useEffect, useRef, useState} from 'react'
import './App.css'
function App() {
  
  const ws = useRef(null);
  const [queue, setQueue] = useState("0");

  useEffect(() => {
    ws.current = new WebSocket("ws://localhost:8080/ws");
    console.log("test");

    const sendMessage = setInterval(() => {
      ws.current.send("hello");
    }, 2000);

    ws.current.onmessage = function(event) {
      setQueue(event.data);
    }

    return () => {
      clearInterval(sendMessage);
    }
  }, [])

  return (
    <div className="App">
      <div className='message'>잠시만 기다려주세요.</div>
      {queue && (
        <div className='queue-message'>
          대기열 : {queue}
        </div>
      )}
    </div>
  );
}

export default App;

 

처음에 웹소켓과 연결하고 2초마다 hello 메세지를 보내서 현재 대기열이 얼마나 남았는지 알려주는 역할을 합니다. 그리고 UseState를 이용해서 값을 지속적으로 변경해주도록 하겠습니다. 

 

그럼 결과물이 이렇게 나옵니다. 

 

 

 

 

마치며

Redis는 참 보면볼수록 오묘한... 그런 녀석인 것 같습니다. 저는 캐시로 사용하고 땡인줄 알았는데 다양한 자료구조로 다양한 기능을 만들 수 있는 것을 보고 새삼 왜 사람들이 Redis를 많이 사용하는지 알겠더군요. 

 

저는 이번 기회에 Redis에 대해 좀 더 깊이있게 공부하게 된 것 같아 정말 뿌듯합니다. 오늘도 긴 글 읽어주셔서 감사합니다. 즐거운 하루 되세요~