devspoon 오픈소스 개발자 번뇌 일지

Database/redis

Redis 실무 가이드 3 - FastAPI로 구현하는 백엔드 실무 패턴 9선

bluebamus 2026. 9. 8.

목차

Part 5. [Pattern] 백엔드 실무 활용 패턴 구현 (with FastAPI)

준비 단계

실무 패턴 9선

Part 5. [Pattern] 백엔드 실무 활용 패턴 구현 (with FastAPI)

5-1. 가상환경부터 첫 엔드포인트까지 — FastAPI 셋업

1. 시작 전 갖춰야 할 것

파이썬 백엔드 개발을 위해 딱 두 가지만 먼저 확인하겠습니다.

  • Python 3.10 이상: 최신 버전(3.14 이상)을 추천합니다.
  • VS Code (Visual Studio Code): 가장 가볍고 강력한 무료 에디터입니다.
bash
# (참고) 현재 설치된 파이썬 버전 확인
python --version

2. 작업 폴더 만들기

C나 D 디렉터리 하위 편한 곳에 실습용 폴더를 하나 만들어주세요.

  • 폴더명 예시: fastapi-study

3. 가상환경 만들고 들어가기 ★가장 많이 틀리는 곳★

파이썬 가상환경(venv)

  • 파이썬의 가상 환경을 만들어 주는 표준 도구
  • 파이썬 프로젝트를 할 때 가장 중요한 습관

왜 필요한가요?

프로젝트마다 필요한 라이브러리 버전이 다를 수 있습니다. 이 프로젝트만을 위한 '독립된 방(격리된 공간)' 을 만드는 것입니다.

가상환경 없이 개발하면

  • (가정) 프로젝트 A → Django 3.2 필요, 프로젝트 B → Django 5.0 필요
  • 👉 하나 설치하면 다른 하나가 깨짐
  • 👉 라이브러리 지옥(dependency hell)

venv를 사용하면

  • 프로젝트 A → 자기만의 파이썬 + 라이브러리
  • 프로젝트 B → 완전히 다른 파이썬 환경
  • 👉 서로 절대 간섭 안 함

터미널에서 아래의 명령어를 차례대로 실행합니다.

bash
python -m venv venv
source venv/bin/activate
pip install fastapi uvicorn

Step 1. 가상환경 생성

터미널에서 아래 명령어를 입력합니다.

bash
# 윈도우/맥 공통
python -m venv venv

# (만약 맥에서 python 명령어가 안 먹히면 python3 -m venv venv)

Step 2. 가상환경 활성화 (방에 입장하기)

명령어를 입력했을 때 터미널 앞부분에 (venv)라고 뜨면 성공입니다!

bash
# Windows (cmd/powershell)
# 역슬래시(\) 주의!
venv\Scripts\activate

# 권한 오류가 나면 PowerShell을 관리자 권한으로 실행하거나
# Set-ExecutionPolicy RemoteSigned 설정이 필요할 수 있습니다
bash
# Mac / Linux
source venv/bin/activate
bash
# (참고) 가상환경 비활성화 (방에서 나가기)
deactivate

4. FastAPI와 서버 설치하기

가상환경에 들어온 상태((venv) 표시 확인)에서 라이브러리를 설치합니다.

bash
pip install fastapi "uvicorn[standard]"
  • fastapi: 웹 프레임워크 본체
  • uvicorn: ASGI 서버. FastAPI로 만든 웹 API를 실행시켜 줍니다.

[standard]는 uvicorn의 '풀옵션 세트'입니다. 개발할 때 필요한 기본 도구들을 한 번에 설치해 줍니다. (아래 패키지 같이 설치됨)

구성요소 역할
uvloop 고성능 이벤트 루프 (Linux/macOS)
httptools 빠른 HTTP 파서
watchfiles --reload 자동 재시작
python-dotenv .env 환경변수 지원
websockets WebSocket 지원

5. 여덟 줄로 끝내는 첫 엔드포인트

VS Code에서 main.py 파일을 만들고 아래 코드를 작성합니다.

python
from fastapi import FastAPI

# 1. 앱 생성
app = FastAPI()


# 2. 경로(Path) 동작 생성
@app.get("/")
def read_root():
    return {"Hello": "World"}
  • @app.get("/") : 사용자가 도메인 주소(http://.../)로 들어왔을 때, 아래 함수를 실행하라는 뜻입니다. (데코레이터)
  • return : 파이썬 딕셔너리(dict)를 리턴하면, FastAPI가 알아서 JSON 형태로 바꿔줍니다.

6. 서버 띄우고 브라우저로 확인하기

터미널에서 다음 명령어로 서버를 켭니다.

bash
uvicorn main:app --reload
  • main : 파일 이름 (main.py)
  • app : 파일 내에서 생성한 FastAPI 객체 변수명 (app = FastAPI())
  • --reload : 코드를 저장할 때마다 서버를 자동으로 재시작 (개발 모드 꿀기능!)

확인 방법

  1. 브라우저 주소창에 http://127.0.0.1:8000 입력
  2. 화면에 {"Hello":"World"}가 보이면 성공!

5-2. redis-py 비동기 연결을 FastAPI 수명주기에 붙이기

이 글에서 만들 연결 구조

  • 파이썬 공식 Redis 클라이언트 라이브러리(redis-py)를 설치하고 특징을 이해합니다.
  • FastAPI 애플리케이션에서 Redis 서버에 연결하는 객체를 생성합니다.
  • 가장 기본적인 데이터 조작 명령어인 SET과 GET을 API 엔드포인트로 구현해 봅니다.

1. redis-py 설치와 동기·비동기 선택

이전 강의에서 준비한 가상환경(venv)이 활성화되어 있는지 확인한 후, 터미널에 아래 명령어를 입력하여 파이썬용 레디스 라이브러리를 설치합니다.

bash
# 가상환경 활성화
venv\Scripts\activate

# 파이썬용 Redis 클라이언트 라이브러리 설치
pip install redis

🛠 redis-py 라이브러리 소개

파이썬 생태계에는 여러 Redis 라이브러리가 있지만, redis 패키지(redis-py 프로젝트)를 사용하는 이유는 다음과 같습니다.

  • 공식 클라이언트 — Redis에서 공식적으로 지원하는 파이썬 클라이언트 라이브러리입니다.
  • 가장 널리 사용 — 현업 파이썬 백엔드 생태계에서 가장 널리 사용되는 사실상의 표준(Standard)입니다.
  • Sync / Async 모두 지원 — 동기(Synchronous) 방식과 비동기(Asynchronous) 방식을 하나의 라이브러리에서 모두 지원합니다.
  • aioredis 통합 — 과거에는 비동기 처리를 위해 aioredis라는 라이브러리를 따로 썼지만, 최근 redis-py에 해당 기능이 완벽하게 통합되어 이제는 이것 하나만 설치하면 충분합니다!

2. [실습] lifespan에서 커넥션 풀 열고 닫기

이전 시간에 만든 main.py 파일을 열고, Redis 연결하는 코드를 추가해 봅니다. FastAPI는 비동기 프레임워크이므로 redis.asyncio 모듈을 사용하며, lifespan과 app.state를 통해 서버의 시작/종료 주기에 맞춰 안전하게 연결을 관리합니다.

📢 앞으로 진행될 백엔드 패턴 실습(Pattern 01~07) 코드에서도 지금 배우는 이 "app.state" 방식을 표준 연결 패턴으로 계속 사용할 예정입니다.

python
from contextlib import asynccontextmanager
from fastapi import FastAPI
import redis.asyncio as redis  # 비동기 모듈을 임포트합니다!


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 1. 서버 시작 시: Redis 연결 풀(Pool) 생성 후 app.state에 저장
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    print("✅ Redis 연결 성공!")

    yield

    # 2. 서버 종료 시: Redis 연결 안전하게 해제
    await app.state.redis.aclose()
    print("❌ Redis 연결 해제!")


# lifespan을 FastAPI 앱에 등록
app = FastAPI(lifespan=lifespan)

💡 코드 핵심 설명

lifespan - FastAPI 애플리케이션의 시작과 종료 시점에 실행되는 초기화 로직을 정의하는 기능입니다. - 서버 시작 시 yield 위의 코드가 실행되고, 서버 종료 시 yield 아래의 코드가 실행됩니다. - 실무에서 DB 연결, Redis 연결, 외부 서비스 클라이언트 등 애플리케이션에서 공유할 자원을 초기화하고 정리할 때 사용하는 권장 패턴입니다.

app.state.redis - FastAPI 앱 객체의 상태(state) 공간에 Redis 연결 객체를 보관합니다. 이렇게 하면 애플리케이션 전체에서 동일한 Redis 연결 객체를 공유하여 사용할 수 있습니다.

redis.from_url(...) - redis://호스트:포트/DB번호 형태로 연결할 Redis의 IP와 포트번호와 DB 번호를 지정합니다. - DB 번호: Redis는 기본적으로 0번부터 15번까지 16개의 논리적 데이터베이스를 제공합니다. 하지만 실무에서는 대부분 0번 DB만 사용합니다. 간혹 개발/테스트 환경을 DB 번호로 나누어 사용하기도 합니다.

decode_responses=True - 이 옵션을 넣지 않으면 Redis에서 가져온 데이터가 파이썬에서 바이트 형태(b'Hello')로 출력됩니다. 이 옵션을 켜두면 우리가 아는 일반 문자열('Hello')로 자동 변환해 줍니다.

💡 [실무 Tip] Connection Pool

Redis 클라이언트는 내부적으로 Connection Pool을 관리하기 때문에 요청마다 새로 생성하지 않고 애플리케이션 전체에서 공유하는 것이 일반적인 패턴입니다.

3. 💡 파이썬에서 쓰는 Redis 기본 호출

본격적으로 API를 만들기 전에, redis-py 라이브러리에서 Redis 명령어를 어떻게 쓰는지 알아봅시다. 규칙은 아주 간단합니다. 우리가 터미널(CLI)에서 사용하던 명령어의 소문자 형태가 곧 파이썬 메서드 이름이며, 앞에 await만 붙여주면 됩니다!

기능 Redis CLI 명령어 파이썬 비동기 문법 (rd 객체)
데이터 저장 SET key value await rd.set('key', 'value')
데이터 조회 GET key await rd.get('key')
데이터 삭제 DEL key await rd.delete('key')
만료 시간(TTL) EXPIRE key 60 await rd.expire('key', 60)
만료 동시 저장 SET key value EX 60 await rd.set('key', 'value', ex=60)
숫자 증가 INCR key await rd.incr('key')

🚨 [주의] SETEX는 안 쓰나요?

과거에는 데이터를 넣으면서 동시에 만료 시간을 정할 때 SETEX나 SETNX 같은 명령어를 썼습니다. 하지만 이러한 명령어들이 향후 삭제(Deprecated)될 가능성이 있습니다.

따라서 공식 권장 사항에 맞게 SET 명령어에 옵션(EX, NX 등)을 추가하는 방식(rd.set(..., ex=60))을 사용하는 것이 실무 표준입니다!

이후 배우게 될 리스트(await rd.lpush()), 해시(await rd.hset()) 등도 이와 같은 1:1 직관적인 매핑 규칙을 따릅니다.

4. [실습] SET·GET을 API 두 개로 노출하기

이제 뼈대 코드 위에, Redis에 간단한 String 타입의 값을 넣고(SET), 그 값을 다시 조회하는(GET) API를 추가해 보겠습니다. 비동기 객체를 사용하므로 함수 앞에 async def를 붙이고, 레디스 명령어 앞에 await를 붙여야 합니다.

python
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    print("✅ Redis 연결 성공!")

    yield

    await app.state.redis.aclose()
    print("❌ Redis 연결 해제!")


app = FastAPI(lifespan=lifespan)


@app.post("/items/{item_id}")
async def set_item(item_id: str, value: str, request: Request):
    """
    Redis에 데이터를 저장합니다. (SET)
    """
    # Request 객체를 통해 app.state에 저장된 레디스 연결을 가져옵니다.
    rd = request.app.state.redis
    key = f"item:{item_id}"

    # await rd.set(키, 값)
    await rd.set(key, value)

    return {"message": "Data saved to Redis successfully", "key": key, "value": value}


@app.get("/items/{item_id}")
async def get_item(item_id: str, request: Request):
    """
    Redis에서 데이터를 조회합니다. (GET)
    """
    rd = request.app.state.redis
    key = f"item:{item_id}"

    # await rd.get(키)
    result = await rd.get(key)

    # Redis에 데이터가 없으면 None을 반환합니다.
    if result is None:
        return {"error": "Item not found in Redis"}

    return {"key": key, "value": result}

5. Swagger UI에서 바로 눌러 확인하기

코드를 모두 작성했다면 터미널에서 Uvicorn을 이용해 서버를 띄워봅시다.

bash
uvicorn main:app --reload
  1. 웹 브라우저를 열고 http://localhost:8000/docs (Swagger UI)에 접속합니다.
  2. POST /items/{item_id} API를 열고 item_id에 "1", value에 "Apple"을 입력하고 실행(Execute)합니다.
  3. GET /items/{item_id} API를 열고 item_id에 "1"을 입력하여 방금 저장한 "Apple"이 레디스에서 잘 조회되는지 확인합니다.

🚩 정리 — 연결 객체를 전역에 두면 안 되는 이유

  • 파이썬에서 Redis와 통신할 때는 redis-py 라이브러리를 사용하며, 명령어는 우리가 CLI에서 배운 것과 거의 동일합니다.
  • FastAPI 환경에서는 lifespan과 app.state를 사용하여 Redis 비동기 연결 풀을 애플리케이션 시작 시 한 번 생성하고 모든 API 요청에서 공유하여 재사용하는 패턴을 사용합니다.
  • 축하합니다! 이제 여러분의 백엔드 서버와 Redis 서버가 파이프라인으로 연결되었습니다.

5-3. 패턴 01: Cache-Aside — DB 앞에 캐시를 세우는 기본형

이 글에서 구현할 캐시 흐름

  • 현업에서 가장 많이 쓰이는 캐싱 전략인 Cache-Aside(Look-aside) 패턴을 이해합니다.
  • FastAPI와 비동기 Redis(app.state)를 연동하여 실제 서비스 로직을 구현하는 법을 익힙니다.
  • 캐시 히트(Hit)와 캐시 미스(Miss)의 개념을 파악하고, 캐시 무효화(Invalidation)의 중요성을 배웁니다.

1. 캐시를 먼저 보고 없으면 DB로 가는 방식

Cache-Aside 패턴은 애플리케이션이 데이터를 찾을 때 먼저 캐시(Redis)를 확인하고, 데이터가 없으면 DB를 조회하여 가져오는 가장 기본적이고 널리 쓰이는 방식입니다.

즉, 첫 요청은 느릴 수 있지만 두 번째부터 빨라지는 구조가 Cache-Aside의 핵심입니다.

📖 데이터 조회 흐름

  1. 애플리케이션이 캐시(Cache) 를 먼저 조회
  2. 캐시에 데이터가 있으면 캐시에서 반환 (Cache Hit 🚀)
  3. 캐시에 없으면 DB 조회 (Cache Miss 🐌) 후 반환
  4. (이때) DB에서 조회한 데이터를 캐시에 저장
  5. 이후 요청부터는 캐시 사용

2. 요청이 캐시와 DB를 오가는 순서

text
    [ 앱 ]              [ Redis (Cache) ]         [ 메인 DB ]
      |                      |                      |
      |--- (1) 데이터 요청 -->|                      |
      |                      |                      |
      |<-- (2) 데이터 존재? --|                      |
      |      (Cache Hit!)    |                      |
      |                      |                      |
      |    (3) 데이터 없음     |                      |
      |      (Cache Miss)    |                      |
      |                      |                      |
      |    (4) DB에서 조회 ------------------------->|
      |                      |                      |
      |    (5) 캐시에 저장 -->|                      |
      |     (SET with TTL)   |                      |
      |                      |                      |
      |<-- (6) 최종 데이터 반환 ---------------------|
  • Cache Hit: (1) → (2)에서 즉시 응답이 끝납니다. (초고속)
  • Cache Miss: (1) → (3) → (4) → (5) → (6)의 전체 경로를 탑니다. 첫 요청은 느리지만, (5)번 과정으로 다음 요청부터는 Cache Hit가 됩니다.

3. 어떤 데이터에 적용하면 이득인가

✅ 장점

  • DB 부하 감소: 자주 조회되는 데이터를 캐싱하여 메인 DB 서버의 부하를 줄일 수 있습니다.
  • 응답 속도 향상: 메모리 기반인 Redis에서 매우 빠르게 응답하므로 사용자 경험(UX)이 크게 개선됩니다.
  • 캐시는 선택적(Optional) 계층: Redis 장애 시에도 애플리케이션은 DB를 직접 조회하여 동작할 수 있습니다. 다만 성능 저하와 DB 부하 증가는 발생할 수 있습니다.

🎯 어떤 데이터를 캐싱해야 할까? (도입 기준)

모든 데이터를 캐싱할 필요는 없습니다. 아래 2가지 조건을 만족하는 데이터가 캐싱의 1순위 타겟입니다.

  1. 읽기(Read) 비중이 압도적으로 높은 데이터 (예: 공지사항, 설정 값, 상품 상세 정보, 약관)
  2. 자주 바뀌지 않는 데이터 (1분에 수십 번씩 바뀌는 데이터는 캐시를 갱신하는 비용이 더 큽니다)

4. [실습] 사용자 프로필 조회에 캐시 붙이기

이전 시간에 배운 비동기(Async) 및 app.state 패턴을 적용하여 실제 서비스 수준의 코드를 작성해 봅시다. 객체 데이터를 통째로 넣기 위해 json 모듈을 활용합니다.

python
import asyncio
import json
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi import HTTPException
import redis.asyncio as redis

# 가상의 DB 데이터 (실제로는 MySQL, PostgreSQL 등에서 가져옵니다)
fake_db = {
    "user:1": {"name": "Kim", "email": "kim@example.com", "tier": "Gold"},
    "user:2": {"name": "Lee", "email": "lee@example.com", "tier": "Silver"},
}


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 서버 시작 시 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)

    yield

    # 서버 종료 시 안전하게 해제
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/users/{user_id}")
async def get_user_profile(user_id: str, request: Request):
    rd = request.app.state.redis

    cache_key = f"user:profile:{user_id}"

    # 1. 레디스에서 캐시 확인
    cached_user = await rd.get(cache_key)

    if cached_user:
        # 캐시 Hit
        print(f"🚀 Cache Hit! (user_id: {user_id})")

        # Redis에 저장된 JSON 문자열을 파이썬 딕셔너리로 변환하여 반환
        return json.loads(cached_user)

    # 2. 캐시 Miss 시 실제 DB 조회
    print(f"🐌 Cache Miss! Fetching from DB... (user_id: {user_id})")

    user_data = fake_db.get(f"user:{user_id}")
    if not user_data:
        raise HTTPException(status_code=404, detail="User not found")

    # DB 조회가 느리다고 가정 (2초 대기 시뮬레이션)
    await asyncio.sleep(2)

    # 3. Redis에 데이터 저장 (TTL 300초 설정)
    # 파이썬 딕셔너리를 JSON 문자열로 변환(dumps)하여 저장합니다.
    # [주의] setex 대신 최신 권장 문법인 set(..., ex=...)을 사용합니다.
    await rd.set(cache_key, json.dumps(user_data), ex=300)

    return user_data

