Redis Sorted Set으로 운동대결 리더보드 구현하기

2025년 02월 15일

experience

# Redis# Sorted Set# 리더보드# 비트 연산

들어가며

홈트레이닝 서비스를 운영하면서 사용자들의 운동 리텐션을 높이는 것이 중요한 과제였습니다. 혼자 운동하다 보면 동기부여가 떨어지기 쉬운데, 이를 해결하기 위해 운동대결 기능을 도입하게 되었습니다.

운동대결은 일정 기간 동안 참가자들이 운동을 하며 경험치를 쌓고, 순위를 경쟁하는 기능입니다. 이 글에서는 리더보드를 Redis Sorted Set으로 구현하면서 고민했던 점들을 정리합니다.

이번 글의 범위

다루는 것: Redis Sorted Set을 활용한 리더보드 구현, 복합 정렬을 위한 Score 설계
다루지 않는 것: Redis 클러스터 구성, 대회 생성/참가 로직

문제 상황

리더보드 요구사항

운동대결 리더보드는 다음과 같은 정렬 기준이 필요했습니다:

경험치가 높은 순으로 정렬
경험치가 같으면, 먼저 운동한 사람이 높은 순위
실시간으로 순위 조회 가능

RDBMS의 한계

단순히 RDB에서 ORDER BY xp DESC, completed_at ASC로 조회할 수도 있지만, 다음과 같은 문제가 있었습니다:

참가자가 많아지면 매번 정렬하는 비용이 증가
순위 조회가 빈번할수록 DB 부하 증가
실시간 순위 변동 반영이 어려움

왜 Redis Sorted Set인가

Redis Sorted Set은 리더보드 구현에 최적화된 자료구조입니다.

연산	시간 복잡도	설명
ZADD	O(log N)	멤버 추가/업데이트
ZREVRANK	O(log N)	특정 멤버의 순위 조회
ZREVRANGE	O(log N + M)	상위 N명 조회
ZSCORE	O(1)	특정 멤버의 점수 조회

Sorted Set은 Score 기반으로 자동 정렬되므로, 매번 전체를 정렬할 필요 없이 효율적으로 순위를 관리할 수 있습니다.

핵심 아이디어: Score 설계

문제: 복합 정렬 조건

Redis Sorted Set의 Score는 단일 숫자입니다. 하지만 우리는 두 가지 기준(경험치, 운동 시작 시간)으로 정렬해야 합니다.

이를 해결하기 위해 비트 연산을 활용했습니다.

해결: 비트 연산으로 복합 Score 만들기

def calculate_score(xp: int, now: datetime, ended_at: datetime) -> int:
    """
    ended_at과 now의 차이를 최대 2^22 - 1 = 4,194,303초까지 수용 가능 (약 48일)
    """
    remaining_seconds = int(ended_at.timestamp() - now.timestamp())
    return (xp << 22) + remaining_seconds

동작 원리

Score = (경험치 × 2^22) + 남은 시간

예시: 경험치 100, 남은 시간 3600초
Score = (100 << 22) + 3600
      = 100 × 4,194,304 + 3600
      = 419,430,400 + 3,600
      = 419,434,000

왜 이렇게 설계했는가

경험치 우선: 상위 비트에 경험치를 배치하여 경험치가 1만 달라도 Score 차이가 커짐
시간으로 동점 처리 (Tiebreaker): 하위 비트에 남은 시간을 배치하여 경험치가 같을 때 먼저 운동한 사람이 높은 점수. 이처럼 동점 상황을 해결하는 보조 기준을 **타이브레이커(Tiebreaker)**라고 합니다.
22비트 선택 이유: 약 48일(4,194,303초)을 표현할 수 있어 한 달 단위 대회에 충분

정렬 결과 예시

사용자	경험치	남은 시간	Score	순위
Alice	150	1000초	629,146,600	1
Bob	150	500초	629,146,100	2
Carol	100	2000초	419,432,304	3

Alice와 Bob은 경험치가 같지만, Bob의 남은 시간이 적어(늦게 운동 시작) Score가 낮습니다. ZREVRANGE로 내림차순 조회하면 Alice가 1위가 됩니다.

구현 코드

