Redis 캐싱 전략: FastAPI 성능 10배 올리기
API가 느리다는 말을 들을 때만큼 뼈아픈 순간이 없죠. 대부분의 경우 원인은 단순합니다. 똑같은 DB 쿼리를 매번 반복 실행하고 있기 때문입니다. Redis 캐싱 하나로 이 문제를 근본적으로 해결할 수 있습니다.
이 글에서는 FastAPI 프로젝트에 Redis를 실전 적용하는 방법을 단계별로 살펴봅니다.
Redis가 성능을 끌어올리는 원리
Redis는 인메모리 키-값 저장소입니다. 디스크 기반 DB가 수십~수백 ms 걸리는 작업을 Redis는 1ms 이내로 처리합니다.
- 읽기 속도: 초당 수십만 건 처리 가능
- DB 부하 감소: 반복 쿼리를 캐시로 차단
- 비동기 친화: FastAPI의 async/await와 자연스럽게 결합
핵심 캐싱 패턴 4가지
1. Cache-Aside (Lazy Loading)
가장 보편적인 패턴입니다. 요청이 들어올 때 캐시를 먼저 확인하고, 없으면 DB에서 가져와 캐시에 저장합니다.
import redis.asyncio as redis
import json
from fastapi import FastAPI, Depends
app = FastAPI()
async def get_redis():
client = redis.Redis(host="localhost", port=6379, decode_responses=True)
try:
yield client
finally:
await client.aclose()
@app.get("/products/{product_id}")
async def get_product(product_id: int, redis_client=Depends(get_redis)):
cache_key = f"product:{product_id}"
# 1. 캐시 확인
cached = await redis_client.get(cache_key)
if cached:
return {"source": "cache", "data": json.loads(cached)}
# 2. DB에서 조회 (실제로는 SQLAlchemy 등 사용)
product = await fetch_from_db(product_id)
# 3. 캐시 저장 (TTL 60초)
await redis_client.setex(cache_key, 60, json.dumps(product))
return {"source": "db", "data": product}
2. Write-Through
데이터를 DB에 쓸 때 캐시도 동시에 갱신합니다. 일관성이 중요한 데이터에 적합합니다.
@app.put("/products/{product_id}")
async def update_product(product_id: int, data: dict, redis_client=Depends(get_redis)):
# DB 업데이트
updated = await update_db(product_id, data)
# 캐시도 동시에 갱신
cache_key = f"product:{product_id}"
await redis_client.setex(cache_key, 3600, json.dumps(updated))
return updated
3. Write-Behind (Lazy Write)
Redis에 먼저 쓰고 DB에는 비동기로 나중에 반영합니다. 쓰기 성능이 극대화되지만 장애 시 데이터 손실 위험이 있습니다. 로그, 분석 데이터 등에 활용됩니다.
4. Read-Through
캐시가 DB 조회까지 대신 담당합니다. 애플리케이션 코드가 단순해지는 장점이 있으며, aiocache 같은 라이브러리가 이 패턴을 지원합니다.
from aiocache import cached, Cache
@cached(ttl=300, cache=Cache.REDIS, key="popular_products")
async def get_popular_products():
return await fetch_popular_from_db()
TTL 설정 전략
TTL(Time To Live)은 캐시의 신선도를 결정합니다. 너무 짧으면 캐시 효과가 없고, 너무 길면 낡은 데이터를 제공합니다.
| 데이터 유형 | 권장 TTL | 이유 |
|---|---|---|
| 사용자 세션 | 30분 (1800초) | 보안 및 최신성 |
| 상품 목록 | 1시간 (3600초) | 변경 빈도 낮음 |
| 인기 게시물 | 10분 (600초) | 실시간성 중요 |
| 정적 설정값 | 24시간 (86400초) | 거의 변경 안 됨 |
| 실시간 재고 | 캐시 비권장 | 정확성 필수 |
# 데이터 종류에 따라 TTL을 달리 적용하는 예시
TTL_CONFIG = {
"session": 1800,
"product": 3600,
"popular": 600,
"config": 86400,
}
async def smart_cache_set(redis_client, key: str, data: dict, data_type: str):
ttl = TTL_CONFIG.get(data_type, 300)
await redis_client.setex(key, ttl, json.dumps(data, ensure_ascii=False))
Redis 제거 정책(Eviction Policy)
Redis 메모리가 꽉 찼을 때 어떤 키를 제거할지 결정하는 정책입니다.
# redis.conf 설정 예시
maxmemory 512mb
maxmemory-policy allkeys-lru
추천 정책:
- allkeys-lru: 일반적인 캐싱 용도에 가장 적합. 최근에 사용되지 않은 키부터 제거
- volatile-lru: TTL이 있는 키만 제거. 영구 저장 데이터와 캐시를 함께 쓸 때
- allkeys-lfu: 인기 데이터를 오래 유지하고 싶을 때. 접근 빈도 기반 제거
캐시 무효화 - 가장 어려운 문제
캐시의 어려움은 데이터가 변경됐을 때 캐시를 제때 지우는 것입니다.
# 이벤트 기반 캐시 무효화
@app.delete("/products/{product_id}")
async def delete_product(product_id: int, redis_client=Depends(get_redis)):
# DB에서 삭제
await delete_from_db(product_id)
# 관련 캐시 키 일괄 삭제
keys_to_delete = [
f"product:{product_id}",
"popular_products",
"product_list:*",
]
for key in keys_to_delete:
if "*" in key:
async for matched_key in redis_client.scan_iter(key):
await redis_client.delete(matched_key)
else:
await redis_client.delete(key)
return {"message": "삭제 완료, 캐시도 무효화됨"}
연결 풀링으로 오버헤드 줄이기
매 요청마다 Redis 연결을 새로 만들면 오히려 성능이 저하됩니다. 앱 시작 시 연결 풀을 생성해 재사용하세요.
from contextlib import asynccontextmanager
import redis.asyncio as redis
redis_pool = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global redis_pool
redis_pool = redis.ConnectionPool(
host="localhost",
port=6379,
max_connections=20,
decode_responses=True
)
yield
await redis_pool.aclose()
app = FastAPI(lifespan=lifespan)
async def get_redis():
return redis.Redis(connection_pool=redis_pool)
실전 성능 비교
간단한 벤치마크를 통해 Redis 캐싱 전후를 비교해 볼 수 있습니다.
import time
import asyncio
async def benchmark():
redis_client = redis.Redis(host="localhost", port=6379, decode_responses=True)
# DB 직접 조회 시뮬레이션 (50ms 가정)
start = time.perf_counter()
for _ in range(100):
await asyncio.sleep(0.05)
db_time = time.perf_counter() - start
# Redis 캐시 조회
await redis_client.set("bench_key", "cached_value")
start = time.perf_counter()
for _ in range(100):
await redis_client.get("bench_key")
cache_time = time.perf_counter() - start
print(f"DB 직접: {db_time:.2f}s")
print(f"Redis 캐시: {cache_time:.4f}s")
print(f"속도 향상: {db_time / cache_time:.0f}배")
실제 환경에서 DB 쿼리가 평균 50ms라면, Redis 캐시 히트는 0.5ms 미만으로 100배 이상 빠릅니다. 캐시 히트율이 80~90%만 돼도 전체 API 응답 시간은 극적으로 개선됩니다.
모니터링 - 캐시가 잘 작동하는지 확인하기
# Redis CLI에서 실행
redis-cli INFO stats | grep -E "keyspace_hits|keyspace_misses"
@app.get("/admin/cache-stats")
async def cache_stats(redis_client=Depends(get_redis)):
info = await redis_client.info("stats")
hits = info.get("keyspace_hits", 0)
misses = info.get("keyspace_misses", 0)
total = hits + misses
hit_rate = (hits / total * 100) if total > 0 else 0
return {
"hits": hits,
"misses": misses,
"hit_rate": f"{hit_rate:.1f}%",
"status": "good" if hit_rate >= 80 else "needs_review"
}
보안 체크리스트
- Redis를 공개 인터넷에 직접 노출하지 마세요
- requirepass 옵션으로 비밀번호 설정
- SSL/TLS 연결 사용 (rediss:// 스킴)
- 민감한 개인정보는 암호화 후 캐시에 저장
코드벤터는 FastAPI 기반 프로젝트에서 Redis 캐싱을 기본 아키텍처 요소로 활용하고 있습니다. 반복적인 DB 쿼리를 제거하고 응답 속도를 높이는 것은 사용자 경험 개선의 첫 걸음입니다. 캐싱 전략은 한 번에 완벽하게 구현하기보다, 실제 트래픽 패턴과 히트율 데이터를 보면서 점진적으로 최적화하는 것을 권장합니다. 코드픽에서 더 많은 실전 개발 인사이트를 확인해 보세요.