5. [심화] 수정이 일어났을 때 캐시를 어떻게 할까

Cache-Aside 패턴에서 가장 무서운 적은 "데이터 불일치(Stale Data)" 입니다.

만약 유저가 자신의 이름을 'Kim'에서 'Park'으로 수정했는데, Redis에는 여전히 'Kim'이 300초 동안 남아있다면 유저는 큰 혼란을 겪게 됩니다.

이를 방지하기 위해, DB 데이터가 수정/삭제될 때는 반드시 Redis의 캐시도 함께 지워줘야 합니다. 이를 '캐시 무효화(Cache Invalidation)' 라고 부릅니다.

python
from pydantic import BaseModel


# 정보 업데이트를 위한 Request Body 스키마 정의 (DB 스키마와 일치)
class UserProfileUpdate(BaseModel):
    name: str
    email: str
    tier: str


@app.put("/users/{user_id}")
async def update_user_profile(user_id: str, profile: UserProfileUpdate, request: Request):
    rd = request.app.state.redis
    cache_key = f"user:profile:{user_id}"

    # 1. 실제 DB 데이터 업데이트 (가상)
    if f"user:{user_id}" in fake_db:
        fake_db[f"user:{user_id}"]["name"] = profile.name
        fake_db[f"user:{user_id}"]["email"] = profile.email
        fake_db[f"user:{user_id}"]["tier"] = profile.tier

        # 2. [핵심] 기존 캐시 삭제 (Cache Invalidation)
        # 다음 요청 시 Cache Miss가 발생하여, DB의 최신 데이터를 다시 캐싱하게 만듭니다.
        await rd.delete(cache_key)

        return {
            "message": "updated successfully",
            "current_data": fake_db[f"user:{user_id}"],
        }

    raise HTTPException(status_code=404, detail="User not found")

🚩 정리 — Cache-Aside가 남기는 숙제

  • Cache-Aside는 가장 범용적이고 널리 쓰이는 표준 캐싱 전략입니다.
  • 읽기(조회) 트래픽이 많고 자주 바뀌지 않는 데이터에 특히 효과적입니다.
  • 첫 요청은 느릴 수 있지만, 이후 요청은 Cache Hit로 매우 빠르게 처리할 수 있습니다.
  • 다만 데이터 변경 시 캐시 무효화(Cache Invalidation)를 놓치면 stale data 문제가 발생할 수 있으므로 주의해야 합니다.

5-4. 패턴 02: List로 만드는 '최근 본 상품' 5개

이 글에서 만들 기능

  • Redis Lists를 활용하여 사용자의 활동 이력(최근 본 상품 등)을 관리하는 법을 배웁니다.
  • LPUSH와 LTRIM을 조합하여 메모리를 낭비하지 않는 고정 크기 큐(Fixed-size Queue) 패턴을 익힙니다.
  • 중복 데이터 처리(LREM) 로직을 포함한 실제 비동기 FastAPI 코드를 구현합니다.

1. 중복 제거·최신순·개수 제한을 명령어 세 개로

이커머스 앱에서 "최근 본 상품" 은 항상 최신 항목이 맨 앞에 오고, 일정 개수(예: 5개)가 넘어가면 가장 오래된 항목이 삭제되는 구조를 가집니다.

RDB로도 구현은 가능하지만, 별도 테이블 관리, 정렬 조회, 중복 처리, 오래된 기록 정리 등을 함께 고려해야 해서 로직이 상대적으로 무거워질 수 있습니다. 반면 Redis Lists는 이 패턴을 훨씬 단순하게 구현할 수 있습니다.

📖 흐름도 (Workflow)

text
[ 유저 / 앱 ]            [ FastAPI 서버 ]          [ Redis (Lists) ]
      |                       |                        |
      |-- (1) 상품 'C' 조회 -->|                        |
      |                       |                        |
      |                       |-- (2) 중복 제거 ------->|
      |                       |    (LREM: 기존 C 삭제)  |
      |                       |                        |
      |                       |-- (3) 최상단 추가 ----->|
      |                       |    (LPUSH: 맨 앞 삽입)  |
      |                       |                        |
      |                       |-- (4) 큐 크기 제한 ---->|
      |                       |    (LTRIM: 5개만 유지)  |
      |                       |                        |
      |<-- (5) 성공 응답 ------|                        |

📖 리스트 내부 상태 변화

text
[ 유저가 새로운 상품 'E'를 조회함 ]
1. 기존 리스트 상태  : [D, C, B, A]
2. LPUSH E 실행      : [E, D, C, B, A]  (왼쪽, 즉 맨 앞에 추가됨)
3. LTRIM 0 4 실행    : [E, D, C, B, A]  (인덱스 0~4번까지만 남기고 자름. 5개 유지)

[ 연달아 상품 'F'를 조회한다면? ]
4. LPUSH F 실행      : [F, E, D, C, B, A]  (총 6개가 됨)
5. LTRIM 0 4 실행    : [F, E, D, C, B]     (가장 오래된 A가 자동으로 삭제됨!)

2. [실습] 조회 시 기록하고 최신순으로 꺼내기

실무에서는 단순히 넣고 자르는 것뿐만 아니라, "이미 리스트에 있는 상품을 다시 봤을 때" 중복을 제거하고 맨 앞으로 보내는 디테일이 중요합니다.

python
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 서버 시작 시 비동기 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)

    yield

    # 서버 종료 시 안전하게 해제
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


# 최근 본 상품 업데이트
@app.post("/products/{product_id}/view")
async def view_product(product_id: str, request: Request, user_id: str = "user_1"):
    rd = request.app.state.redis

    key = f"user:{user_id}:recent_views"

    # 1. 기존 리스트에서 동일한 ID가 있다면 제거 (중복 방지 및 끌어올리기)
    # LREM [key] [count] [value] : count가 0이면 일치하는 모든 항목을 삭제합니다.
    await rd.lrem(key, 0, product_id)

    # 2. 최신 상품 ID를 왼쪽(맨 앞)에 추가
    await rd.lpush(key, product_id)

    # 3. 최신 5개만 남기고 자르기 (Fixed-size 큐 유지)
    # 인덱스는 0부터 시작하므로 0~4는 총 5개를 의미합니다.
    await rd.ltrim(key, 0, 4)

    return {"message": f"Product {product_id} added to recent views"}


# 최근 본 상품 조회
@app.get("/users/{user_id}/recent-views")
async def get_recent_views(user_id: str, request: Request):
    rd = request.app.state.redis
    key = f"user:{user_id}:recent_views"

    # 리스트 전체(인덱스 0부터 -1까지) 조회
    views = await rd.lrange(key, 0, -1)

    return {"recent_views": views}

3. RDB로 짰을 때와 비교하면

✅ 1) 압도적인 성능

메인 DB에서 유저별로 수백, 수천 개의 조회 이력을 저장하고, 이를 최신순으로 정렬(ORDER BY)하여 상위 5개만 뽑아내는 작업은 유저가 많아질수록 DB에 많은 부하를 줄 수 있습니다. Redis Lists를 활용하면 빠른 속도로 이 작업을 끝냅니다.

🗣 LPUSH와 LTRIM은 매우 효율적이며, 최근 5개처럼 짧은 리스트를 유지하면 전체적으로 매우 빠르게 동작합니다. 다만 중복 제거용 LREM은 값을 탐색해야 하므로 리스트 길이에 비례하는 비용이 있습니다. 그래서 이 패턴은 리스트 길이를 작게 제한할 때 특히 적합합니다.

✅ 2) 단순하고 우아한 로직

LPUSH + LTRIM 조합만으로 "최신순 정렬"과 "용량(메모리) 관리"가 동시에 해결됩니다. 오래된 데이터를 지우기 위한 별도의 삭제 배치(Batch) 작업이 전혀 필요 없습니다.

4. [고급] 중복을 지우는 LREM 한 줄의 차이

위 코드에서 lrem을 가장 먼저 실행한 이유는 무엇일까요? 만약 유저가 A -> B -> C 순으로 상품을 구경했는데, 다시 C 상품을 클릭했다고 가정해 봅시다.

  • LREM을 안 썼을 때: [C, C, B, A] — 결과에 C가 중복으로 노출되며, 한정된 5칸의 공간을 낭비하게 됩니다.
  • LREM을 썼을 때: [C, B, A] — 기존에 있던 C를 지우고 새로 맨 앞에 C를 넣기 때문에, 중복 없이 해당 상품이 최상단으로 '끌어올려'집니다.

특히 최근 본 상품은 보통 10개 이하처럼 작은 크기로 제한하는 경우가 많아, LREM의 선형 탐색 비용도 실무에서 충분히 감당 가능한 경우가 많습니다.

🚩 정리 — List 기반 목록의 한계

  • Lists는 데이터가 들어온 순서가 엄격하게 보장되어야 하는 이력(History) 관리에 최적의 자료형입니다.
  • 데이터가 무한정 늘어나 메모리를 터뜨리는 것을 방지하기 위해, List를 쓸 때는 항상 LTRIM을 세트로 묶어서 습관적으로 사용해야 합니다.
  • 만약 "최근 본 상품" 기록이 서버가 꺼져도 영구히 보존되어야 할 만큼 중요한 데이터이거나 분석 데이터로 활용해야 한다면, 별도로 메인 DB나 로그 저장소에 비동기 적재하는 구조를 함께 둘 수 있습니다.

5-5. 패턴 03: 서버를 늘려도 로그인이 유지되는 세션 저장소

이 글에서 해결할 상태 문제

  • 단일 서버 세션의 한계와 분산 세션(Distributed Session) 의 필요성을 이해합니다.
  • Redis를 활용해 여러 서버가 공유하는 중앙 집중형 세션 저장소를 설계합니다.
  • 세션 유지와 보안을 위한 TTL(만료 시간) 설정 및 FastAPI 구현법을 익힙니다.

1. 세션을 서버 밖으로 빼낸다는 것

여러 대의 서버가 동작하는 환경에서, 사용자가 어떤 서버에 접속하더라도 동일한 로그인 상태를 유지할 수 있도록 세션 데이터를 중앙 저장소에서 공유하는 아키텍처 패턴입니다.

❓ 왜 Redis를 세션 저장소로 쓰나요?

서버가 2대 이상인 환경에서는 유저의 로그인 정보가 특정 서버의 메모리에만 있다면, 유저가 로드 밸런서를 통해 다른 서버로 접속했을 때 로그인이 풀려버리는 문제가 발생합니다. Redis는 인메모리 DB라 매우 빠르고, TTL 기반으로 세션을 자동 만료시킬 수 있기 때문에 세션 저장소로 가장 많이 사용됩니다.

💡 아키텍처 비교

  1. Sticky Session (특정 서버 고정 방식)
    • 유저를 특정 서버에만 고정시키는 방식입니다.
    • 사용자가 특정 서버에 고정되기 때문에 트래픽이 불균형해질 수 있고, 서버 한 대가 죽으면 해당 서버의 유저들은 모두 로그아웃되는 치명적인 단점이 있습니다.
  2. Session Store (분산 세션 방식)
    • 모든 서버가 무상태(Stateless) 아키텍처를 유지하며 중앙의 Redis를 바라봅니다.
    • 유저가 어느 서버에 접속하든 Redis에서 세션을 조회하므로 로그인 상태가 안전하게 유지됩니다.

2. [실습] 로그인·조회·로그아웃 API 만들기

세션 데이터는 보통 유저 ID, 유저등급, 마지막 접속 IP 등을 포함하는 객체 형태이므로 Hash 자료구조가 가장 적합합니다. 우리가 배운 비동기 패턴과 Pydantic을 활용하여 작성해 봅시다.

python
import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, Response, Cookie, Request
from pydantic import BaseModel
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 서버 시작 시 비동기 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)

    yield

    # 서버 종료 시 안전하게 해제
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)

SESSION_EXPIRE = 3600  # 세션 만료 시간: 1시간(초)