Score 계산

from datetime import datetime

def calculate_score(xp: int, now: datetime, ended_at: datetime) -> int:
    """
    운동대결 리더보드 Score 계산

    - 상위 비트: 경험치 (높을수록 좋음)
    - 하위 22비트: 남은 시간 (많을수록 좋음 = 일찍 운동)
    """
    remaining_seconds = max(0, int(ended_at.timestamp() - now.timestamp()))
    return (xp << 22) + remaining_seconds

Redis 연동

import redis
from datetime import datetime

class LeaderboardService:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client

    def _get_key(self, competition_id: int) -> str:
        return f"leaderboard:competition:{competition_id}"

    def update_score(
        self,
        competition_id: int,
        user_id: int,
        xp: int,
        ended_at: datetime
    ) -> None:
        """사용자의 경험치 업데이트"""
        key = self._get_key(competition_id)
        score = calculate_score(xp, datetime.now(), ended_at)
        self.redis.zadd(key, {str(user_id): score})

    def get_rank(self, competition_id: int, user_id: int) -> int | None:
        """사용자의 순위 조회 (1부터 시작)"""
        key = self._get_key(competition_id)
        rank = self.redis.zrevrank(key, str(user_id))
        return rank + 1 if rank is not None else None

    def get_top_users(self, competition_id: int, count: int = 10) -> list[dict]:
        """상위 N명 조회"""
        key = self._get_key(competition_id)
        results = self.redis.zrevrange(key, 0, count - 1, withscores=True)

        return [
            {
                "user_id": int(user_id),
                "score": int(score),
                "rank": idx + 1
            }
            for idx, (user_id, score) in enumerate(results)
        ]

사용 예시

# 대회 종료 시간
ended_at = datetime(2025, 1, 1, 0, 0, 0)

# 사용자별 경험치 업데이트
leaderboard.update_score(competition_id=1, user_id=100, xp=150, ended_at=ended_at)
leaderboard.update_score(competition_id=1, user_id=101, xp=150, ended_at=ended_at)
leaderboard.update_score(competition_id=1, user_id=102, xp=100, ended_at=ended_at)

# 순위 조회
rank = leaderboard.get_rank(competition_id=1, user_id=100)
print(f"User 100의 순위: {rank}위")

# 상위 10명 조회
top_users = leaderboard.get_top_users(competition_id=1, count=10)
for user in top_users:
    print(f"{user['rank']}위: User {user['user_id']}")

고려사항

22비트 제한

하위 22비트로 표현 가능한 최대값은 2^22 - 1 = 4,194,303초 (약 48일)입니다.

# 대회 기간이 48일을 초과하면 오버플로우 발생
# 필요시 비트 수 조정 가능 (예: 24비트 = 약 194일)
MAX_SECONDS = (1 << 22) - 1  # 4,194,303

대회 종료 후 처리

대회가 종료된 시점에는 ended_at - now가 음수가 됩니다.

def calculate_score(xp: int, now: datetime, ended_at: datetime) -> int:
    remaining_seconds = int(ended_at.timestamp() - now.timestamp())
    # 음수 방지: 대회 종료 후에는 0으로 처리
    remaining_seconds = max(0, remaining_seconds)
    return (xp << 22) + remaining_seconds

Score에서 원래 값 추출

필요시 Score에서 경험치와 남은 시간을 역으로 추출할 수 있습니다.

def extract_from_score(score: int) -> tuple[int, int]:
    """Score에서 경험치와 남은 시간 추출"""
    xp = score >> 22
    remaining_seconds = score & ((1 << 22) - 1)  # 하위 22비트 마스킹
    return xp, remaining_seconds

TTL 설정

대회 종료 후 리더보드 데이터를 일정 기간 후 자동 삭제하도록 TTL을 설정할 수 있습니다.

def set_expiry(self, competition_id: int, days: int = 30) -> None:
    """대회 종료 후 N일 뒤 자동 삭제"""
    key = self._get_key(competition_id)
    self.redis.expire(key, days * 24 * 60 * 60)

향후 개선 방향: 순위 변동 알림

