Redis 캐싱 전략: FastAPI 성능 10배 올리기

FastAPI 서버가 느리다면, 병목은 대부분 데이터베이스 반복 조회에 있습니다. Redis를 캐시 레이어로 도입하면 동일한 데이터를 DB에 매번 요청하지 않고 인메모리에서 즉시 반환할 수 있어, 응답 시간이 극적으로 줄어듭니다. 이 글에서는 Redis 캐싱의 핵심 전략부터 FastAPI 실전 구현까지 단계별로 살펴봅니다.

왜 Redis인가?

Redis(Remote Dictionary Server)는 초당 수십만 건의 읽기/쓰기를 처리할 수 있는 인메모리 Key-Value 저장소입니다. PostgreSQL이나 MySQL 같은 디스크 기반 DB는 쿼리 한 번에 수 밀리초~수십 밀리초가 소요되지만, Redis 조회는 보통 0.1ms 미만입니다.

트래픽이 몰리는 API 엔드포인트에 Redis 캐싱을 붙이면 다음 효과를 기대할 수 있습니다.

DB 부하 90% 이상 감소
평균 응답 시간 10~100배 단축
서버 비용 절감 및 수평 확장 용이

핵심 캐싱 패턴 4가지

1. Cache-Aside (Lazy Loading)

가장 범용적인 패턴입니다. 요청이 들어오면 먼저 Redis를 조회하고, 없으면(Cache Miss) DB에서 가져와 Redis에 저장합니다.

python

import json
import redis.asyncio as aioredis
from fastapi import FastAPI

app = FastAPI()
redis_client = aioredis.from_url("redis://localhost:6379", decode_responses=True)

async def get_user(user_id: int):
    cache_key = f"user:{user_id}"
    cached = await redis_client.get(cache_key)
    if cached:
        return json.loads(cached)  # Cache Hit
    # Cache Miss: DB 조회
    user = await db.fetch_user(user_id)
    await redis_client.setex(cache_key, 300, json.dumps(user))  # TTL 5분
    return user

2. Write-Through

데이터를 쓸 때 DB와 Redis에 동시에 저장합니다. 항상 최신 데이터가 캐시에 존재하지만 쓰기 지연이 약간 증가합니다.

python

async def update_user(user_id: int, data: dict):
    await db.update_user(user_id, data)
    cache_key = f"user:{user_id}"
    await redis_client.setex(cache_key, 300, json.dumps(data))
    return data

3. Write-Behind (Write-Back)

캐시에만 먼저 쓰고, 이후 비동기로 DB에 반영합니다. 쓰기 성능이 가장 빠르지만 캐시 장애 시 데이터 손실 위험이 있습니다. 로그나 이벤트처럼 유실 허용 데이터에 적합합니다.

4. fastapi-cache2 데코레이터

라이브러리를 활용하면 단 두 줄로 캐싱을 적용할 수 있습니다.

python

from fastapi_cache import FastAPICache
from fastapi_cache.backends.redis import RedisBackend
from fastapi_cache.decorator import cache

@app.on_event("startup")
async def startup():
    redis = aioredis.from_url("redis://localhost:6379")
    FastAPICache.init(RedisBackend(redis), prefix="myapp")

@app.get("/products/{product_id}")
@cache(expire=60)
async def get_product(product_id: int):
    return await db.fetch_product(product_id)

TTL과 축출 정책(Eviction Policy)

TTL(Time-To-Live) 설계 원칙

데이터 특성에 따라 TTL을 다르게 설정해야 합니다.

데이터 유형	권장 TTL	이유
사용자 세션	30분	보안 및 최신성
상품 목록	5~10분	재고/가격 변동 고려
공지사항/FAQ	1~6시간	변경 빈도 낮음
통계/집계 데이터	10~30분	근사치 허용
설정/코드 테이블	24시간	거의 변하지 않음

축출 정책 선택 가이드

Redis의 maxmemory에 도달했을 때 어떤 키를 삭제할지 결정하는 정책입니다.

bash

# redis.conf 또는 Redis CLI에서 설정
CONFIG SET maxmemory 512mb
CONFIG SET maxmemory-policy allkeys-lru

allkeys-lru: 가장 오래 사용하지 않은 키 제거. 일반 캐시용으로 추천
volatile-lru: TTL 있는 키 중 LRU 적용. 영구 데이터와 캐시 혼용 시
allkeys-lfu: 접근 빈도가 낮은 키 제거. 인기도 기반 캐시에 적합
volatile-ttl: TTL이 가장 짧게 남은 키 먼저 제거