class LoginRequest(BaseModel):
    user_id: str


@app.post("/login")
async def login(req_data: LoginRequest, response: Response, request: Request):
    rd = request.app.state.redis

    # 1. 고유하고 추측 불가능한 세션 ID 생성
    session_id = str(uuid.uuid4())
    session_key = f"session:{session_id}"

    # 2. 유저 정보를 Hashes로 저장
    user_info = {
        "user_id": req_data.user_id,
        "tier": "Premium",
        "ip": request.client.host if request.client else "127.0.0.1",
    }
    await rd.hset(session_key, mapping=user_info)

    # 3. 세션 만료 시간 설정 (1시간)
    await rd.expire(session_key, SESSION_EXPIRE)

    # 4. 클라이언트 쿠키에 세션 ID 전달 (보안을 위해 HttpOnly, Secure, SameSite 설정)
    # 주의: 로컬 개발 환경(HTTP)에서는 secure=False로 테스트해야 할 수 있습니다.
    response.set_cookie(
        key="session_id",
        value=session_id,
        httponly=True,
        secure=False,     # 실제 운영환경에서는 True로 설정해야 합니다
        samesite="lax",
    )

    return {"message": "Login Success", "session_id": session_id}


# Cookie(None)을 사용하면 FastAPI가 클라이언트의 '쿠키'에서 해당 값을 자동으로 추출합니다.
@app.get("/me")
async def get_my_info(request: Request, session_id: str | None = Cookie(None)):
    if not session_id:
        return {"error": "Not logged in"}

    rd = request.app.state.redis
    session_key = f"session:{session_id}"

    # 1. 레디스에서 세션 정보 조회
    user_info = await rd.hgetall(session_key)

    if not user_info:
        return {"error": "Session expired or invalid"}

    # 2. 활동 중이므로 세션 시간 연장 (Sliding Window 방식)
    await rd.expire(session_key, SESSION_EXPIRE)

    return user_info

3. 세션 TTL과 쿠키에서 챙길 것

✅ 1) Sliding Window 방식의 만료 시간

유저가 GET /me 등의 API를 호출할 때마다 EXPIRE 명령어를 다시 실행해서 만료 시간을 초기화합니다. 이렇게 하면 유저가 서비스를 활발히 이용하는 동안에는 로그인이 유지되고, 장시간 활동이 없는 세션만 자동 만료시켜서 보안과 편의성을 모두 잡을 수 있습니다.

✅ 2) 보안 (HttpOnly & Secure & SameSite)

세션 ID는 탈취당하면 내 계정을 그대로 빼앗기는 민감한 정보입니다. 쿠키를 생성할 때 반드시 다음 보안 옵션들을 설정해야 합니다.

  • httponly=True : 자바스크립트(XSS 공격)로 세션 ID를 읽어가지 못하도록 차단합니다.
  • secure=True : 암호화된 HTTPS 통신 환경에서만 쿠키가 전송되도록 하여, 네트워크 중간에서 패킷을 가로채는 공격(Sniffing)을 방어합니다.
  • samesite=lax : 크로스 사이트(cross-site) 요청에서 쿠키 전송을 제한하여 CSRF 공격을 완화하는 옵션입니다.

✅ 3) Session 키 구조

Redis에 세션을 저장할 때 키(key)를 session:{session_id} 형태로 저장합니다. 이렇게 하면 SCAN session:* 명령어로 관리하기가 편리해집니다.

4. [심화] String·Hash·RedisJSON 중 무엇에 담을까

유저의 권한 목록(Roles)이나 설정(Preferences) 등 세션 정보가 복잡해질 때, 어떤 자료구조를 쓰는 것이 가장 좋을까요? 실무에서는 다음 3단계를 거쳐 발전해 왔습니다.

  1. Strings (JSON 직렬화 방식)
    • 세션 전체를 json.dumps()로 구워 저장합니다. 구현은 가장 쉽습니다.
    • 단점: last_login_time 같은 필드 하나만 바꿀 때도 전체 데이터를 다시 파싱하고 덮어써야 하는 비효율이 있습니다.
  2. Hash (현재 실습 방식)
    • 항목들을 개별 필드로 관리하여 특정 필드만 즉시 수정(HSET)할 수 있습니다.
    • 단점: 구조가 기본적으로 평면적(1단계 Key-Field-Value)이어서 리스트나 중첩된(Nested) 객체를 직접적으로 표현하기는 어렵습니다.
  3. RedisJSON (✨ 최신 실무 트렌드)
    • 우리가 이전 레슨에서 배운 Redis Stack의 핵심 기능입니다!
    • 복잡한 JSON의 중첩된 객체 구조를 그대로 저장하면서도, 특정 경로(Path)의 값만 수정할 수 있어 복잡한 세션 관리에 가장 완벽한 대안으로 꼽힙니다.

🛠 지식 박스: 세션(Session) 인증 vs JWT 인증

우리가 방금 구현한 방식은 Redis라는 저장소에 유저의 상태를 보관하고 확인하는 세션(Stateful) 인증 방식입니다. 최근에는 서버에 상태를 저장하지 않는 JWT(JSON Web Token, Stateless) 방식도 많이 사용됩니다.

Q. 그럼 JWT를 쓰면 레디스가 필요 없을까요? A. JWT를 쓰더라도 Redis를 병행하는 경우가 있습니다.

  • JWT의 단점: 한 번 발급된 토큰은 만료 시간이 지날 때까지 서버가 강제로 정지시킬 수 없습니다. (예: 해커에게 토큰이 탈취되거나, 유저가 '모든 기기에서 로그아웃'을 눌렀을 때)
  • Redis의 역할: 이를 해결하기 위해 실무에서는 강제로 만료시켜야 할 JWT 토큰을 Redis에 올려두는 블랙리스트(Blacklist) 패턴이나, 보안을 위한 리프레시 토큰(Refresh Token) 저장소로 Redis를 병행해서 사용하는 경우가 있습니다.

🚩 정리 — 세션 저장소 설계 기준

  • 분산 세션은 서버 확장(Scale-out) 시 서버 간 인증 공유를 위해 도입하는 아키텍처입니다.
  • Redis의 In-memory 특성 덕분에 매 API 요청마다 세션을 조회하더라도 서비스에 지연 시간(Latency)이 거의 발생하지 않습니다.
  • 세션 데이터는 만에 하나 사라져도 치명적이지 않으므로(사용자가 다시 로그인하면 됨), 무거운 메인 DB보다 가볍고 빠른 Redis가 압도적으로 훌륭한 저장소입니다.

5-6. 패턴 04: INCR로 조회수를 한 건도 놓치지 않기

이 글에서 없앨 유실

  • 멀티 스레드/멀티프로세스 환경에서 발생하는 경쟁 상태(Race Condition) 의 위험성을 이해합니다.
  • Redis의 원자적 연산(Atomic Operation) 이 어떻게 데이터 부정합을 해결하는지 배웁니다.
  • FastAPI 환경에서 수천 건의 동시 요청에도 안전한 조회수/좋아요 로직을 구현합니다.

1. 문제 상황: 읽고-더하고-쓰는 사이에 끼어드는 요청

인기 게시글에 100명이 동시에 '좋아요'를 눌렀다고 가정해 봅시다. 다음과 같이 DB 업데이트 로직(Read-Modify-Write)을 구현했다면 값이 비정상적으로 변경될 수 있습니다.

⚠ 경쟁 상태(Race Condition) 발생 시나리오

text
서버 A: (1) DB 조회 (views: 100) -------------------------> (3) DB 저장 (100+1=101)
서버 B:               (2) DB 조회 (views: 100) --------> (4) DB 저장 (100+1=101)

결과: 2명이 '좋아요'를 눌렀지만, 최종 숫자는 102가 아닌 101이 됩니다. 여러 요청이 동시에 같은 데이터를 읽고 수정하면서 마지막에 저장한 값이 이전 변경을 덮어써 버리는 현상을 Race Condition(경쟁 상태)라고 합니다.

🛠 지식 박스: RDB의 UPDATE count = count + 1은 안 되나요?

사실 MySQL 등 RDB에서도 이렇게 쿼리를 짜면 데이터 정합성은 지킬 수 있습니다. DB가 해당 행(Row)에 락(Lock)을 걸어 순서대로 처리하기 때문입니다. 하지만 실무에서 Redis를 쓰는 진짜 이유는 따로 있습니다.

  1. 잠금 경합(Lock Contention): 수만 명이 동시에 같은 글에 '좋아요'를 누르면, RDB는 한 번에 하나의 쓰기만 허용하기 위해 줄을 세웁니다. 이 과정에서 다른 일반 조회 쿼리들까지 덩달아 멈추는(Blocking) 병목 현상이 발생합니다.
  2. 디스크 I/O vs 메모리: RDB는 매번 디스크에 기록하려 하지만, Redis는 메모리에서만 연산하므로 수십 배 이상 압도적으로 빠릅니다.
  3. 아키텍처 효율성: 단순한 카운팅 때문에 비싼 RDB의 CPU 자원을 소모하는 것보다, 가볍고 빠른 Redis에 맡기는 것이 좋습니다.

2. 해결책: 서버가 한 번에 처리하는 원자적 증가

Redis의 INCR 명령어는 "값을 읽고, 1을 더하고, 다시 저장하는" 전체 과정을 단 하나의 끊어지지 않는 단계(Atomic)로 처리합니다.

💡 Redis의 INCR 동작 방식

text
[ 요청 A, B가 동시에 도달함 ]
          ↓
[ Redis (싱글 스레드 기반) ]
  -> (1) 요청 A 처리: INCR 실행 (101 반환)
  -> (2) 요청 B 처리: INCR 실행 (102 반환)
          ↓
[ 최종 결과: 102 (정합성 보장!) ]
  • 원리: Redis는 기본적으로 명령어를 싱글 스레드에서 하나씩 직렬로 처리합니다. 아무리 많은 요청이 동시에 몰려와도 Redis가 알아서 줄을 세워 순차적으로 계산합니다.
  • 특징: 앱에서 "현재 값을 읽어올 필요 없이" 무조건 1을 증가시키므로 데이터가 꼬일 일이 없습니다.

3. [실습] 조회수 API를 INCR로 다시 짜기

조회수는 읽기와 쓰기 빈도가 매우 높으므로 Redis에서 모든 것을 처리하는 것이 효율적입니다.

python
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 비동기 Redis 연결 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    yield
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/articles/{article_id}/view")
async def increase_view_count(article_id: str, request: Request):
    rd = request.app.state.redis
    view_key = f"article:{article_id}:views"

    # 1. 원자적으로 값 증가 (INCR)
    # 별도로 값을 GET 해서 더할 필요가 없습니다. (Race Condition 완벽 차단)
    current_views = await rd.incr(view_key)

    # 2. 특정 수치마다 DB에 백업하는 로직 (선택 사항)
    # 예: 조회수가 100단위로 오를 때만 실제 메인 DB(RDB)에 쿼리를 실행하여 동기화
    if current_views % 100 == 0:
        print(f"📢 [Backup] Article {article_id} reached {current_views} views. Syncing to DB...")

    return {"article_id": article_id, "total_views": current_views}


@app.get("/articles/{article_id}/stats")
async def get_article_stats(article_id: str, request: Request):
    rd = request.app.state.redis
    view_key = f"article:{article_id}:views"
    like_key = f"article:{article_id}:likes"

    # 여러 카운터 값을 가져오기
    # 데이터가 없을 경우(None)를 대비해 int 변환 및 기본값 0 처리
    views = await rd.get(view_key)
    likes = await rd.get(like_key)

    return {
        "article_id": article_id,
        "views": int(views) if views else 0,
        "likes": int(likes) if likes else 0,
    }

4. 카운터를 모아 DB에 반영하는 Write-Back

실무에서는 Redis의 카운터 값을 실시간으로 매번 RDB에 저장하지 않습니다. 너무 비효율적이기 때문입니다. 하지만 Redis는 메모리 기반이기 때문에 장애 발생 시 일부 데이터가 유실될 수 있습니다. 따라서 주기적으로 RDB에 동기화하는 전략이 필요합니다.

  1. Redis 우선 처리: 유저에게는 Redis가 계산한 최신 카운터 값을 즉시 보여줍니다. (신속한 응답)
  2. 배치 저장 (Write-Back): 5분마다 혹은 일정 클릭 수(예: 100회)마다 Redis의 최종 결과값을 RDB에 한 번에 몰아서 업데이트합니다.
  3. 압도적 효율성: 이 방식을 사용하면 DB 입장에서는 UPDATE 횟수를 획기적으로 줄일 수 있어 부하가 크게 감소합니다.

5. [응용] Set을 얹어 같은 사람의 중복 집계 막기

단순히 INCR만 쓰면 한 유저가 새로고침(F5)을 100번 할 때마다 숫자가 100 올라갑니다. 이를 막으려면 이전 레슨에서 배운 Sets를 함께 조합해야 합니다.

python
# 1. Sets에 유저 ID 추가 시도
# 반환값이 1이면 새로운 유저(성공), 0이면 이미 존재하는 유저(실패)
is_new_viewer = await rd.sadd(f"article:{article_id}:viewers", user_id)

# 2. 새로운 유저일 때만 조회수 1 증가
if is_new_viewer == 1:
    await rd.incr(f"article:{article_id}:views")
  • SADD는 이미 존재하는 값이면 추가하지 않고 0을 반환합니다. 따라서 이 반환값을 이용하면 "중복 조회 여부"를 쉽게 판단할 수 있습니다.
  • 위와 같이 구성하면, 아무리 요청이 쏟아져도 완벽하게 "1인 1조회수(혹은 1좋아요)" 정합성을 지킬 수 있습니다.

🚩 정리 — 카운터 설계에서 남는 문제

  • INCR은 동시성(Race Condition) 문제 없이 원자적으로 숫자를 증가시킬 수 있습니다.
  • 잦은 업데이트를 유발하는 카운팅 작업은 메인 DB의 큰 적입니다. 반드시 Redis를 두어 트래픽을 완화하세요.
  • 데이터를 즉시 DB에 저장할지, 모아서 저장할지(Write-Back) 결정하는 것이 아키텍처 설계의 핵심입니다.

5-7. 패턴 05: TTL로 저절로 사라지는 인증번호 만들기

이 글에서 없앨 배치 작업

  • Redis의 만료 시간(TTL) 기능을 활용하여 임시 데이터를 관리하는 법을 배웁니다.
  • 인증번호 검증 로직에서 Redis가 RDB보다 유리한 이유를 이해합니다.
  • FastAPI 환경에서 실제 인증번호 발송 및 검증 프로세스를 구현합니다.

1. 인증번호를 RDB에 넣지 않는 이유

휴대폰 본인인증이나 이메일 인증번호는 보통 3~5분이라는 짧은 유효 기간이 있습니다. 이를 RDB(MySQL 등)에 저장하면 다음과 같은 번거로움이 생깁니다.

  • 불필요한 디스크 I/O — 잠깐 쓰고 버릴 데이터 때문에 매번 디스크를 소모하는 것은 아키텍처 관점에서 낭비입니다.
  • 쓰레기 데이터의 누적 — 만료된 인증번호를 지워주는 별도의 배치(Batch) 작업이나 스케줄러 로직을 직접 짜서 계속 돌려야 합니다.
  • 자동 삭제 기능(TTL) — Redis는 SET 명령어에 EX(초) 옵션을 주면 저장과 동시에 "정확히 n초 뒤 삭제"를 엔진 레벨에서 100% 보장합니다.

2. 발급부터 검증까지의 흐름

