Python Redis 활용 예제
import redis
import json
from datetime import timedelta
from typing import Optional, Any
# Redis 클라이언트 초기화
redis_client = redis.Redis(
host='localhost',
port=6379,
db=0,
decode_responses=True # 문자열 자동 디코딩
)
# ================================
# 1. 기본 Key-Value 연산
# ================================
def cache_user(user_id: int, user_data: dict, ttl_seconds: int = 3600):
"""사용자 정보를 캐시에 저장"""
key = f"user:{user_id}"
redis_client.setex(key, ttl_seconds, json.dumps(user_data))
def get_cached_user(user_id: int) -> Optional[dict]:
"""캐시에서 사용자 정보 조회"""
key = f"user:{user_id}"
data = redis_client.get(key)
return json.loads(data) if data else None
# 사용
cache_user(123, {"name": "홍길동", "email": "hong@example.com"})
user = get_cached_user(123)
print(user) # {'name': '홍길동', 'email': 'hong@example.com'}
# ================================
# 2. 캐시 패턴: Cache-Aside
# ================================
def get_user_with_cache(user_id: int, db_session) -> dict:
"""캐시 우선 조회, 없으면 DB에서 로드"""
cache_key = f"user:{user_id}"
# 1. 캐시 확인
cached = redis_client.get(cache_key)
if cached:
print("Cache HIT")
return json.loads(cached)
# 2. 캐시 미스 -> DB 조회
print("Cache MISS")
user = db_session.query(User).get(user_id)
if user:
user_dict = {"id": user.id, "name": user.name, "email": user.email}
# 3. 캐시에 저장 (1시간 TTL)
redis_client.setex(cache_key, 3600, json.dumps(user_dict))
return user_dict
return None
# ================================
# 3. 분산 락 구현
# ================================
import uuid
def acquire_lock(lock_name: str, timeout: int = 10) -> Optional[str]:
"""분산 락 획득"""
lock_key = f"lock:{lock_name}"
lock_id = str(uuid.uuid4())
# SET NX (Not eXists) - 키가 없을 때만 설정
acquired = redis_client.set(lock_key, lock_id, nx=True, ex=timeout)
return lock_id if acquired else None
def release_lock(lock_name: str, lock_id: str) -> bool:
"""분산 락 해제 (Lua 스크립트로 원자적 처리)"""
lock_key = f"lock:{lock_name}"
# Lua 스크립트: 본인이 획득한 락인지 확인 후 삭제
script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = redis_client.eval(script, 1, lock_key, lock_id)
return result == 1
# 사용
lock_id = acquire_lock("payment:order:123")
if lock_id:
try:
# 결제 처리 (동시에 하나만 실행됨)
process_payment(order_id=123)
finally:
release_lock("payment:order:123", lock_id)
# ================================
# 4. 조회수/카운터 (Atomic Operations)
# ================================
def increment_view_count(article_id: int) -> int:
"""게시글 조회수 증가 (원자적 연산)"""
key = f"article:{article_id}:views"
return redis_client.incr(key)
def get_view_count(article_id: int) -> int:
"""조회수 조회"""
key = f"article:{article_id}:views"
count = redis_client.get(key)
return int(count) if count else 0
# ================================
# 5. 리더보드 (Sorted Set 활용)
# ================================
def update_leaderboard(game_id: str, user_id: str, score: float):
"""리더보드 점수 업데이트"""
key = f"leaderboard:{game_id}"
redis_client.zadd(key, {user_id: score})
def get_top_players(game_id: str, count: int = 10) -> list:
"""상위 N명 조회 (점수 내림차순)"""
key = f"leaderboard:{game_id}"
# ZREVRANGE: 점수 높은 순으로 조회
return redis_client.zrevrange(key, 0, count - 1, withscores=True)
def get_user_rank(game_id: str, user_id: str) -> Optional[int]:
"""특정 사용자의 순위 조회 (0부터 시작)"""
key = f"leaderboard:{game_id}"
rank = redis_client.zrevrank(key, user_id)
return rank + 1 if rank is not None else None # 1부터 시작하도록
# 사용
update_leaderboard("game1", "user:123", 1500)
update_leaderboard("game1", "user:456", 2200)
top_10 = get_top_players("game1", 10)
my_rank = get_user_rank("game1", "user:123")
print(f"내 순위: {my_rank}위")