리더보드를 구현하고 나니, 자연스럽게 순위가 바뀌면 알림을 보내면 좋겠다는 요구사항이 생겼습니다. 하지만 우리 서비스는 운동 중 약 10초마다 경험치가 쌓이기 때문에, 매번 순위 변동을 체크하고 알림을 보내는 건 현실적이지 않습니다.

문제: 잦은 업데이트

매번 알림을 보내면 사용자에게 스팸이 되고, 서버 부하도 커집니다.

방안 1: 스냅샷 비교 (배치 처리)

일정 주기로 순위 스냅샷을 저장하고, 이전 스냅샷과 비교하여 변동이 있을 때만 알림을 보내는 방식입니다.

async def check_rank_changes(competition_id: int):
    key = f"leaderboard:competition:{competition_id}"
    snapshot_key = f"leaderboard:snapshot:{competition_id}"

    # 현재 순위 조회
    current_ranks = {
        user_id: rank
        for rank, user_id in enumerate(
            redis.zrevrange(key, 0, -1), start=1
        )
    }

    # 이전 스냅샷과 비교
    previous_ranks = redis.hgetall(snapshot_key)

    for user_id, new_rank in current_ranks.items():
        old_rank = int(previous_ranks.get(user_id, new_rank))
        if new_rank < old_rank:
            await notify_rank_up(user_id, old_rank, new_rank)

    # 스냅샷 갱신
    redis.hset(snapshot_key, mapping=current_ranks)

장점: 구현이 단순하고, 알림 빈도를 주기로 제어 가능
단점: 실시간성이 떨어짐 (5분~30분 주기)
적합한 경우: 대부분의 일반적인 리더보드

방안 2: 의미 있는 변동만 알림

모든 순위 변동이 아닌, 의미 있는 순간에만 알림을 보내는 방식입니다.

def should_notify(old_rank: int, new_rank: int) -> bool:
    # 10위권 진입
    if old_rank > 10 and new_rank <= 10:
        return True

    # 3위권 진입
    if old_rank > 3 and new_rank <= 3:
        return True

    # 1등 달성
    if new_rank == 1 and old_rank != 1:
        return True

    return False

장점: 정말 중요한 순간에만 알림 → 알림의 가치 상승
단점: 세부 순위 변동은 알 수 없음
적합한 경우: 경쟁이 치열한 상위권 알림

방안 3: 대회 종료 임박 시에만 알림

평소에는 알림을 보내지 않다가, 대회 종료 N시간 전부터만 순위 변동 알림을 활성화하는 방식입니다.

def is_notification_period(ended_at: datetime) -> bool:
    remaining = ended_at - datetime.now()
    # 종료 24시간 전부터만 알림 활성화
    return remaining <= timedelta(hours=24)

장점: 막판 역전의 긴장감 극대화
단점: 대회 초반~중반에는 알림이 없음
적합한 경우: 마지막 스퍼트가 중요한 대회

방안 비교

방안	실시간성	알림 빈도	구현 복잡도	서버 부하
스냅샷 비교	낮음	중간	낮음	낮음
의미 있는 변동만	높음	낮음	중간	중간
종료 임박 시만	조건부	낮음	낮음	낮음

현재 계획

우리 서비스에서는 방안들을 조합해서 사용할 예정입니다:

평소: 알림 없음 (앱에서 실시간 순위 확인 가능)
종료 24시간 전: 30분마다 스냅샷 비교
알림 조건: 10위권 진입, 1등 탈환/피탈 시에만

이렇게 하면 불필요한 알림은 줄이면서도, 중요한 순간에는 사용자에게 동기부여를 줄 수 있을 것으로 기대합니다.

마치며

Redis Sorted Set과 비트 연산을 활용하면 복합 정렬 조건을 가진 리더보드도 효율적으로 구현할 수 있습니다.

핵심 포인트:

단일 Score로 복합 정렬: 비트 연산으로 여러 기준을 하나의 숫자에 인코딩
실시간 순위 조회: O(log N) 시간복잡도로 빠른 조회 가능
자동 정렬: ZADD만으로 정렬까지 완료

실제 운영에서는 Redis 장애 대비, 데이터 백업, 모니터링 등을 추가로 고려해야 합니다.

Pak Min gi