text
[ 유저 ]                    [ 서버 (App) ]                [ Redis ]
   |                           |                            |
   |--- (1) 인증번호 요청 ----->|                            |
   |                           |--- (2) 번호 생성 (123456) ->|
   |                           |                            |
   |                           |--- (3) SET key val EX 300 ->| (5분 뒤 자동 삭제 보장)
   |                           |                            |
   |<-- (4) SMS/이메일 발송 ----|                            |
   |                           |                            |
   |--- (5) 인증번호 입력 ----->|                            |
   |      (123456)             |--- (6) GET key ----------->|
   |                           |                            |
   |                           |<-- (7) 값 비교 및 결과 반환 -|
   |<-- (8) 인증 완료/실패 -----|                            |

3. [실습] 발급·검증 API와 5분 만료 걸기

인증번호는 단순한 문자열이므로 String 타입을 사용합니다.

python
import random
import hashlib
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    # Redis에 비밀번호가 설정된 경우
    # app.state.redis = redis.from_url("redis://default:<비밀번호>@localhost:6379/0", decode_responses=True)

    yield

    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)

# 인증번호 유효 시간 (300초 = 5분)
AUTH_TIMEOUT = 300


class SendCodeRequest(BaseModel):
    phone: str


class VerifyCodeRequest(BaseModel):
    phone: str
    input_code: str


@app.post("/auth/send")
async def send_verification_code(req_data: SendCodeRequest, request: Request):
    rd = request.app.state.redis

    # 1. 6자리 난수 생성
    code = str(random.randint(100000, 999999))

    # [보안] 전화번호 등 개인정보(PII)는 그대로 키로 쓰지 않고 해싱하여 저장합니다.
    hashed_phone = hashlib.sha256(req_data.phone.encode()).hexdigest()
    cache_key = f"auth:code:{hashed_phone}"

    # 2. Redis에 저장 (Key: 해싱된 전화번호, Value: 인증번호, TTL: 300초)
    await rd.set(cache_key, code, ex=AUTH_TIMEOUT)

    # 3. 실제로는 여기에 SMS 발송 처리를 합니다.
    print(f"📧 [SMS 발송] To: {req_data.phone}, Code: {code}")

    return {"message": "Verification code sent", "code": code, "expires_in": AUTH_TIMEOUT}


@app.post("/auth/verify")
async def verify_code(req_data: VerifyCodeRequest, request: Request):
    rd = request.app.state.redis

    # 동일하게 전화번호를 해싱하여 키를 생성합니다.
    hashed_phone = hashlib.sha256(req_data.phone.encode()).hexdigest()
    cache_key = f"auth:code:{hashed_phone}"

    # 1. Redis에서 해당 번호의 값 조회
    saved_code = await rd.get(cache_key)

    # Redis 6.2 이상일 경우 GETDEL 사용 권장
    # 단, 사용자 실수로 오타를 낸 경우에도 다시 인증번호를 발급 받아야 함
    # 아래 🚨[2026.03.20 Update] 코멘트 확인
    # saved_code = await rd.getdel(cache_key)

    # 2. 값이 없으면 만료되었거나 생성된 적이 없는 것
    if not saved_code:
        raise HTTPException(status_code=400, detail="Code expired or not requested")

    # 3. 입력값 비교
    if saved_code != req_data.input_code:
        raise HTTPException(status_code=400, detail="Invalid code")

    # 4. [핵심] 인증 성공 시 보안을 위해 즉시 삭제
    await rd.delete(cache_key)

    return {"message": "Authentication successful"}

🚨 [2026.03.20 Update] GETDEL 명령어

GETDEL 명령어는 GET과 DEL을 동시에 실행해서 조회와 삭제 사이에 다른 요청이 개입할 수 없도록 하여 원자성을 보장합니다. 다만 현재 샘플 구조에서 GETDEL을 사용하면 값을 읽어옴과 동시에 키가 바로 삭제되므로, 유저가 실수로 오타를 냈을 때 재시도할 기회조차 없이 바로 폐기되어 새 인증번호를 요청해야만 합니다.

GETDEL은 아주 엄격한 보안이 요구되어 "단 1회의 비교 시도만 허용(일치 여부와 무관하게 즉시 파기)"하는 완전한 일회성 토큰의 원자성을 보장하고 싶을 때 사용하면 좋은 명령어입니다. 하지만 그렇지 않은 일반적인 서비스에서는 사용자 편의성(UX)을 저하시킬 수 있습니다.

현재 샘플 구조에서 원자성과 사용자 편의성(UX)을 모두 만족시키려면 Lua Script를 이용해서 '인증번호 조회 후 일치할 경우 키 삭제'하는 로직을 구현하는 것이 정석입니다.

4. 무차별 대입과 재발급 남용 막기

🛡 1) 인증 성공 후 즉시 삭제 (Delete on Success)

위 코드의 await rd.delete(cache_key)가 매우 중요합니다. 한 번 성공한 인증번호가 유효 시간(5분) 내에 해커나 악의적인 스크립트에 의해 재사용(Replay Attack)되는 것을 원천 차단해야 합니다.

🛡 2) 도배 방지 (Rate Limiting)

인증번호 문자(SMS) 발송은 건당 비용이 발생합니다. 누군가 악의적으로 1초에 100번씩 요청하면 금전적인 손실과 보안 위험이 큽니다.

python
# 1분 이내에 같은 번호로 요청이 오면 거절하는 방어 로직 예시
hashed_phone = hashlib.sha256(req_data.phone.encode()).hexdigest()
is_allowed = await rd.set(f"auth:limit:{hashed_phone}", "1", ex=60, nx=True)

if not is_allowed:
    raise HTTPException(status_code=429, detail="Please wait 1 minute before requesting again.")

위와 같이 nx=True(없을 때만 저장) 옵션을 활용해 "1분 이내 재요청 금지" 로직을 send API 최상단에 간단히 추가할 수 있습니다.

🚩 정리 — 인증번호 저장에서 챙길 것

  • TTL은 개발자가 백그라운드 삭제 배치를 신경 쓰지 않아도 되게 해주는 강력한 '자동 청소부'입니다.
  • Redis의 String 타입은 인증번호, 일회성 토큰, 제한 플래그 같은 짧고 단순한 임시 데이터를 다루기에 최적입니다.
  • 성공한 인증번호는 반드시 즉시 삭제하고, API 호출 빈도를 제한(Rate Limit)하여 보안 취약점을 방어하는 것이 실무 백엔드 엔지니어의 필수 역량입니다.

5-8. 패턴 06: SET NX로 재고 차감을 한 요청만 통과시키기

이 글에서 막을 동시성 사고

  • 분산 환경에서 발생하는 동시성 문제(Concurrency Issue) 의 심각성을 이해합니다.
  • Redis를 활용해 여러 서버 사이의 교통정리를 해주는 분산 락(Distributed Lock) 의 원리를 배웁니다.
  • SET NX PX 명령어를 조합하여 안전하게 자원을 선점하고 해제하는 FastAPI 로직을 익힙니다.

1. 한정판 재고가 마이너스로 떨어지는 순간

서버가 1대일 때(단일 프로세스)는 파이썬 자체의 Lock(threading.Lock, asyncio.Lock)으로 해결 가능합니다. 하지만 서버가 여러 대이거나 멀티 프로세스 환경이라면 어떨까요?

  • 상황: 나이키 한정판 신발 재고가 딱 1개 남은 상황.
  • 사건: 서버 A와 서버 B에 0.001초 차이로 주문 요청이 들어옴.
  • 결과: 두 서버 모두 DB에서 "재고 1개 있음"을 확인하고 각각 1개씩 판매 처리함. → 재고는 -1이 되고 데이터 정합성이 완전히 깨짐.

이때 필요한 것이 "서버가 달라도 공통으로 바라보는 단 하나의 공유 자물쇠"인 분산 락(Distributed Lock) 입니다.

2. 락을 거는 원리 — SET NX PX와 소유권 토큰

과거에는 SETNX와 EXPIRE를 두 단계로 나누어 썼지만, 중간에 서버가 죽으면 영원히 Lock이 풀리지 않는 문제가 있었습니다. 이 원자성(Atomicity)을 보장하기 위해 지금은 SET 명령어의 옵션을 하나로 합쳐서 사용합니다.

redis
# 'lock:item:100'이라는 키가 없을 때만(NX) 저장하고, 5000밀리초(5초) 뒤에 자동 삭제(PX) 해라!
SET lock:item:100 "unique_token_abc" NX PX 5000
  • NX (Not eXists)
    • 아직 아무도 Lock을 획득하지 않았다면 내가 Lock을 획득합니다.
    • 반면 누군가 이미 Lock을 획득했다면 실패합니다. (교통 정리의 핵심)
  • PX 5000 (Milliseconds)
    • Lock을 잡은 서버가 예기치 않게 다운되더라도 5초 뒤에 자동으로 Lock이 풀리도록 해서 다른 서버가 다시 Lock을 잡을 수 있게 합니다. (데드락[Deadlock] 방지용 안전장치)
  • Value ("unique_token_abc")
    • Lock을 해제할 때 "남이 잡은 Lock을 실수로 풀지 않도록" 내가 생성한 고유값으로 Lock을 생성하고 해제합니다.

3. [실습] 락을 잡고 재고를 깎는 API 구현

실무에서는 Lock을 얻지 못했을 때 포기하지 않고 잠시 기다렸다가 다시 시도하는 재시도(Retry) 로직이 포함되어야 합니다. 배운 비동기 패턴을 적용하여 작성해 봅시다.

python
import asyncio
import time
import uuid
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException, Request
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 비동기 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    yield
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


# --- Lock 획득 로직 ---
async def acquire_lock(rd, lock_name: str, acquire_timeout: float = 10.0, lock_timeout_ms: int = 5000):
    """
    acquire_timeout: Lock을 얻기 위해 무한정 기다리지 않고 포기할 최대 대기 시간 (초)
    lock_timeout_ms: Lock 자체의 유효 시간 (밀리초)
    """
    identifier = str(uuid.uuid4())  # 내가 잡은 락임을 증명하는 고유 토큰
    end_time = time.time() + acquire_timeout

    while time.time() < end_time:
        # NX: 키가 없을 때만 생성, PX: 밀리초 단위 TTL 설정
        if await rd.set(lock_name, identifier, nx=True, px=lock_timeout_ms):
            return identifier

        # Lock 획득 실패 시 다른 서버가 풀 때까지 0.1초 대기 후 재시도 (Spin Lock)
        await asyncio.sleep(0.1)

    return False


# --- Lock 해제 로직 ---
async def release_lock(rd, lock_name: str, identifier: str):
    # [주의] 실제 환경에서는 GET과 DEL 사이의 원자성을 위해 Lua 스크립트를 사용해야 완벽합니다.
    # 본 실습에서는 이해를 돕기 위해 파이썬 레벨에서 검증 후 삭제합니다.
    if await rd.get(lock_name) == identifier:
        await rd.delete(lock_name)
        return True
    return False


@app.post("/stock/reduce/{item_id}")
async def reduce_stock(item_id: str, request: Request, user_id: str = "unknown"):
    rd = request.app.state.redis
    lock_name = f"lock:item:{item_id}"

    # 1. 락 획득 시도
    lock_id = await acquire_lock(rd, lock_name)
    if not lock_id:
        raise HTTPException(
            status_code=409,
            detail=f"현재 접속자가 많아 처리가 지연되고 있습니다. 다시 시도해주세요. (요청자: {user_id})",
        )

    try:
        # 2. 임계 영역 (Critical Section): 실제 DB 재고 차감 로직 수행
        # 서버 콘솔에서 어떤 유저의 요청이 처리 중인지 식별 가능하도록 로그 출력
        print(f"📦 [Success] Item {item_id} 재고 차감 작업 중... (요청자: {user_id})")

        await asyncio.sleep(2)  # 무거운 비즈니스 로직 시뮬레이션

        return {"message": f"Item {item_id} stock reduced successfully for {user_id}"}

    finally:
        # 3. 예외가 발생하더라도 작업 완료 후 반드시 락을 해제해야 함!
        await release_lock(rd, lock_name, lock_id)

4. [시연 가이드] 락 없이 돌려 문제를 재현해보기

실제 환경처럼 여러 명의 유저가 구매 버튼을 누르는 상황을 시뮬레이션하는 테스트 스크립트입니다.

⚠ 실행 전 터미널에서 pip install httpx 설치가 필요합니다.

python
# test_lock.py로 저장하여 실행
import asyncio
import httpx
import time


async def send_request(client, item_id, req_num):
    print(f"🚀 요청 {req_num} 출발!")
    try:
        # FastAPI 서버로 POST 요청 전송 (서버에서 식별할 수 있도록 user_id 전달)
        response = await client.post(
            f"http://localhost:8000/stock/reduce/{item_id}?user_id=User_{req_num}",
            timeout=15.0,
        )
        print(f"✅ 요청 {req_num} 결과: {response.json()}")
    except Exception as e:
        print(f"❌ 요청 {req_num} 에러: {e}")


async def main():
    start_time = time.time()
    async with httpx.AsyncClient() as client:
        # 10명의 유저가 정확히 동시에 'item_100'을 구매한다고 가정
        tasks = [send_request(client, "100", i) for i in range(1, 11)]
        await asyncio.gather(*tasks)

    print(f"⏰ 총 소요 시간: {time.time() - start_time:.2f}초")


if __name__ == "__main__":
    asyncio.run(main())

[시연 핵심 포인트]

테스트 스크립트를 실행하면 10개의 요청이 동시에 출발합니다. 하지만 FastAPI 서버의 로그를 확인해 보면, User_1부터 User_10까지의 요청이 한 번에 하나씩만 2초 간격으로 순서대로 진입하여 처리되는 것을 명확히 볼 수 있습니다. 분산 락이 10개의 동시 요청을 완벽하게 '줄 세우기' 한 것입니다!

5. Redlock 논쟁과 라이브러리를 쓸 때의 판단

직접 SET NX PX와 while 루프를 돌며 구현하는 것도 훌륭한 공부가 되지만, 실무 환경에서는 여러 엣지 케이스(Edge case)를 방어하기 위해 검증된 라이브러리를 쓰는 것을 권장합니다.

  • Python: redlock-py 또는 aioredlock
  • Java (Spring): Redisson (실무 분산 락의 사실상 표준)

Redlock 알고리즘

  • 단일 레디스가 아닌 여러 대(보통 3~5대)의 레디스 노드에 락을 걸어, 레디스 서버 한 대가 물리적으로 죽더라도 락의 안정성을 보장하는 분산 합의 알고리즘입니다.
  • 다만 Redlock은 모든 상황의 만능 해법이라기보다, 시스템의 일관성 요구 수준과 장애 허용 범위를 고려해 선택해야 하는 전략입니다.

🛠 지식 박스: 대기 전략 — Polling(Spin Lock + Sleep) vs Pub/Sub

위 실습 코드에서는 while 루프 안에서 0.1초마다 계속 락 획득을 시도하는 스핀 락(Spin Lock / Polling) 방식을 사용했습니다. 구현은 쉽지만 대기하는 서버가 많아질수록 Redis에 불필요한 재시도 트래픽이 증가합니다.

실무에서 많이 쓰이는 Java의 Redisson 같은 라이브러리는 Polling 대신 Pub/Sub 기반으로 락 대기(wait)를 처리합니다.

원리: Lock을 얻지 못한 서버는 무작정 재시도하지 않고, "락이 풀림"을 알리는 특정 채널을 구독(Subscribe) 하고 얌전히 대기합니다. 락을 쥐고 있던 서버가 작업을 마치고 PUBLISH로 해제 알림을 보내면, 그때 대기하던 서버들이 일제히 깨어나 락 획득을 재시도합니다. 레디스의 부하를 획기적으로 줄여주는 스마트한 아키텍처입니다!