FastAPI Lifespan으로 연결 관리

FastAPI 최신 버전에서는 on_event 대신 lifespan을 사용합니다.

python

from contextlib import asynccontextmanager
from fastapi import FastAPI
import redis.asyncio as aioredis

redis_pool = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global redis_pool
    redis_pool = aioredis.ConnectionPool.from_url(
        "redis://localhost:6379",
        max_connections=20,
        decode_responses=True
    )
    yield
    await redis_pool.disconnect()

app = FastAPI(lifespan=lifespan)

async def get_redis():
    return aioredis.Redis(connection_pool=redis_pool)

캐시 무효화 전략

올바른 무효화 없이는 오래된 데이터가 캐시에 남아 심각한 버그를 일으킬 수 있습니다.

python

# 데이터 변경 시 명시적 캐시 삭제
async def delete_product(product_id: int, redis: aioredis.Redis):
    await db.delete_product(product_id)
    await redis.delete(f"product:{product_id}")
    await redis.delete("products:list")

# 패턴 기반 대량 삭제 (운영 환경에서 SCAN 사용 권장)
async def invalidate_by_pattern(pattern: str, redis: aioredis.Redis):
    cursor = 0
    while True:
        cursor, keys = await redis.scan(cursor, match=pattern, count=100)
        if keys:
            await redis.delete(*keys)
        if cursor == 0:
            break

성능 모니터링

Redis의 상태를 지속적으로 모니터링하면 캐시 효율을 최적화할 수 있습니다.

bash

# Redis CLI로 실시간 통계 확인
redis-cli INFO stats | grep -E "keyspace_hits|keyspace_misses|evicted_keys"

핵심 지표 세 가지를 주기적으로 확인하세요.

캐시 히트율(Hit Rate): 80% 이상 유지 목표
축출 키 수(Evicted Keys): 급증하면 maxmemory 증설 검토
메모리 사용량: maxmemory의 70~80% 이하 유지

실전 팁: Cache Stampede 방지

인기 캐시 키가 만료되는 순간 수백 개의 요청이 동시에 DB를 조회하는 현상을 Cache Stampede라 합니다. 분산 락으로 방지할 수 있습니다.

python

import asyncio

async def get_with_lock(key: str, ttl: int, fetch_fn, redis: aioredis.Redis):
    cached = await redis.get(key)
    if cached:
        return json.loads(cached)

    lock_key = f"lock:{key}"
    acquired = await redis.set(lock_key, "1", ex=10, nx=True)

    if acquired:
        try:
            data = await fetch_fn()
            await redis.setex(key, ttl, json.dumps(data))
            return data
        finally:
            await redis.delete(lock_key)
    else:
        await asyncio.sleep(0.1)
        return await get_with_lock(key, ttl, fetch_fn, redis)

정리

Redis 캐싱은 FastAPI 서버의 성능을 단기간에 극적으로 향상시킬 수 있는 검증된 방법입니다. Cache-Aside 패턴으로 시작해 데이터 특성에 맞는 TTL을 설계하고, 캐시 히트율을 모니터링하면서 점진적으로 최적화하는 것이 실용적인 접근법입니다.

코드벤터는 스타트업과 1인 개발자가 복잡한 인프라 없이도 프로덕션 수준의 성능을 낼 수 있도록, 실전 중심의 기술 콘텐츠를 지속적으로 공유합니다. Redis 캐싱 도입에 어려움이 있다면 코드픽(codepick.kr)의 전문가 네트워크를 통해 도움을 받아보세요.

Redis 캐싱 전략: FastAPI 성능 10배 올리기

Redis 캐싱 전략: FastAPI 성능 10배 올리기

왜 Redis인가?

핵심 캐싱 패턴 4가지

1. Cache-Aside (Lazy Loading)

2. Write-Through

3. Write-Behind (Write-Back)

4. fastapi-cache2 데코레이터

TTL과 축출 정책(Eviction Policy)

TTL(Time-To-Live) 설계 원칙

축출 정책 선택 가이드

FastAPI Lifespan으로 연결 관리

캐시 무효화 전략

성능 모니터링

실전 팁: Cache Stampede 방지

정리

AI 서비스나 플랫폼 개발을고민 중이신가요?

AI 서비스나 플랫폼 개발을
고민 중이신가요?