🚩 정리 — 락으로 해결되지 않는 것

  • 분산 락은 데이터 정합성이 생명인 금융 결제, 이커머스 예약, 선착순 쿠폰 발급 시스템의 필수 뼈대입니다.
  • TTL(PX, EX) 설정은 선택이 아닌 필수입니다. 설정하지 않으면 락을 쥔 서버가 죽었을 때 시스템 전체가 영원히 멈추는 데드락(Deadlock)에 빠집니다.
  • 락을 해제할 때는 반드시 identifier를 대조하여 내가 잡은 락이 맞는지 확인해야 합니다. 남이 기껏 잡은 락을 실수로 지워버리면 대참사가 발생합니다.

5-9. 패턴 07: 미들웨어 한 겹으로 API 요청 수 제한하기

이 글에서 세울 방어선

  • 서비스의 안정성을 지키는 처리량 제한(Rate Limiting) 의 개념과 필요성을 이해합니다.
  • Redis의 INCR와 EXPIRE를 활용한 고정 윈도우(Fixed Window) 알고리즘을 익힙니다.
  • FastAPI 환경에서 미들웨어(Middleware) 형태의 실전 API 제한 로직을 구현합니다.

1. 한 클라이언트가 서버를 마비시키는 시나리오

아무리 성능이 좋은 서버라도 초당 수만 건의 요청이 한꺼번에 몰리면 시스템이 마비될 수 있습니다.

  • 악의적인 공격 (DoS): 특정 IP에서 무차별적으로 API를 호출하여 서버 리소스를 고갈시키려는 시도.
  • 유료 API 정책: 사용자 등급에 따라 "분당 10회", "시간당 100회" 등으로 사용량을 엄격하게 제한해야 하는 경우.
  • 자원 보호: 크롤러나 봇(Bot)이 데이터를 너무 빠르게 긁어가서 일반 유저의 응답 속도가 느려지는 것을 방지.

이런 상황에서 Redis는 "누가, 언제, 몇 번 요청했는지?" 를 메모리에서 빠르게 계산하여 요청의 통과/차단을 결정하는 강력한 '문지기' 역할을 수행할 수 있습니다.

2. 작동 원리: 1분 단위로 창을 끊는 Fixed Window

가장 단순하고 명확한 방식으로, "특정 시간대(Window) 바구니(버킷) 안에 들어있는 카운트가 기준을 넘었는가?" 를 체크합니다.

💡 데이터 흐름 시각화 (예: 1분당 5회 제한)

text
[ 유저 요청 ] ----> [ Redis 확인 ]
                      |
                      |-- (1) Key: "rate_limit:127.0.0.1:28590538"
                      |-- (2) INCR 실행 (키가 없으면 0에서 1로 자동 생성!)
                      |-- (3) 만약 결과가 1이면? -> EXPIRE 60초 설정
                      |-- (4) 만약 결과가 5 초과면? -> 429 Too Many Requests 반환 (차단)
                      |-- (5) 5 이하면? -> 통과! API 로직 정상 실행

Fixed Window 방식은 구현이 매우 단순하고 Redis 연산이 적기 때문에 대부분의 서비스에서 가장 먼저 사용하는 기본 Rate Limiting 방식입니다.

3. [실습] 모든 요청을 거쳐 가는 미들웨어 만들기

실무에서는 개별 API 함수마다 제한 코드를 넣지 않고, 미들웨어(Middleware)를 통해 모든 요청을 공통으로 검사하고 처리합니다. FastAPI의 Middleware는 모든 요청이 API 라우터에 도달하기 전에 반드시 거치는 공통 처리 계층입니다.

python
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 비동기 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    yield
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)

# 제한 설정: 1분당 최대 5회 (시연을 위해 짧게 설정)
LIMIT = 5
WINDOW = 60


@app.middleware("http")
async def rate_limit_middleware(request: Request, call_next):
    # [Tip] Swagger UI(/docs) 접속은 제한에서 제외해야 테스트가 편합니다.
    if request.url.path in ["/docs", "/openapi.json"]:
        return await call_next(request)

    rd = request.app.state.redis

    # Rate limiting에서는 key 설계가 중요합니다. (Rate Limiting 기준 식별자)
    # 서비스 정책에 따라 IP, user_id, API key 등을 사용할 수 있습니다.
    # ex) rate_limit:{ip}, rate_limit:{user_id}, rate_limit:{api_key}
    user_identifier = request.client.host if request.client else "127.0.0.1"

    # [핵심 로직] 현재 시간을 WINDOW(60초) 단위의 정수로 변환
    # 이 연산을 통해 0~59초 사이의 요청은 모두 동일한 식별자(current_minute)를 가집니다.
    current_minute = int(time.time() // WINDOW)
    cache_key = f"rate_limit:{user_identifier}:{current_minute}"

    # 1. 카운트 증가 (SET 없이 바로 INCR!)
    count = await rd.incr(cache_key)

    # 2. 처음 생성된 키(값이 1)라면 만료 시간 설정
    if count == 1:
        await rd.expire(cache_key, WINDOW)

    # 3. 제한 초과 여부 확인
    if count > LIMIT:
        # [주의] FastAPI 미들웨어 안에서는 HTTPException을 raise하면 500 에러가 발생합니다.
        # 따라서 반드시 JSONResponse를 직접 반환하여 429 상태 코드를 내려주어야 합니다.
        return JSONResponse(
            status_code=429,
            content={
                "error": "Too Many Requests",
                "detail": f"1분에 {LIMIT}회까지만 요청 가능합니다.",
                "retry_after": f"{WINDOW - (int(time.time()) % WINDOW)}s",
            },
        )

    # 4. 검사 통과 시 실제 API 라우터로 요청 전달
    response = await call_next(request)
    return response


@app.get("/data")
async def get_sensitive_data():
    return {"data": "This is protected by rate limiting"}

INCR + EXPIRE 조합은 Redis를 이용한 Rate Limiting에서 가장 널리 알려진 기본 패턴입니다.

🛠 핵심 원리

1. current_minute 계산식의 의미

  • time.time() : 현재 시간을 초 단위 유닉스 타임스탬프로 가져옵니다. (예: 1715432245.123)
  • // WINDOW : 이를 60으로 나눈 몫만 취합니다.
    • 10시 05분 01초: 1715432301 // 60 = 28590538
    • 10시 05분 59초: 1715432359 // 60 = 28590538
    • 즉, 같은 60초 구간이면 같은 값.
  • 분산 환경의 이점: 시계상의 정각(00초~59초)을 기준으로 하는 절대적인 시간 바구니(Time Bucket)가 형성됩니다. 서버가 여러 대라도 시간만 동기화되어 있다면 동일한 cache_key를 바라보게 되어 정확히 합산 제어가 가능합니다.

2. 왜 SET을 사용하지 않나요?

  • Redis의 INCR 명령어는 키가 존재하지 않을 경우, 자동으로 값을 0으로 생성한 뒤 1을 더해 반환합니다.
  • "존재 확인 후 생성"이라는 두 단계를 하나로 합친 원자적(Atomic) 연산이므로, 동시에 수많은 요청이 들어와도 동시성 이슈 없이 아주 안전하고 간결하게 카운팅됩니다.

4. [실습] 429가 뜨는지 확인하는 테스트 클라이언트

동시에 10개의 요청을 FastAPI로 전송하는 테스트 스크립트입니다.

python
# test_limit.py로 저장하여 실행 (pip install httpx 필요)
import asyncio
import httpx


async def send_request(client, req_num):
    # FastAPI 서버로 GET 요청 전송
    response = await client.get("http://localhost:8000/data")

    if response.status_code == 200:
        print(f"✅ 요청 {req_num:02d}: 성공! 통과됨")
    elif response.status_code == 429:
        print(f"🚫 요청 {req_num:02d}: 차단됨! (429 Too Many Requests) - {response.json()['retry_after']} 대기 필요")
    else:
        print(f"❌ 요청 {req_num:02d}: 서버 에러 ({response.status_code})")


async def main():
    async with httpx.AsyncClient() as client:
        print("🚀 10개의 API 요청을 동시에 발사합니다...\n")
        # 10번을 0.01초 간격 없이 동시에 전송
        tasks = [send_request(client, i) for i in range(1, 11)]
        await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(main())

[시연 결과 확인]

출력 결과를 보면 정확히 5개의 요청만 200 성공이 뜨고, 나머지 5개는 Redis의 문지기 역할에 의해 429 차단되는 것을 확인할 수 있습니다.

5. Sliding Window와 Token Bucket으로 넘어가기

위에서 배운 '고정 윈도우'는 구현이 쉽지만 단점이 있습니다.

⚠ 고정 윈도우의 단점: 경계선 폭주 (Boundary Burst)

시간 경계선(예: 59초와 01초 사이)에 트래픽이 몰리면, 순간적으로 제한량의 2배를 처리하게 되는 문제가 발생합니다.

  • 상황: 10시 05분 59초에 5번 요청 (통과) ➔ 10시 06분 01초에 5번 요청 (통과)
  • 결과: 서버 입장에서는 단 2초 사이에 10번의 요청이 쏟아져 들어와 순간적인 과부하에 빠질 수 있습니다.

✅ 이를 해결하는 대안 알고리즘

① Sliding Window Log (슬라이딩 윈도우 로그)

  • 개념: 고정된 시간 바구니(예: 00초~59초)를 쓰지 않고, 요청이 들어온 바로 그 순간부터 정확히 '과거 60초'를 하나의 창문(Window)으로 봅니다. 시간이 흐름에 따라 이 창문이 미끄러지듯(Sliding) 계속 함께 이동합니다.
  • Redis 구현 (Sorted Sets 활용):
    1. 모든 요청의 타임스탬프를 Sorted Sets에 기록합니다. (Score: 타임스탬프, Member: 타임스탬프+고유ID 형태)
    2. 요청이 오면 ZREMRANGEBYSCORE 명령어로 '현재 시간 - 60초'보다 오래된 과거의 요청 타임스탬프를 먼저 다 지웁니다.
    3. ZCARD 명령어로 현재 창문 안에 남은 데이터 개수를 셉니다.
    4. 이 개수가 제한(Limit)을 넘지 않았다면, ZADD로 현재 요청 시간을 기록하고 통과시킵니다.
  • 장점: 경계선 폭주 문제가 원천 차단되며, 언제나 100% 완벽하고 정확하게 요청 횟수를 제한할 수 있습니다.
  • 단점: 숫자 1만 더해주는 INCR와 다르게, 사용자가 요청한 모든 개별 타임스탬프 기록을 Sorted Sets에 저장해야 하므로 트래픽이 많아질수록 서버 메모리 소모량이 기하급수적으로 커집니다.

② Token Bucket (토큰 버킷)

  • 개념: 일정한 속도(Rate)로 토큰이 채워지는 가상의 바구니(버킷)가 있습니다. API 요청이 들어올 때마다 바구니에서 토큰을 하나씩 꺼내 사용하며, 바구니가 비어있으면 요청을 차단합니다.
  • Redis 구현 (상태값 저장 + 시간 계산 활용):
    1. 요청이 오면 '마지막으로 토큰을 채운 시간'과 '현재 시간'의 차이를 계산하여, 그동안 새로 생성되었어야 할 토큰 개수를 구합니다.
    2. 기존 토큰 개수에 새로 생성된 토큰을 더합니다. (단, 바구니의 최대 용량 제한을 넘을 수는 없습니다.)
    3. 토큰이 1개 이상 있다면 토큰을 1개 차감하고 요청을 통과시키며, 0개라면 즉시 차단합니다.
    4. 이 과정(조회-계산-저장)이 동시에 발생할 때 동시성 문제가 생기지 않도록, 실무에서는 보통 Lua 스크립트를 작성하여 한 번에 원자적으로 실행합니다.
  • 장점: 트래픽이 일시적으로 몰리는 폭주(Burst) 현상을 바구니의 최대 크기만큼 유연하게 허용하면서도, 장기적으로는 일정한 처리 속도를 부드럽게(Smooth) 유지할 수 있습니다. 또한 타임스탬프를 모두 저장하는 슬라이딩 윈도우 로그에 비해 메모리 사용량이 매우 적어, AWS 등 글로벌 클라우드 서비스에서 가장 널리 쓰이는 표준 방식입니다. (AWS API Gateway, Nginx, Envoy 등 많은 API Gateway가 내부적으로 Token Bucket 또는 그 변형 알고리즘을 사용합니다.)
  • 단점: 시간 계산과 상태 업데이트를 결합해야 하므로 고정 윈도우나 슬라이딩 윈도우보다 내부 구현 로직이 다소 복잡합니다.

Token Bucket을 구현한 파이썬 로직

python
import time

# [환경 설정]
CAPACITY = 10      # 버킷 최대 용량 (한번에 허용 가능한 최대 횟수)
REFILL_RATE = 1.0  # 초당 채워지는 토큰 수 (1초에 1개씩)


def allow_request(user_id: str) -> bool:
    current_time = time.time()  # 현재 시간 (예: 1715432003.0)

    # 1. Redis에서 기존 상태 가져오기 (실제로는 HGETALL 등 사용)
    # 예: {"tokens": 5, "last_refill_time": 1715432000.0} (3초 전 상태)
    last_tokens = 5
    last_refill_time = 1715432000.0

    # 2. 시간 차이 계산
    time_passed = max(0, current_time - last_refill_time)  # 3초

    # 3. 그동안 생성되었어야 할 토큰 수 계산
    new_tokens = time_passed * REFILL_RATE  # 3.0 * 1.0 = 3개

    # 4. 토큰 더하기 (단, 최대 용량 CAPACITY를 넘을 수 없음)
    # min(10, 5 + 3) = 8개
    current_tokens = min(CAPACITY, last_tokens + new_tokens)

    # 5. 요청 처리 (토큰이 1개 이상이면 통과)
    if current_tokens >= 1.0:
        current_tokens -= 1.0
        # -> 이후 Redis에 갱신된 값(tokens: 7.0, last_refill_time: current_time)을
        #    다시 저장 (HSET)
        return True
    else:
        # 토큰이 부족하면 차단
        return False
  • 토큰 버킷을 구현할 때는 주로 Redis의 Hash 자료구조를 사용해서 현재 남은 토큰 수와 마지막으로 토큰을 계산한 시간을 함께 저장합니다.
  • 실무에서는 가져오기(GET) → 계산하기 → 저장하기(SET)를 따로따로 실행하면, 동시성 문제가 발생할 수 있으므로 Lua 스크립트로 원자적으로 처리하는 것을 권장합니다.

🚩 정리 — 어떤 알고리즘을 고를 것인가

  • Rate Limiting은 악의적인 도배 공격으로부터 서버 자원을 보호하고 공정한 서비스 이용을 보장하는 백엔드의 안전장치입니다.
  • Redis의 INCR과 TTL 조합은 가장 적은 비용(메모리)으로 매우 효율적인 처리량 제한기를 만들 수 있게 해줍니다.
  • 차단(429 에러) 발생 시, Retry-After 메시지를 클라이언트에 함께 전달하여 "언제 다시 시도해야 할지" 알려주는 것이 친절하고 좋은 API 설계입니다.

보충: INCR을 정확히 읽고 Key를 설계하는 법

섹션 5. [Practice] FastAPI와 Redis 실전 패턴 · 유닛 33 / 48 · 보충 학습 자료

이 보충을 붙인 이유

앞 본문의 "SET 없이 바로 INCR" 이라는 표현은 이 단원에서 가장 자주 오해되는 문장입니다. 이 자료는 그 문장을 정확히 읽는 것에서 시작해, INCR이 실제로 어떤 타입을 다루며 Rate Limiting의 key를 어떻게 설계해야 하는지까지를 이어서 설명합니다.

  • "SET 없이"가 "key 없이"가 아님을 구분합니다. key는 언제나 필수 인자입니다.
  • SET 대신 INCR을 쓰는 이유를 원자성 - TTL 보존 - 초기화 불필요 세 가지로 나눠 봅니다.
  • INCR이 다루는 데이터 타입이 String이며, TYPE과 OBJECT ENCODING이 서로 다른 층위임을 이해합니다.
  • key를 누가 - 어디 - 언제 세 축으로 설계하고, IP를 안전하게 식별하는 방법을 익힙니다.
  • INCR과 EXPIRE 사이의 틈이 만드는 영구 차단 사고와 그 해결책을 확인합니다.

1. 본문 한 줄에 숨어 있던 오해 짚기

앞 본문에는 INCR에 관한 문장이 세 군데 나옵니다. 이 세 문장이 머릿속에서 섞이면서 실제로는 존재하지 않는 개념이 만들어집니다.

text
[본문 2쪽]  INCR 실행 (키가 없으면 0에서 1로 자동 생성!)
[본문 4쪽]  # 1. 카운트 증가 (SET 없이 바로 INCR!)
[본문 6쪽]  2. 왜 SET을 사용하지 않나요?

가장 흔한 오해

2쪽의 "키가 없으면"과 4쪽의 "SET 없이"가 합쳐져 "key 없이 INCR을 쓴다" 로 읽히는 경우가 많습니다. 그런 명령은 Redis에 존재하지 않습니다.

Redis는 key-value 저장소이고, key는 어떤 값을 만질지 지정하는 유일한 수단입니다. INCR의 명령 문법에서 key는 선택이 아니라 필수입니다.

redis
INCR key
     ^^^ 선택이 아니라 필수 인자

127.0.0.1:6379> INCR
(error) ERR wrong number of arguments for 'incr' command

본문 실습 코드에도 key는 분명히 존재합니다.

python
cache_key = f"rate_limit:{user_identifier}:{current_minute}"
count = await rd.incr(cache_key)
#                    ^^^^^^^^^ 이것이 key 다
읽은 방식 성립 실제 의미
"key 없이 INCR을 쓴다" 불가 key 인자가 빠지면 문법 오류로 거절된다
"SET 없이 바로 INCR을 쓴다" 성립 SET으로 key를 미리 만들어 둘 필요가 없다는 뜻
"키가 없으면 자동 생성된다" 성립 key가 아직 존재하지 않아도 INCR이 0에서 시작해 만들어 준다

한 문장 정리

"SET 없이"는 key 인자가 없다는 뜻이 아니라, 초기화 명령이 없다는 뜻입니다.

2. INCR 명세를 정의 그대로 읽기

Redis 공식 문서의 INCR 정의는 다음 한 문장입니다.

INCR 정의

"If the key does not exist, it is set to 0 before performing the operation." (키가 존재하지 않으면 연산을 수행하기 전에 0으로 설정한다.)

redis
127.0.0.1:6379> EXISTS page:views
(integer) 0          <- 키가 없다

127.0.0.1:6379> INCR page:views
(integer) 1          <- 0으로 만들고 +1 하여 1

127.0.0.1:6379> INCR page:views
(integer) 2

다른 저장소였다면 아래 왼쪽처럼 써야 할 코드가 오른쪽 한 줄로 끝납니다. 본문이 말하는 "존재 확인 후 생성이라는 두 단계를 하나로 합친" 이 바로 이것입니다.

text
[초기화가 필요한 방식]              [INCR]
if not exists(key):                 incr(key)
    set(key, 0)
incr(key)

3. SET으로는 안 되는 세 가지 이유

3.1 SET은 세 단계라서 동시 요청에 무너진다

카운터를 SET으로 올리려면 반드시 읽고 - 계산하고 - 쓰는 세 단계를 거칩니다.

text
개념 코드 (권장하지 않음)
current = GET rate_limit:...     # 1) 읽기
new     = current + 1            # 2) 애플리케이션에서 계산
SET rate_limit:... new           # 3) 쓰기

요청 두 개가 동시에 들어오면 둘 다 같은 값을 읽고 같은 값을 씁니다. 한 번의 사용이 사라집니다(lost update).

text
시각     요청 A            요청 B            Redis의 값
------------------------------------------------------------
 t1     GET -> 5                            5
 t2                       GET -> 5          5
 t3     5 + 1 = 6                           5
 t4                       5 + 1 = 6         5
 t5     SET 6                               6
 t6                       SET 6             6   <- A의 요청이 증발

실제로 처리한 요청 수: 7건 / 카운터에 남은 값: 6

Rate Limiting에서의 의미

누락은 항상 제한이 느슨해지는 방향으로 발생합니다. 동시 요청이 많을수록 더 많이 새기 때문에, 정작 방어가 필요한 폭주 상황에서 제한이 가장 크게 무력화됩니다. 본문이 "동시성 이슈 없이"라고 강조한 이유가 이것입니다.

INCR은 이 세 단계가 Redis 서버 안에서 하나의 명령으로 끝납니다. Redis는 명령을 하나씩 처리하므로 두 요청이 끼어들 틈 자체가 없습니다.

3.2 SET은 TTL을 파괴한다 — 가장 결정적인 이유

Fixed Window 방식에서 key의 TTL이 곧 윈도우의 남은 시간입니다. 그런데 SET 계열은 TTL을 그대로 두지 않습니다.

명령 TTL에 생기는 일 결과
SET key 6 TTL이 제거됨 키가 영구화되어 해당 사용자가 영원히 차단된다
SET key 6 EX 60 TTL이 60초로 되감김 계속 요청하는 사용자는 윈도우가 끝나지 않아 예산이 회복되지 않는다
SET key 6 KEEPTTL TTL 유지 TTL 문제는 없지만 3.1의 경쟁 문제는 그대로 남는다
INCR key TTL을 건드리지 않음 "첫 요청에만 TTL을 걸고 이후에는 그대로 둔다"가 그대로 성립한다

핵심

본문의 if count == 1: await rd.expire(...)이 성립하는 이유가 바로 INCR이 TTL을 보존하기 때문입니다. SET이었다면 이 코드는 동작하지 않습니다.

3.3 반환값이 곧 몇 번째 요청인지를 알려준다

INCR은 증가시킨 뒤의 값을 돌려줍니다. 그래서 별도의 GET 왕복이 필요 없고, 반환값이 1이라는 사실 자체가 "이번이 이 윈도우의 첫 요청"이라는 신호가 됩니다. 본문 코드가 count == 1로 최초 생성을 판별하는 근거입니다.

4. 그래서 저장되는 타입은 무엇인가

여기서 한 번 더 놀랍니다

Redis에는 정수 타입도, 카운터 타입도 없습니다. INCR은 String 계열 명령입니다.

redis
127.0.0.1:6379> INCR page:views
(integer) 1

127.0.0.1:6379> TYPE page:views
string               <- 정수가 아니라 string

127.0.0.1:6379> GET page:views
"1"                  <- 따옴표. 문자열 "1"이다

4.1 그럼 어떻게 더하는가

INCR이 하는 일을 풀어 쓰면 다음과 같고, 이 전부가 하나의 원자적 연산입니다.

text
1) key의 값(문자열)을 꺼낸다
2) 64비트 부호 있는 정수(long long)로 파싱한다
3) 파싱에 실패하면 오류를 반환한다
4) 성공하면 +1
5) 결과를 다시 문자열로 저장한다
6) 새 값을 정수로 반환한다
redis
127.0.0.1:6379> SET mykey "hello"
OK
127.0.0.1:6379> INCR mykey
(error) ERR value is not an integer or out of range

4.2 타입과 인코딩은 다른 층위다

"그럼 매번 문자열을 파싱하느라 느린 것 아닌가" 하는 걱정은 필요 없습니다. Redis는 숫자로만 이루어진 String을 내부적으로 진짜 정수로 들고 있습니다.

redis
127.0.0.1:6379> SET a "12345"
127.0.0.1:6379> OBJECT ENCODING a
"int"                <- 문자열이 아니라 long으로 저장돼 있다

127.0.0.1:6379> SET b "hello"
127.0.0.1:6379> OBJECT ENCODING b
"embstr"             <- 44바이트 이하의 짧은 문자열

127.0.0.1:6379> SET c "<44바이트를 넘는 긴 문자열>"
127.0.0.1:6379> OBJECT ENCODING c
"raw"
구분 확인 명령 의미
타입 TYPE key 결과 string 논리적 계약. 어떤 명령을 쓸 수 있는지를 결정한다
인코딩 OBJECT ENCODING key 결과 int 물리적 저장 방식. Redis의 최적화 영역이며 사용자가 지정하지 않는다

인코딩이 int이면 INCR은 파싱 없이 long에 1을 더합니다. 즉 겉보기 타입은 String이지만 내부는 진짜 정수이며, 변환 비용을 걱정할 필요가 없습니다.

4.3 범위와 경계

  • 범위: -9,223,372,036,854,775,808 부터 9,223,372,036,854,775,807 (부호 있는 64비트)
  • 넘으면 값이 되돌아가는 것이 아니라 오류가 납니다. (ERR increment or decrement would overflow)
  • Rate Limiting 카운터가 이 범위에 닿을 일은 없습니다. 60초 윈도우에 int64를 채우려면 초당 10경 번 규모의 요청이 필요합니다.

INCRBYFLOAT은 성격이 다릅니다

타입은 여전히 String이지만 인코딩은 embstr 또는 raw가 되고, 부동소수점 오차가 발생합니다. 카운터에는 사용하지 않습니다.

4.4 타입이 다르면 명령도 다르다

redis
127.0.0.1:6379> LPUSH mylist a
127.0.0.1:6379> INCR mylist
(error) WRONGTYPE Operation against a key holding the wrong kind of value
타입 증가 명령 무엇을 올리는가
String INCR / INCRBY 값 자체 (이 단원의 Rate Limiting)
Hash HINCRBY / HINCRBYFLOAT 특정 field의 값 (5-6 조회수/좋아요에서 재등장)
Sorted Set ZINCRBY member의 score (5-10 실시간 랭킹에서 재등장)

5. Key를 '누가·어디서·언제'로 쪼개기

본문은 "Rate limiting에서는 key 설계가 중요합니다"라고 짚고 넘어갑니다. 왜 중요한지를 여기서 이어서 봅니다.

출발점

Redis에는 SQL의 GROUP BY user_id 같은 것이 없습니다. Redis가 제공하는 그룹핑 수단은 key 문자열뿐입니다. 따라서 "IP별로 세고 싶다"는 요구는 곧 "IP를 key 문자열에 넣는다"로만 실현됩니다.

본문의 key는 두 축으로 이루어져 있습니다.

text
rate_limit:127.0.0.1:28590538
           ^^^^^^^^^ ^^^^^^^^
           누가        언제

- 누가 : IP가 다르면 key가 달라지고, key가 다르면 카운터가 자동으로 분리된다
- 언제 : int(time.time() // 60)이므로 60초마다 새 key가 생긴다
         새 key는 값이 없으므로 카운터가 0에서 다시 시작한다
         지난 윈도우의 key는 EXPIRE가 알아서 지운다 (별도 정리 작업 불필요)

실무에서는 여기에 "어디서" 축을 하나 더 둡니다. 로그인과 단순 조회는 위험도도 비용도 다르므로 예산을 공유하면 안 됩니다.

text
rate:{ip_1381}:sessions:29342422
     ^^^^^^^^^ ^^^^^^^^ ^^^^^^^^
     누가       어디       언제
축 값의 예 없으면 생기는 문제
누가 ip_1381, sess_ab12..., key_9f3c... 한 사람이 쓴 예산에 다른 사람이 막힌다. 기능이 성립하지 않는다
어디 sessions, inventory__sku__reserve 로그인 폭주가 단순 조회 예산까지 소진시킨다
언제 29342422 (윈도우 번호) 카운터가 영원히 누적되어 한 번 막힌 사용자가 풀리지 않는다

"어디" 축은 경로 문자열을 그대로 쓰지 않습니다

request.url.path를 그대로 넣으면 /products/1, /products/2처럼 경로 파라미터 값마다 key가 생겨 key 개수가 폭발합니다. 라우터가 매칭한 경로 템플릿(/products/{product_id})을 사용합니다.

6. 가장 어려운 것은 '누가'를 정하는 일

본문은 request.client.host를 사용하고, "서비스 정책에 따라 IP, user_id, API key 등을 사용할 수 있습니다"라고 안내합니다. 실제 서비스에서 이 선택이 왜 까다로운지 세 가지만 짚습니다.

6.1 프록시 뒤에서는 모두가 한 IP가 된다

text
[프록시가 없을 때]                [nginx/ALB가 앞에 있을 때]
client ---------> FastAPI         client ---> nginx ---> FastAPI
request.client.host               request.client.host
  = 실제 client 주소                 = 항상 nginx의 주소
                                    모든 사용자가 하나로 뭉친다

그래서 원래 주소가 담긴 X-Forwarded-For 헤더를 봅니다. 그런데 이것은 그냥 HTTP 헤더라 누구나 마음대로 쓸 수 있습니다. 검증 없이 신뢰하면 공격자가 요청마다 값을 바꾸는 것만으로 제한이 완전히 무력화됩니다.

python
peer = request.client.host if request.client else "unknown"
forwarded = request.headers.get("x-forwarded-for")

# 내가 직접 연결된 상대가 "내가 등록한 프록시"일 때만 헤더를 믿는다
if forwarded and peer in TRUSTED_PROXIES:
    # XFF는 "client, proxy1, proxy2" 순으로 왼쪽에 쌓이므로 맨 앞이 최초 client
    peer = forwarded.split(",")[0].strip() or peer

6.2 IP를 그대로 key에 넣으면 key 개수가 폭발한다

IP를 그대로 쓰면 서로 다른 주소 개수만큼 key가 생깁니다. IPv6에서는 사실상 무한합니다. 공격자가 주소를 바꿔가며 요청하면 제한은 걸지 못하면서 메모리만 소모하는, Rate Limiter 자체가 공격 벡터가 되는 상황이 됩니다.

해결책은 익명 주소를 고정된 개수의 버킷으로 접는 것입니다.

python
bucket = int(hashlib.sha256(peer.encode()).hexdigest()[:8], 16)
principal = f"ip_{bucket % 4096}"
IP sha256 앞 8자리 (정수) % 4096 principal
203.0.113.7 4274333029 1381 ip_1381
203.0.113.9 3630282729 2025 ip_2025
198.51.100.4 345067605 85 ip_85

받아들인 대가

버킷이 4096개이므로 무관한 IP 두 개가 같은 칸에 떨어지면 예산을 나눠 쓰게 됩니다. 이는 "무관한 사용자가 가끔 손해 본다"가 "key 폭발로 Redis가 죽는다"보다 낫다는 판단입니다. 인증된 사용자는 각자 고유한 principal을 가지므로 이 충돌의 영향을 받지 않습니다.

6.3 사용자가 제시한 식별자를 검증 없이 믿으면 안 된다

rate_limit:{api_key}처럼 사용자가 보낸 값을 그대로 주체로 쓰면, 공격자는 요청마다 새로운 값을 보내는 것만으로 예산을 무한히 초기화할 수 있습니다. 등록된 key만 주체로 인정하고, 나머지는 IP 버킷으로 떨어뜨려야 합니다.

특히 위험한 곳: 스스로 identity를 발급하는 엔드포인트

로그인 엔드포인트는 응답으로 새 세션을 발급합니다. 세션을 주체로 삼으면 다음 요청의 주체가 매번 달라져 예산이 계속 초기화되고, 무한 로그인 시도가 가능해집니다. 이런 엔드포인트에서는 주체가 제시한 어떤 식별자도 무시하고 위조 불가능한 네트워크 주소로만 세어야 합니다.

7. INCR과 EXPIRE 사이에 벌어지는 틈

본문의 코드는 명령이 두 개로 나뉘어 있습니다.

python
count = await rd.incr(cache_key)        # 명령 1
if count == 1:
    await rd.expire(cache_key, WINDOW)  # 명령 2  <- 이 사이에 틈이 있다

사고 시나리오

INCR은 성공했는데 그 직후 프로세스가 죽거나 네트워크가 끊겨 EXPIRE가 실행되지 않으면, TTL이 없는 카운터가 남습니다. 이 key는 스스로 사라지지 않으므로 해당 사용자는 영구히 차단됩니다. 확률은 낮지만 발생하면 수동 개입 없이는 복구되지 않습니다.

"증가"와 "첫 증가라면 TTL 설정"은 사실 하나의 상태 전이입니다. 이를 Lua 스크립트로 묶으면 Redis가 스크립트 전체를 하나의 명령처럼 처리하므로 중간에 끼어들 틈이 사라집니다.

lua
-- KEYS[1] counter, ARGV[1] window(ms), ARGV[2] limit
local count = redis.call('INCR', KEYS[1])
local pttl  = redis.call('PTTL', KEYS[1])

if pttl < 0 then
  -- 첫 증가이거나, 과거 사고로 TTL이 유실된 상태. 둘 다 여기서 복구한다.
  redis.call('PEXPIRE', KEYS[1], ARGV[1])
  pttl = tonumber(ARGV[1])
end

local allowed = 0
if count <= tonumber(ARGV[2]) then allowed = 1 end

return {allowed, count, pttl}

덤으로 얻는 것

반환된 pttl을 그대로 Retry-After에 사용하면 재시도 시각의 기준이 애플리케이션 시계가 아니라 Redis의 TTL이 됩니다. 서버 간 시계 오차와 무관해집니다.

스스로 확인해볼 질문

  • INCR을 key 인자 없이 호출할 수 있는가? (아니오. 문법 오류다)
  • "SET 없이"는 무엇이 없다는 뜻인가? (초기화 명령. key가 아니다)
  • TYPE이 string인데 어떻게 덧셈이 되는가? (인코딩이 int라 내부는 long이다)
  • SET key 6 EX 60으로 카운터를 올리면 무엇이 깨지는가? (TTL이 되감겨 윈도우가 끝나지 않는다)
  • Redis에서 "IP별로 센다"를 실현하는 유일한 수단은 무엇인가? (key 문자열에 IP를 넣는 것)
  • X-Forwarded-For를 언제 믿어도 되는가? (직접 연결된 peer가 등록된 프록시일 때만)
  • INCR 직후 프로세스가 죽으면 무슨 일이 생기는가? (TTL 없는 카운터가 남아 영구 차단된다)
질문 답
"key 없이 INCR"이 가능한가 불가능. 본문의 "SET 없이"는 초기화가 필요 없다는 뜻
키가 없을 때 INCR의 동작 0으로 설정한 뒤 +1 하여 1을 반환
데이터 타입 String. Redis에 정수 타입은 없다
내부 저장 방식 숫자 값은 인코딩이 int. 파싱 비용 없음
SET을 쓰면 안 되는 결정적 이유 동시성 누락과 TTL 파괴
값을 어떻게 구분하는가 오직 key 문자열. 구분할 축을 전부 문자열에 넣는다
IP 식별의 함정 프록시 뒤 뭉침, XFF 위조, key 개수 폭발
INCR과 EXPIRE의 위험 두 명령 사이의 틈. Lua로 하나의 전이로 묶는다

5-10. 패턴 08: Sorted Set으로 즉시 갱신되는 리더보드

이 글에서 만들 랭킹

  • Redis Sorted Sets를 활용하여 실시간으로 순위가 변하는 랭킹 시스템의 원리를 이해합니다.
  • RDB의 ORDER BY 연산의 부담과, Redis가 이를 대규모 데이터에서도 매우 빠르게 처리하는 구조를 이해합니다.
  • 점수 업데이트, 상위권 조회, 내 등수와 내 주변 순위 목록 등 실전 리더보드 API를 구현합니다.

1. 순위를 DB로 계산하면 무너지는 이유

수십만/수백만 명의 유저가 경쟁하는 게임이나 이커머스의 '인기 판매 순위'를 구현할 때, 일반적인 RDB(MySQL 등)는 병목 현상을 겪을 수 있습니다.

  • RDB의 한계: 순위를 매기려면 매번 모든 데이터를 특정 점수 기준으로 정렬(ORDER BY)해야 합니다. 데이터가 많아질수록 CPU 부하가 기하급수적으로 늘어나며, 실시간 응답에 부담이 될 수 있습니다.
  • Redis의 해결책: 데이터가 들어오는 순간 이미 정렬을 마치는 Sorted Sets 자료구조를 사용합니다. Sorted Set은 점수 변경과 순위 조회를 매우 빠르게 처리하며, 대규모 데이터에서도 O(log N) 수준으로 동작해 실시간 리더보드에 적합합니다.

2. 작동 원리: 점수를 넣으면 정렬은 서버가 한다

Sorted Sets는 각 멤버(Member)에게 점수(Score)를 부여하고, 이를 기반으로 항상 정렬된 상태를 유지해 줍니다.

💡 리더보드 데이터 흐름 (실시간 점수 예시)

text
[ 유저 점수 획득 ] ----> [ Redis (Sorted Sets) ]
                          |
                          |-- (1) ZINCRBY game:rank 500 "user_A" (점수 500점 추가)
                          |-- (2) 내부 자동 정렬 (O(log N))
                          |-- (3) ZREVRANGE game:rank 0 9 (상위 10명 즉시 추출)
                          |-- (4) ZREVRANK game:rank "user_A" (user_A의 현재 등수 확인)

💡 ZREVRANGE 대신 ZRANGE ... REV

Redis 6.2 버전부터는 과거의 여러 범위 조회 명령어(ZREVRANGE, ZRANGEBYSCORE 등)를 ZRANGE 하나로 통합하여 사용하는 것을 권장합니다.

  • 과거: ZREVRANGE key 0 9 (내림차순 상위 10명)
  • 현재: ZRANGE key 0 9 REV (동일한 동작, 확장된 문법)
  • 이유: 단일 명령어로 다양한 옵션(정렬 방향, 스코어 기준 조회 등)을 일관성 있게 제어하기 위함입니다. 최근 공식 문서와 실무 코드에서는 이 통합 문법이 주로 사용됩니다.

3. [실습] 점수 등록·상위권 조회·내 순위 API

게임이나 서비스의 '오늘의 인기 유저' 랭킹을 관리하는 API를 만들어 봅시다.

python
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request, HTTPException
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 비동기 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    yield
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)

RANKING_KEY = "leaderboard:daily:2026-03-13"


@app.post("/rank/score")
async def update_score(user_id: str, score_delta: float, request: Request):
    """유저의 점수를 실시간으로 업데이트합니다."""
    rd = request.app.state.redis

    # ZINCRBY: 기존 점수에 합산 (키나 멤버가 없으면 자동 생성)
    new_score = await rd.zincrby(RANKING_KEY, score_delta, user_id)

    return {"user_id": user_id, "current_score": new_score}


@app.get("/rank/top10")
async def get_top_rankers(request: Request):
    """상위 10명의 정보를 최신 문법(REV)으로 가져옵니다."""
    rd = request.app.state.redis

    # redis-py에서는 zrange 메서드의 desc=True 옵션이 REV 문법으로 동작합니다.
    top_list = await rd.zrange(RANKING_KEY, 0, 9, desc=True, withscores=True)

    # 튜플의 리스트를 가공된 JSON 배열로 변환
    result = [{"rank": i + 1, "user_id": m, "score": s} for i, (m, s) in enumerate(top_list)]

    return {"top_rankers": result}


@app.get("/rank/around-me/{user_id}")
async def get_nearby_rank(user_id: str, request: Request):
    """특정 유저를 기준으로 앞뒤 유저를 포함한 '내 주변 5명'을 조회합니다."""
    rd = request.app.state.redis

    # 1. 나의 현재 등수 확인 (0-indexed)
    my_rank = await rd.zrevrank(RANKING_KEY, user_id)

    if my_rank is None:
        raise HTTPException(status_code=404, detail="Ranking data not found")

    # 2. 내 주변 범위 계산 (내 앞 2명 ~ 내 뒤 2명)
    start = max(0, my_rank - 2)
    end = my_rank + 2

    # 3. 해당 범위 유저들 추출
    nearby_list = await rd.zrange(RANKING_KEY, start, end, desc=True, withscores=True)

    result = [
        {"rank": start + i + 1, "user_id": m, "score": s}
        for i, (m, s) in enumerate(nearby_list)
    ]

    return {"user_id": user_id, "nearby_rankers": result}

💡 [Tip] 키(Key)가 없으면 어떻게 되나요?

ZINCRBY는 해당 키가 Redis에 존재하지 않으면 자동으로 새로운 Sorted Set을 생성하고, 점수를 0에서 시작하여 인자로 전달된 값만큼 더해줍니다. 즉, 별도의 존재 확인(EXISTS)이나 초기 생성 과정 없이 바로 호출해도 안전하게 작동합니다.

💡 [Tip] 랭킹 데이터의 생명주기 관리

일간/주간 랭킹은 위 예제와 같이 보통 날짜가 포함된 키(leaderboard:daily:YYYY-MM-DD)를 사용합니다. 데이터가 무한히 쌓여 메모리를 점유하는 것을 방지하기 위해, 키 생성 시 TTL(EXPIRE)을 설정하거나 별도의 배치 작업을 통해 이전 시즌 데이터를 정기적으로 정리하는 것이 좋습니다.

🚩 정리 — 리더보드 운영에서 챙길 것

  • Sorted Sets는 단순 저장을 넘어 '실시간 연산'을 담당하는 Redis의 핵심 무기입니다.
  • O(log N) 의 성능 덕분에 대규모 사용자 환경에서도 상시 정렬된 상태를 유지할 수 있습니다.
  • 리더보드 시스템뿐만 아니라 선착순 대기열, 최신순 피드 관리 등 '순서'가 중요한 모든 곳에 이 패턴을 적용해 보세요.

5-11. 패턴 09: Pub/Sub으로 브라우저에 실시간 공지 밀어내기

이 글에서 연결할 실시간 경로

  • CLI 테스트를 넘어, 실제 FastAPI 서버 환경에서 Redis Pub/Sub을 활용하는 구조를 설계합니다.
  • 비동기 루프를 활용해 구독(Subscribe) 커넥션을 효율적으로 관리하는 실무 패턴을 배웁니다.
  • SSE(Server-Sent Events) 기술과 결합하여 서버의 이벤트를 클라이언트에 실시간으로 푸시하는 로직을 익힙니다.

1. Redis Pub/Sub을 다시 짚고 가기

Pub/Sub은 Publish(발행)와 Subscribe(구독)의 약자로, 메시지를 보내는 쪽과 받는 쪽이 서로 누구인지 몰라도 통신할 수 있게 해주는 패턴입니다.

구조 — Publisher → Channel → Subscriber

  1. 발행자(Publisher)가 특정 채널(Channel)에 메시지를 던집니다.
  2. 채널을 구독하고 있던 모든 구독자(Subscriber)에게 메시지가 동시에 전달됩니다.

2. 웹 서버에 Pub/Sub을 붙이면 까다로운 지점

일반적인 Redis 명령(GET/SET 등)은 요청 → 즉시 응답 구조입니다. 하지만 SUBSCRIBE는 동작 방식이 다릅니다.

  • SUBSCRIBE를 실행하면 Redis 연결이 메시지를 계속 기다리는 상태(Blocking)가 되며, 이 연결은 메시지 수신 전용 모드가 됩니다.
  • 따라서 같은 연결에서는 GET, SET 같은 일반 Redis 명령을 사용할 수 없습니다.

해결 방법

실무에서는 Redis 연결을 두 가지로 분리합니다.

  • 일반 API용 연결 → GET, SET 등 데이터 처리
  • Pub/Sub 전용 연결 → SUBSCRIBE로 메시지 수신

FastAPI에서는 보통 비동기 루프나 별도 Task를 사용해 Pub/Sub 메시지를 지속적으로 수신하도록 구현합니다.

3. [실습] 공지 발행과 SSE 수신 붙이기

모든 애플리케이션 서버들에(모든 접속자에게) 즉시 메시지를 보내는 '전역 시스템 공지' 기능을 만들어 봅시다. 여기서는 클라이언트가 새로고침 없이 메시지를 받는 SSE(Server-Sent Events) 방식을 함께 사용합니다.

① 서버 코드 작성 (pub_sub_server.py)

python
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, FileResponse
import redis.asyncio as redis


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 서버 시작 시 Redis 연결 풀 생성
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    yield
    # 서버 종료 시 연결 해제
    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)

# 공지사항용 채널명
NOTICE_CHANNEL = "system:notices"


@app.get("/pub_sub")
async def index():
    # pub_sub.html 파일을 웹 브라우저에 띄웁니다.
    return FileResponse("pub_sub.html")


@app.post("/publish-notice")
async def send_notice(message: str, request: Request):
    """[발행자] Swagger에서 공지를 발행합니다."""
    rd = request.app.state.redis
    # 채널에 메시지 전송 (구독 중인 수신자 수를 반환)
    subscriber_count = await rd.publish(NOTICE_CHANNEL, message)
    return {"status": "success", "received_subscribers": subscriber_count}


@app.get("/stream-notices")
async def stream_notices(request: Request):
    """[수신자] SSE를 통해 실시간 알림을 수신합니다."""

    async def event_generator():
        # [핵심] rd.pubsub()으로 구독 전용 객체 생성 (커넥션 분리)
        async with request.app.state.redis.pubsub() as pubsub:

            await pubsub.subscribe(NOTICE_CHANNEL)

            try:
                while True:
                    # 메시지 대기 (ignore_subscribe_messages=True로 설정)
                    message = await pubsub.get_message(
                        ignore_subscribe_messages=True, timeout=1.0
                    )

                    if message and message["type"] == "message":
                        data = message["data"]

                        # SSE 표준 포맷 전송 (data: [내용]\n\n)
                        yield f"data: {data}\n\n"

                    # 브라우저가 창을 닫으면 루프를 종료하여 커넥션 반환
                    if await request.is_disconnected():
                        break

                    await asyncio.sleep(0.01)
            finally:
                # 작업 종료 시 명시적 구독 해제
                await pubsub.unsubscribe(NOTICE_CHANNEL)

    return StreamingResponse(event_generator(), media_type="text/event-stream")

(주의) Pub/Sub 연결 객체

현재 예제 구조는 클라이언트 수만큼 Redis Pub/Sub 연결이 생성되므로, 실제 서비스에서는 보통 Redis 구독을 소수의 연결로 유지하고 서버 내부에서 클라이언트들에게 브로드캐스트하는 구조를 사용합니다.

실무에서는 Redis Pub/Sub 구독을 애플리케이션 시작 시 생성하고 서버가 종료될 때 구독을 해제하는 방식으로 운영합니다.

🔍 1) 서버 소스 핵심 포인트 (FastAPI & Redis)

커넥션 독점(Connection Monopoly) 문제

  • 문제: lifespan에서 만든 app.state.redis는 여러 API가 공유하는 연결 풀입니다. 만약 여기서 하나를 뽑아 SUBSCRIBE를 해버리면, 그 커넥션은 메시지 대기 전용이 되어버려 다른 GET/SET 명령어를 처리할 수 없게 됩니다.
  • 해결: async with rd.pubsub() as pubsub
    • 필요할 때마다 구독 전용 커넥션 객체를 할당받아 사용합니다. 이렇게 해서 커넥션 풀(Pool) 내의 다른 커넥션들을 방해하지 않도록 합니다.
    • 그리고 async with를 사용해 함수가 종료되면 자동으로 커넥션이 풀로 반환되도록 구현했습니다.

기타

  • ignore_subscribe_messages=True : Redis는 구독 시작 시 "구독 성공"이라는 메타 메시지를 먼저 보냅니다. 실무 로직에 불필요한 이 메시지를 걸러내기 위한 설정입니다.
  • StreamingResponse : 클라이언트에게 응답을 한 번에 다 보내는 것이 아니라, 데이터가 생길 때마다 지속적으로 흘려보내기(Streaming) 위해 사용합니다.
  • request.is_disconnected() : SSE 연결은 서버가 끊기 전까지 유지되려 합니다. 사용자가 브라우저 탭을 닫았을 때 무한 루프를 멈추고 서버 자원을 회수하는 안전 장치입니다.

💡 2) SSE (Server-Sent Events) 활용

웹소켓(WebSocket)은 양방향 통신이라 설정이 복잡하지만, SSE는 서버에서 클라이언트로만 쏘는 단방향 통신입니다. "공지사항", "주식 시세", "알림 서비스"처럼 서버의 이벤트를 전달할 때 가장 가볍고 효율적인 기술입니다.

💡 3) 실무적 이점: 서비스 간 디커플링 (Decoupling)

이 구조를 사용하면 '결제 서버'에서 결제가 일어났을 때 레디스에 PUBLISH만 하면 됩니다. 그러면 '알림 서버', '로그 분석 서버', '배송 관리 서버'가 각자 레디스를 구독하고 있다가 알아서 자기 할 일을 처리합니다. 서로의 서버 주소를 알 필요가 없는 느슨한 결합이 완성됩니다.

② 브라우저 코드 작성 (pub_sub.html)

html
<!DOCTYPE html>
<html lang="ko">
<head>
    <meta charset="UTF-8">
    <title>Redis Pub/Sub 실시간 알림</title>
</head>
<body style="background: #1a1a1a; color: white; font-family: sans-serif; padding: 50px;">
    <h1>🔔 실시간 시스템 공지</h1>
    <p>현재 접속 포트: <span id="port" style="color: #deff9a; font-weight: bold;"></span></p>

    <div id="log" style="border: 1px solid #deff9a; padding: 20px; height: 300px; overflow-y: auto; background: #000; border-radius: 10px;">
        <div style="color: #888;">알림을 기다리는 중...</div>
    </div>

    <script>
        // 현재 브라우저가 어느 포트 서버에 접속했는지 표시
        document.getElementById("port").innerText = window.location.port || "80";

        // SSE 연결 시작
        const eventSource = new EventSource("/stream-notices");

        eventSource.onmessage = (event) => {
            const log = document.getElementById("log");
            const time = new Date().toLocaleTimeString();

            // 새 알림 추가
            const entry = document.createElement("div");
            entry.style.marginBottom = "10px";
            entry.innerHTML = `> <span style="color: #deff9a;">${event.data}</span> <small style="color:#666; margin-left: 10px;">(${time})</small>`;

            log.appendChild(entry);
            log.scrollTop = log.scrollHeight; // 스크롤 자동 이동
        };

        eventSource.onerror = (err) => {
            console.error("SSE 연결 오류:", err);
        };
    </script>
</body>
</html>

🔍 클라이언트 소스 핵심 포인트 (HTML/JS)

  • EventSource : HTML5 표준 API로, 서버가 일방적으로 데이터를 밀어주는 SSE(Server-Sent Events) 연결을 맺는 도구입니다. WebSocket보다 설정이 훨씬 간편합니다.
  • onmessage : 서버의 yield 문을 통해 전송된 메시지가 도착할 때마다 호출됩니다. event.data에 서버가 보낸 텍스트가 담겨 있습니다.
  • window.location.port : 현재 페이지가 로드된 포트 번호를 읽어옵니다. 다중 서버 시연 시 유저가 어느 서버에 붙어있는지 시각적으로 보여주기 위한 실무적 팁입니다.
  • log.scrollTop = log.scrollHeight : 메시지가 쌓일 때 사용자가 일일이 스크롤하지 않아도 최신 메시지를 볼 수 있게 해주는 UI/UX 디테일입니다.

🧪 실습 가이드: 서버 두 대에 동시에 도달하는지 확인하기

하나의 레디스 채널을 여러 개의 서로 다른 FastAPI 서버가 동시에 구독하고, 메시지가 모든 서버에 실시간으로 전달되는지 확인하는 과정입니다.

Redis Pub/Sub의 진가는 서버가 여러 대일 때 나타납니다. 동일한 소스로 두 개의 서버를 띄워 테스트해 봅니다.

1단계: 서버 2개 실행 (터미널 2개 사용)

bash
# 터미널 1 (기본 8000번 포트)
uvicorn pub_sub_server:app --port 8000 --reload

# 터미널 2 (8001번 포트)
uvicorn pub_sub_server:app --port 8001 --reload

2단계: 브라우저 접속(구독)

  • http://localhost:8000/pub_sub 접속 (서버 1 수신자)
  • http://localhost:8001/pub_sub 접속 (서버 2 수신자)

3단계: 메시지 발행 (Swagger를 통한 메시지 발행)

두 서버 모두에서 (Swagger UI로) 메시지를 서로 발행해 봅니다.

  • http://localhost:8000/docs
  • http://localhost:8001/docs

🧐 확인 결과 (Observation)

결과: 8000번 탭과 8001번 탭 동시에 알림 메시지가 출력됩니다.

의미: - Redis가 발행한 메시지는 보관되지 않고 즉시 모든 구독자에게 전송됩니다(Push). - 서버가 100대, 1,000대로 늘어나더라도 동일한 채널을 구독하고 있다면 모든 유저에게 실시간으로 메시지를 보낼 수 있습니다. - 단, 구독자가 순간적으로 끊기거나 처리 중 장애가 나면 그 사이 메시지는 복구되지 않고 유실될 수 있다는 점도 유의해야 합니다.

🚩 정리 — 이 구조가 안고 있는 문제

  • 브로드캐스팅: Redis Pub/Sub은 1:N 통신을 지원하므로 분산 서버 환경에서 이벤트를 공유하는 가장 강력한 수단입니다.
  • 커넥션 분리: rd.pubsub()으로 만든 전용 객체를 사용해야 API 서버의 일반적인 데이터 요청(GET/SET) 커넥션이 마비되지 않습니다.
  • SSE 활용: 웹소켓보다 가볍고 구현이 간단한 SSE는 서버에서 클라이언트로 쏘는 단방향 실시간 알림 시스템에 최적입니다.
  • Pub/Sub은 메시지를 보관하지 않으므로, 현재 접속 중인 유저에게 즉각적인 휘발성 알림을 줄 때 가장 빠르고 저렴합니다.
  • Pub/Sub은 놓쳐도 치명적이지 않은 휘발성 메시지에는 잘 맞지만, "반드시 처리돼야 하는 주문/결제 이벤트 처리" 같은 비즈니스에는 Redis Streams나 별도 MQ를 고려하는 게 좋습니다.

5-12. 패턴 09-개선: 리스너 하나로 커넥션 고갈 막기

💡 이 강의 노트는 바로 직전 강의인 '패턴 09: 실시간 시스템 알림 - FastAPI와 Pub/Sub 연동'의 FastAPI 코드의 문제점을 개선해서 권장되는 코드 패턴을 제시하기 위해 작성된 강의 노트입니다. 직전 강의를 먼저 듣고 이 노트를 확인해 주세요.

1. 접속자마다 구독을 여는 방식이 터지는 지점

이전 예제처럼 클라이언트의 요청이 올 때마다 rd.pubsub()을 호출하면 다음과 같은 문제가 발생합니다.

  • 연결 폭주 — 동시 접속자가 1,000명이면 Redis 커넥션도 1,000개가 생성됩니다. Redis 설정의 maxclients를 초과하여 서버가 중단될 수 있습니다.
  • 불필요한 중복 — 동일한 채널을 구독하는데 모든 서버가 각자 Redis에 물어보는 것은 자원 낭비입니다.

✅ 해결책: 싱글 리스너 & 팬아웃 (Fan-out)

애플리케이션은 단 1개의 구독 전용 연결만 유지합니다. Redis에서 메시지가 오면, 애플리케이션이 메모리에 들고 있는 클라이언트 목록(Subscribers)을 순회하며 메시지를 직접 나누어 줍니다.

2. 단일 리스너와 큐로 구조 바꾸기

① 개선된 서버 코드 (pub_sub_server.py)

python
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, FileResponse
import redis.asyncio as redis


async def redis_listener(app: FastAPI):
    """
    [핵심] Redis 채널을 구독하고 메시지를 앱 내부 클라이언트들에게 전달하는 단일 루프
    """
    pubsub = app.state.redis.pubsub()
    await pubsub.subscribe("system:notices")

    try:
        while True:
            # 메시지 대기
            message = await pubsub.get_message(ignore_subscribe_messages=True, timeout=1.0)

            if message and message["type"] == "message":
                data = message["data"]
                # [개선] 순회 중 집합 변경으로 인한 에러 방지를 위해 list() 사본으로 순회
                if app.state.connected_clients:
                    for client_queue in list(app.state.connected_clients):
                        # 큐가 가득 찼을 경우를 대비해 비블로킹으로 전송 시도
                        try:
                            client_queue.put_nowait(data)
                        except asyncio.QueueFull:
                            pass

            await asyncio.sleep(0.01)

    except asyncio.CancelledError:
        # 태스크 종료 시 정상 흐름으로 간주하고 전파
        raise
    except Exception as e:
        print(f"Listener Error: {e}")
    finally:
        await pubsub.unsubscribe("system:notices")
        await pubsub.close()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 1. Redis 연결 및 클라이언트 관리 셋업
    app.state.redis = redis.from_url("redis://localhost:6379/0", decode_responses=True)
    app.state.connected_clients = set()  # 전역보다 구조적인 app.state 활용

    # 2. 백그라운드에서 단일 리스너 실행
    listener_task = asyncio.create_task(redis_listener(app))

    yield

    # 3. [개선] 종료 시 리스너 중단 및 명시적 종료 대기
    listener_task.cancel()
    try:
        await listener_task
    except asyncio.CancelledError:
        pass

    await app.state.redis.aclose()


app = FastAPI(lifespan=lifespan)


@app.get("/pub_sub")
async def index():
    return FileResponse("pub_sub.html")


@app.post("/publish-notice")
async def send_notice(message: str, request: Request):
    """관리자가 공지를 발행합니다."""
    await request.app.state.redis.publish("system:notices", message)
    return {"status": "sent", "active_clients": len(request.app.state.connected_clients)}


@app.get("/stream-notices")
async def stream_notices(request: Request):
    """클라이언트 연결 시 내부 큐를 생성하고 관리합니다."""

    # [설명] 실무에서는 큐 크기 제한(maxsize)을 두어 느린 클라이언트로 인한 메모리 폭증을 방지합니다.
    client_queue = asyncio.Queue(maxsize=100)
    request.app.state.connected_clients.add(client_queue)

    async def event_generator():
        try:
            while True:
                try:
                    # 5초 간격으로 연결 상태 확인을 위한 wait_for
                    data = await asyncio.wait_for(client_queue.get(), timeout=5.0)
                    yield f"data: {data}\n\n"
                except asyncio.TimeoutError:
                    pass

                if await request.is_disconnected():
                    break
        finally:
            # [개선] remove() 대신 discard()를 사용하여 이미 제거된 경우의 예외 방지
            request.app.state.connected_clients.discard(client_queue)

    return StreamingResponse(event_generator(), media_type="text/event-stream")

🔍 바뀐 부분을 코드로 짚어보기

  • app.state.connected_clients = set() : 현재 서버에 SSE로 연결된 모든 클라이언트의 asyncio.Queue 객체들을 보관합니다. 전역 변수보다 FastAPI의 app.state에 저장하는 것이 관리 및 테스트 관점에서 훨씬 구조적입니다. set을 사용하여 중복을 방지하고 빠르게 추가/삭제합니다.
  • redis_listener : 서버가 시작될 때 딱 하나만 실행되는 백그라운드 태스크입니다. Redis와 단일 채널을 유지하며, 메시지가 올 때마다 내부 connected_clients를 순회하며 각 큐에 메시지를 넣어줍니다(Fan-out).
  • list(app.state.connected_clients) : 메시지를 분배(Fan-out)할 때 사본 리스트를 만들어 순회합니다. 순회 도중 새로운 클라이언트가 접속하거나 종료되어 집합(set)의 크기가 변하더라도 RuntimeError 없이 안전하게 동작합니다.
  • asyncio.CancelledError 처리: 리스너는 서버 종료 시 cancel()되므로, 취소는 에러라기보다 정상 종료 흐름입니다. 이를 일반 Exception과 분리하여 잡음으로써 비정상적인 에러 로그가 남지 않도록 깔끔하게 정리합니다.
  • lifespan : FastAPI 앱의 시작과 종료를 관리합니다. 서버 시작 시 create_task를 통해 리스너를 실행하고, 서버 종료 시 cancel() 후 await를 통해 태스크가 리소스를 모두 정리하고 종료될 때까지 명시적으로 기다려주는 가장 깔끔한 정리 방식입니다.
  • client_queue = asyncio.Queue(maxsize=100) : 각 브라우저 연결마다 생성되는 고유한 메시지함입니다. Redis에서 직접 메시지를 받는 것이 아니라, 앱의 리스너가 던져주는 메시지를 잠시 담아두는 역할을 합니다. 실무에서는 큐 크기 제한(maxsize)을 두어 느린 클라이언트로 인해 큐가 무한정 커져 서버 메모리를 고갈시키는 현상을 방지합니다.
  • await asyncio.wait_for(..., timeout=5.0) : 연결 해제를 즉각 감지하기 위한 핵심 장치입니다. 만약 wait_for 없이 get()만 수행하면 메시지가 올 때까지 코드가 멈춰있어, 클라이언트가 이미 브라우저를 닫았더라도 이를 감지하지 못하고 자원을 계속 점유하게 됩니다. 타임아웃을 통해 주기적으로 루프를 깨워 연결 상태를 확인함으로써 좀비 세션을 방지하고 메모리 누수를 막습니다.
  • discard(client_queue) : remove()와 달리 대상이 집합에 없어도 에러(KeyError)를 발생시키지 않습니다. 클라이언트가 이미 어떤 이유로 목록에서 빠진 상태에서 중복으로 정리 로직이 돌아가더라도 안전하게 처리됩니다.

이하 클라이언트 코드와 테스트 방법은 이전 강의와 동일합니다.

저작자표시 비영리 동일조건 (새창열림)

'Database > redis' 카테고리의 다른 글

Redis 실무 가이드 4 — 운영·장애 대응과 대용량 캐시 설계  (1) 2026.09.08
Redis 실무 가이드 2 - Pub/Sub, Stream, Redis Stack, 트랜잭션과 Lua Script  (0) 2026.09.08
Redis 실무 가이드 1 - 왜 Redis인가, 그리고 자료형 완전 정복  (0) 2026.09.08
Redis Stream에 대한 정리  (0) 2025.01.30
Redis-OM 객체 매핑을 지원하는 redis 라이브러리  (0) 2025.01.30

댓글

티스토리툴바