Django Redis 캐시 완전 가이드: 성능 최적화의 핵심
Django 애플리케이션의 성능 병목을 해결하고 싶으신가요? Redis를 활용한 캐시 시스템은 Django 성능 최적화의 핵심 요소입니다. 이 가이드에서는 Django에서 Redis 캐시를 구현하는 방법부터 고급 최적화 기법까지 실전 경험을 바탕으로 상세히 알아보겠습니다....
Django 애플리케이션의 성능 병목을 해결하고 싶으신가요? Redis를 활용한 캐시 시스템은 Django 성능 최적화의 핵심 요소입니다. 이 가이드에서는 Django에서 Redis 캐시를 구현하는 방법부터 고급 최적화 기법까지 실전 경험을 바탕으로 상세히 알아보겠습니다.
📚 Redis 캐시의 기본 이해
왜 Redis인가?
Redis(Remote Dictionary Server)는 메모리 기반 키-값 저장소로, Django 캐시 백엔드로 가장 인기 있는 선택입니다:
Redis의 장점:
- 🚀 빠른 성능: 메모리 기반으로 마이크로초 단위 응답
- 🔄 다양한 데이터 구조: String, Hash, List, Set, Sorted Set 지원
- 💾 영속성: 데이터 백업 및 복구 가능
- 🔀 확장성: 클러스터링 및 레플리케이션 지원
- 🎯 만료 시간: TTL(Time To Live) 자동 관리
Django 캐시 프레임워크 개요
Django는 여러 캐시 레벨을 제공합니다:
# 캐시 레벨별 적용 범위
1. 사이트 전체 캐시 # 가장 상위 레벨
2. 뷰 레벨 캐시 # 특정 뷰 결과 캐시
3. 템플릿 조각 캐시 # 템플릿 일부 캐시
4. 저수준 캐시 API # 개발자가 직접 제어
🛠️ Redis 설치 및 Django 설정
Redis 서버 설치
# macOS (Homebrew)
brew install redis
brew services start redis
# Ubuntu/Debian
sudo apt-get update
sudo apt-get install redis-server
sudo systemctl start redis-server
# CentOS/RHEL
sudo yum install redis
sudo systemctl start redis
# Docker 사용
docker run -d --name redis-server -p 6379:6379 redis:alpine
Django-Redis 설치 및 설정
# Django-Redis 패키지 설치
pip install django-redis
# settings.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1', # 데이터베이스 1 사용
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
'CONNECTION_POOL_KWARGS': {
'max_connections': 20,
'retry_on_timeout': True,
},
'SERIALIZER': 'django_redis.serializers.json.JSONSerializer',
'COMPRESSOR': 'django_redis.compressors.zlib.ZlibCompressor',
},
'KEY_PREFIX': 'myproject', # 키 충돌 방지
'VERSION': 1,
'TIMEOUT': 300, # 5분 기본 만료 시간
}
}
# 캐시 키 생성 함수 설정
def make_key(key, key_prefix, version):
"""캐시 키 생성 로직 커스터마이징"""
return f"{key_prefix}:{version}:{key}"
CACHES['default']['KEY_FUNCTION'] = 'myproject.utils.cache.make_key'
환경별 Redis 설정
# settings/base.py
import os
REDIS_URL = os.environ.get('REDIS_URL', 'redis://127.0.0.1:6379/1')
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': REDIS_URL,
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
},
}
}
# settings/development.py
CACHES['default']['OPTIONS'].update({
'CONNECTION_POOL_KWARGS': {
'max_connections': 5,
}
})
# settings/production.py
CACHES['default']['OPTIONS'].update({
'CONNECTION_POOL_KWARGS': {
'max_connections': 50,
'retry_on_timeout': True,
'retry_on_error': [ConnectionError, TimeoutError],
},
'SERIALIZER': 'django_redis.serializers.msgpack.MSGPackSerializer', # 더 빠른 직렬화
'COMPRESSOR': 'django_redis.compressors.zlib.ZlibCompressor',
})
🎯 Django 캐시 구현 전략
1. 사이트 전체 캐시
가장 간단하고 효과적인 캐시 방법입니다:
# settings.py
MIDDLEWARE = [
'django.middleware.cache.UpdateCacheMiddleware', # 맨 위
'django.middleware.common.CommonMiddleware',
# ... 다른 미들웨어들
'django.middleware.cache.FetchFromCacheMiddleware', # 맨 아래
]
# 사이트 전체 캐시 설정
CACHE_MIDDLEWARE_ALIAS = 'default'
CACHE_MIDDLEWARE_SECONDS = 600 # 10분
CACHE_MIDDLEWARE_KEY_PREFIX = 'site'
주의사항:
- 사용자별 개인화 콘텐츠가 있으면 사용 불가
- 실시간 업데이트가 필요한 사이트에는 부적합
2. 뷰 레벨 캐시
특정 뷰의 결과를 캐시합니다:
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator
from django.views.generic import ListView
# 함수 기반 뷰
@cache_page(60 * 15) # 15분 캐시
def product_list(request):
products = Product.objects.select_related('category').all()
return render(request, 'products/list.html', {'products': products})
# 클래스 기반 뷰
@method_decorator(cache_page(60 * 15), name='dispatch')
class ProductListView(ListView):
model = Product
template_name = 'products/list.html'
context_object_name = 'products'
def get_queryset(self):
return Product.objects.select_related('category').filter(active=True)
# 조건부 캐시
from django.views.decorators.vary import vary_on_headers
@cache_page(60 * 15)
@vary_on_headers('User-Agent', 'Accept-Language')
def localized_view(request):
# 사용자 에이전트와 언어별로 다른 캐시
return render(request, 'localized.html')
3. 템플릿 조각 캐시
템플릿의 특정 부분만 캐시합니다:
<!-- products/list.html -->
```html
<!-- products/list.html -->
{%raw%}{% load cache %}{%endraw%}
<!-- 비싼 계산 결과 캐시 -->
{%raw%}{% cache 300 expensive_calculation request.user.id %}{%endraw%}
<div class="statistics">
{%raw%}{% for stat in expensive_stats %}{%endraw%}
<div class="stat-item">{%raw%}{{ stat.name }}: {{ stat.value }}{%endraw%}</div>
{%raw%}{% endfor %}{%endraw%}
</div>
{%raw%}{% endcache %}{%endraw%}
<!-- 카테고리별 캐시 -->
{%raw%}{% for category in categories %}{%endraw%}
{%raw%}{% cache 600 category_products category.id %}{%endraw%}
<div class="category-section">
<h3>{%raw%}{{ category.name }}{%endraw%}</h3>
{%raw%}{% for product in category.products.all %}{%endraw%}
<div class="product-item">{%raw%}{{ product.name }}{%endraw%}</div>
{%raw%}{% endfor %}{%endraw%}
</div>
{%raw%}{% endcache %}{%endraw%}
{%raw%}{% endfor %}{%endraw%}
<!-- 조건부 템플릿 캐시 -->
{%raw%}{% cache 300 user_specific_content request.user.id only if not user.is_staff %}{%endraw%}
<!-- 일반 사용자에게만 캐시 적용 -->
<div class="user-dashboard">
<!-- 복잡한 사용자 대시보드 -->
</div>
{%raw%}{% endcache %}{%endraw%}
4. 저수준 캐시 API
가장 세밀한 제어가 가능합니다:
from django.core.cache import cache
from django.core.cache.utils import make_template_fragment_key
import hashlib
import json
class ProductService:
@staticmethod
def get_product_stats(product_id):
"""제품 통계 캐시"""
cache_key = f'product_stats:{product_id}'
stats = cache.get(cache_key)
if stats is None:
# 복잡한 통계 계산
stats = {
'total_sales': Product.objects.get(id=product_id).calculate_total_sales(),
'avg_rating': Product.objects.get(id=product_id).calculate_avg_rating(),
'view_count': Product.objects.get(id=product_id).get_view_count(),
}
cache.set(cache_key, stats, timeout=3600) # 1시간 캐시
return stats
@staticmethod
def get_related_products(product_id, limit=5):
"""관련 상품 캐시"""
cache_key = f'related_products:{product_id}:{limit}'
products = cache.get(cache_key)
if products is None:
# 복잡한 추천 알고리즘
products = Product.objects.filter(
category__products__id=product_id
).exclude(id=product_id)[:limit]
# QuerySet을 직렬화 가능한 형태로 변환
products_data = [
{
'id': p.id,
'name': p.name,
'price': str(p.price),
'image_url': p.image.url if p.image else None,
}
for p in products
]
cache.set(cache_key, products_data, timeout=1800) # 30분 캐시
products = products_data
return products
# 캐시 키 생성 헬퍼
def generate_cache_key(*args, **kwargs):
"""일관된 캐시 키 생성"""
key_parts = [str(arg) for arg in args]
key_parts.extend([f"{k}:{v}" for k, v in sorted(kwargs.items())])
key_string = ":".join(key_parts)
return hashlib.md5(key_string.encode()).hexdigest()
# 사용 예시
def get_user_recommendations(user_id, category=None, limit=10):
cache_key = generate_cache_key(
'user_recommendations',
user_id=user_id,
category=category,
limit=limit
)
recommendations = cache.get(cache_key)
if recommendations is None:
# 복잡한 추천 로직
recommendations = calculate_recommendations(user_id, category, limit)
cache.set(cache_key, recommendations, timeout=7200) # 2시간
return recommendations
🔄 캐시 무효화 전략
신호 기반 자동 무효화
from django.db.models.signals import post_save, post_delete
from django.dispatch import receiver
from django.core.cache import cache
from django.core.cache.utils import make_template_fragment_key
@receiver(post_save, sender=Product)
def invalidate_product_cache(sender, instance, **kwargs):
"""상품 저장시 관련 캐시 무효화"""
# 상품별 캐시 삭제
cache.delete(f'product_stats:{instance.id}')
cache.delete(f'product_detail:{instance.id}')
# 카테고리 관련 캐시 삭제
cache.delete(f'category_products:{instance.category_id}')
# 템플릿 조각 캐시 삭제
fragment_key = make_template_fragment_key(
'product_detail',
[instance.id]
)
cache.delete(fragment_key)
# 관련 상품 캐시 삭제 (패턴 매칭)
cache.delete_pattern(f'related_products:*')
@receiver(post_delete, sender=Product)
def invalidate_product_cache_on_delete(sender, instance, **kwargs):
"""상품 삭제시 캐시 정리"""
# 모든 관련 캐시 삭제
invalidate_product_cache(sender, instance, **kwargs)
# 추가적으로 카테고리 개수 캐시도 무효화
cache.delete(f'category_count:{instance.category_id}')
버전 기반 캐시 관리
from django.core.cache import cache
from django.conf import settings
class VersionedCache:
"""버전 관리가 가능한 캐시 래퍼"""
def __init__(self, namespace):
self.namespace = namespace
def _get_version_key(self):
return f'{self.namespace}:version'
def _get_versioned_key(self, key):
version = cache.get(self._get_version_key(), 1)
return f'{self.namespace}:{version}:{key}'
def get(self, key, default=None):
versioned_key = self._get_versioned_key(key)
return cache.get(versioned_key, default)
def set(self, key, value, timeout=None):
versioned_key = self._get_versioned_key(key)
cache.set(versioned_key, value, timeout)
def invalidate_all(self):
"""네임스페이스의 모든 캐시 무효화"""
version_key = self._get_version_key()
current_version = cache.get(version_key, 1)
cache.set(version_key, current_version + 1)
# 사용 예시
product_cache = VersionedCache('products')
def get_product_list():
products = product_cache.get('list')
if products is None:
products = Product.objects.all().values('id', 'name', 'price')
product_cache.set('list', list(products), timeout=3600)
return products
# 모든 상품 캐시 무효화
def invalidate_all_product_caches():
product_cache.invalidate_all()
태그 기반 캐시 무효화
from django.core.cache import cache
from django.core.cache.backends.base import DEFAULT_TIMEOUT
class TaggedCache:
"""태그 기반 캐시 관리"""
@staticmethod
def set_with_tags(key, value, tags, timeout=DEFAULT_TIMEOUT):
"""태그와 함께 캐시 설정"""
cache.set(key, value, timeout)
# 각 태그에 대해 키 목록 관리
for tag in tags:
tag_key = f'tag:{tag}'
tagged_keys = cache.get(tag_key, set())
tagged_keys.add(key)
cache.set(tag_key, tagged_keys, timeout=None) # 태그는 영구 보존
@staticmethod
def invalidate_tag(tag):
"""특정 태그의 모든 캐시 무효화"""
tag_key = f'tag:{tag}'
tagged_keys = cache.get(tag_key, set())
if tagged_keys:
# 태그된 모든 키 삭제
cache.delete_many(list(tagged_keys))
# 태그 자체도 삭제
cache.delete(tag_key)
# 사용 예시
def cache_product_with_tags(product):
cache_key = f'product:{product.id}'
tags = [
f'category:{product.category_id}',
f'brand:{product.brand_id}',
'products'
]
TaggedCache.set_with_tags(
cache_key,
product.to_dict(),
tags,
timeout=3600
)
# 카테고리 변경시 관련 상품 캐시 모두 무효화
def invalidate_category_caches(category_id):
TaggedCache.invalidate_tag(f'category:{category_id}')
🗄️ 세션 저장소로 Redis 활용
Redis 세션 백엔드 설정
# settings.py
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'
SESSION_COOKIE_AGE = 1209600 # 2주
SESSION_SAVE_EVERY_REQUEST = False # 성능 최적화
SESSION_EXPIRE_AT_BROWSER_CLOSE = False
# 보안 설정
SESSION_COOKIE_SECURE = True # HTTPS에서만 전송
SESSION_COOKIE_HTTPONLY = True # XSS 방지
SESSION_COOKIE_SAMESITE = 'Lax' # CSRF 방지
커스텀 세션 관리
from django.contrib.sessions.models import Session
from django.contrib.auth import get_user_model
from django.core.cache import cache
import json
User = get_user_model()
class SessionManager:
@staticmethod
def get_active_users():
"""현재 활성 사용자 목록 캐시"""
cache_key = 'active_users'
active_users = cache.get(cache_key)
if active_users is None:
# Redis에서 세션 데이터 조회
from django_redis import get_redis_connection
redis_conn = get_redis_connection("default")
# 세션 키 패턴으로 활성 세션 찾기
session_keys = redis_conn.keys(':1:django.contrib.sessions.cache*')
user_ids = set()
for session_key in session_keys:
try:
session_data = redis_conn.get(session_key)
if session_data:
session_dict = json.loads(session_data)
user_id = session_dict.get('_auth_user_id')
if user_id:
user_ids.add(int(user_id))
except (json.JSONDecodeError, ValueError):
continue
active_users = list(User.objects.filter(id__in=user_ids))
cache.set(cache_key, active_users, timeout=300) # 5분 캐시
return active_users
@staticmethod
def force_logout_user(user_id):
"""특정 사용자의 모든 세션 강제 종료"""
from django_redis import get_redis_connection
redis_conn = get_redis_connection("default")
# 해당 사용자의 모든 세션 찾아서 삭제
session_keys = redis_conn.keys(':1:django.contrib.sessions.cache*')
for session_key in session_keys:
try:
session_data = redis_conn.get(session_key)
if session_data:
session_dict = json.loads(session_data)
if session_dict.get('_auth_user_id') == str(user_id):
redis_conn.delete(session_key)
except (json.JSONDecodeError, ValueError):
continue
# 활성 사용자 캐시 무효화
cache.delete('active_users')
📊 캐시 성능 모니터링
캐시 히트율 측정
from django.core.cache import cache
from django.utils.decorators import method_decorator
from django.views.decorators.cache import never_cache
import time
import logging
logger = logging.getLogger(__name__)
class CacheMonitoringMixin:
"""캐시 성능 모니터링 믹스인"""
def get_with_stats(self, cache_key, callback, timeout=300):
"""캐시 히트율 통계와 함께 데이터 조회"""
start_time = time.time()
# 캐시에서 조회 시도
data = cache.get(cache_key)
cache_hit = data is not None
if not cache_hit:
# 캐시 미스 - 콜백으로 데이터 생성
data = callback()
cache.set(cache_key, data, timeout)
# 통계 기록
execution_time = time.time() - start_time
self._record_cache_stats(cache_key, cache_hit, execution_time)
return data
def _record_cache_stats(self, cache_key, cache_hit, execution_time):
"""캐시 통계 기록"""
stats_key = f'cache_stats:{cache_key}'
stats = cache.get(stats_key, {
'hits': 0,
'misses': 0,
'total_time': 0,
'count': 0
})
if cache_hit:
stats['hits'] += 1
else:
stats['misses'] += 1
stats['total_time'] += execution_time
stats['count'] += 1
stats['hit_ratio'] = stats['hits'] / stats['count']
stats['avg_time'] = stats['total_time'] / stats['count']
# 통계는 24시간 보존
cache.set(stats_key, stats, timeout=86400)
# 히트율이 낮으면 로깅
if stats['count'] > 10 and stats['hit_ratio'] < 0.5:
logger.warning(
f"Low cache hit ratio for {cache_key}: {stats['hit_ratio']:.2%}"
)
class ProductService(CacheMonitoringMixin):
def get_popular_products(self, limit=10):
def fetch_popular_products():
return Product.objects.filter(
is_active=True
).order_by('-view_count')[:limit]
return self.get_with_stats(
f'popular_products:{limit}',
fetch_popular_products,
timeout=3600
)
캐시 상태 대시보드
from django.http import JsonResponse
from django.views.generic import View
from django_redis import get_redis_connection
import json
class CacheStatsView(View):
"""캐시 상태 모니터링 API"""
def get(self, request):
redis_conn = get_redis_connection("default")
# Redis 서버 정보
redis_info = redis_conn.info()
# 캐시 통계 수집
cache_stats = {}
stats_keys = cache.keys('cache_stats:*')
for stats_key in stats_keys:
stats_data = cache.get(stats_key)
if stats_data:
cache_name = stats_key.replace('cache_stats:', '')
cache_stats[cache_name] = stats_data
# 전체 통계 계산
total_hits = sum(stats['hits'] for stats in cache_stats.values())
total_misses = sum(stats['misses'] for stats in cache_stats.values())
total_requests = total_hits + total_misses
overall_hit_ratio = total_hits / total_requests if total_requests > 0 else 0
return JsonResponse({
'redis_info': {
'used_memory': redis_info['used_memory_human'],
'connected_clients': redis_info['connected_clients'],
'total_commands_processed': redis_info['total_commands_processed'],
'keyspace_hits': redis_info.get('keyspace_hits', 0),
'keyspace_misses': redis_info.get('keyspace_misses', 0),
},
'cache_stats': cache_stats,
'overall_stats': {
'total_hits': total_hits,
'total_misses': total_misses,
'hit_ratio': overall_hit_ratio,
'total_requests': total_requests,
}
})
# 캐시 상태 명령행 도구
from django.core.management.base import BaseCommand
class Command(BaseCommand):
help = 'Display cache statistics'
def handle(self, *args, **options):
redis_conn = get_redis_connection("default")
info = redis_conn.info()
self.stdout.write(
self.style.SUCCESS('=== Redis Cache Statistics ===')
)
self.stdout.write(f"Used Memory: {info['used_memory_human']}")
self.stdout.write(f"Connected Clients: {info['connected_clients']}")
self.stdout.write(f"Total Commands: {info['total_commands_processed']}")
# 키 개수 표시
total_keys = 0
for db_info in info.get('db0', {}).values():
if isinstance(db_info, dict) and 'keys' in db_info:
total_keys += db_info['keys']
self.stdout.write(f"Total Keys: {total_keys}")
# 히트율 표시
hits = info.get('keyspace_hits', 0)
misses = info.get('keyspace_misses', 0)
if hits + misses > 0:
hit_ratio = hits / (hits + misses)
self.stdout.write(f"Hit Ratio: {hit_ratio:.2%}")
⚡ 고급 최적화 기법
캐시 워밍업 전략
from django.core.management.base import BaseCommand
from django.core.cache import cache
from concurrent.futures import ThreadPoolExecutor
import time
class CacheWarmer:
"""캐시 사전 로딩 클래스"""
def __init__(self, max_workers=5):
self.max_workers = max_workers
def warm_product_caches(self):
"""상품 관련 캐시 사전 로딩"""
def warm_category_cache(category_id):
"""카테고리별 상품 캐시"""
cache_key = f'category_products:{category_id}'
if not cache.get(cache_key):
products = Product.objects.filter(
category_id=category_id,
is_active=True
).select_related('category')[:20]
cache.set(cache_key, list(products.values()), timeout=3600)
return f"Warmed cache for category {category_id}"
# 모든 활성 카테고리 ID 조회
category_ids = Category.objects.filter(
is_active=True
).values_list('id', flat=True)
# 병렬로 캐시 워밍업
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = [
executor.submit(warm_category_cache, cat_id)
for cat_id in category_ids
]
for future in futures:
try:
result = future.result(timeout=30)
print(result)
except Exception as e:
print(f"Cache warming failed: {e}")
def warm_user_caches(self):
"""사용자 관련 캐시 사전 로딩"""
def warm_user_recommendations(user_id):
cache_key = f'user_recommendations:{user_id}'
if not cache.get(cache_key):
# 사용자 추천 계산 (시간이 오래 걸리는 작업)
recommendations = calculate_user_recommendations(user_id)
cache.set(cache_key, recommendations, timeout=7200)
# 최근 활성 사용자들에 대해 캐시 워밍업
recent_users = User.objects.filter(
last_login__gte=timezone.now() - timedelta(days=7)
).values_list('id', flat=True)[:100] # 최근 100명만
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
for user_id in recent_users:
executor.submit(warm_user_recommendations, user_id)
# 관리 명령어
class Command(BaseCommand):
help = 'Warm up caches with frequently accessed data'
def add_arguments(self, parser):
parser.add_argument(
'--type',
choices=['products', 'users', 'all'],
default='all',
help='Type of cache to warm up'
)
def handle(self, *args, **options):
warmer = CacheWarmer()
start_time = time.time()
if options['type'] in ['products', 'all']:
self.stdout.write('Warming product caches...')
warmer.warm_product_caches()
if options['type'] in ['users', 'all']:
self.stdout.write('Warming user caches...')
warmer.warm_user_caches()
elapsed = time.time() - start_time
self.stdout.write(
self.style.SUCCESS(
f'Cache warming completed in {elapsed:.2f} seconds'
)
)
분산 캐시 락 구현
import time
import uuid
from django_redis import get_redis_connection
class DistributedLock:
"""Redis를 이용한 분산 락"""
def __init__(self, lock_name, timeout=10, retry_delay=0.1):
self.lock_name = f'lock:{lock_name}'
self.timeout = timeout
self.retry_delay = retry_delay
self.redis_conn = get_redis_connection('default')
self.lock_value = str(uuid.uuid4())
def acquire(self):
"""락 획득"""
end_time = time.time() + self.timeout
while time.time() < end_time:
# SET 명령어로 락 설정 (NX: key가 없을 때만, EX: 만료 시간)
if self.redis_conn.set(
self.lock_name,
self.lock_value,
nx=True,
ex=self.timeout
):
return True
time.sleep(self.retry_delay)
return False
def release(self):
"""락 해제"""
# Lua 스크립트로 원자적 해제 (자신이 설정한 락만 해제)
lua_script = """
if redis.call("GET", KEYS[1]) == ARGV[1] then
return redis.call("DEL", KEYS[1])
else
return 0
end
"""
return self.redis_conn.eval(lua_script, 1, self.lock_name, self.lock_value)
def __enter__(self):
if not self.acquire():
raise Exception(f"Failed to acquire lock: {self.lock_name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
# 사용 예시: 중복 계산 방지
def get_expensive_calculation(param):
cache_key = f'expensive_calc:{param}'
result = cache.get(cache_key)
if result is None:
# 분산 락으로 중복 계산 방지
with DistributedLock(f'calc_lock:{param}', timeout=30):
# 락 획득 후 다시 한 번 캐시 확인
result = cache.get(cache_key)
if result is None:
# 실제 계산 수행
result = perform_expensive_calculation(param)
cache.set(cache_key, result, timeout=3600)
return result
캐시 압축 및 직렬화 최적화
import pickle
import zlib
import json
from django.core.cache.backends.base import BaseCache
class OptimizedRedisCache:
"""최적화된 Redis 캐시 래퍼"""
def __init__(self):
self.redis_conn = get_redis_connection('default')
def _serialize_and_compress(self, data):
"""데이터 직렬화 및 압축"""
# JSON으로 직렬화 시도 (빠름)
try:
serialized = json.dumps(data).encode('utf-8')
compression_type = 'json'
except (TypeError, ValueError):
# JSON이 안되면 pickle 사용 (느리지만 모든 객체 지원)
serialized = pickle.dumps(data)
compression_type = 'pickle'
# 데이터가 크면 압축
if len(serialized) > 1024: # 1KB 이상이면 압축
compressed = zlib.compress(serialized)
if len(compressed) < len(serialized) * 0.9: # 10% 이상 압축되면 사용
return compressed, f'{compression_type}+zlib'
return serialized, compression_type
def _decompress_and_deserialize(self, data, metadata):
"""데이터 압축 해제 및 역직렬화"""
compression_type = metadata
# 압축 해제
if '+zlib' in compression_type:
data = zlib.decompress(data)
compression_type = compression_type.replace('+zlib', '')
# 역직렬화
if compression_type == 'json':
return json.loads(data.decode('utf-8'))
elif compression_type == 'pickle':
return pickle.loads(data)
else:
raise ValueError(f"Unknown compression type: {compression_type}")
def set(self, key, value, timeout=300):
"""최적화된 캐시 설정"""
data, compression_type = self._serialize_and_compress(value)
# 메타데이터와 함께 저장
cache_data = {
'data': data,
'metadata': compression_type,
'timestamp': time.time()
}
serialized_cache_data = pickle.dumps(cache_data)
self.redis_conn.setex(key, timeout, serialized_cache_data)
def get(self, key, default=None):
"""최적화된 캐시 조회"""
try:
cache_data = self.redis_conn.get(key)
if cache_data is None:
return default
cache_dict = pickle.loads(cache_data)
return self._decompress_and_deserialize(
cache_dict['data'],
cache_dict['metadata']
)
except Exception:
return default
# 사용 예시
optimized_cache = OptimizedRedisCache()
def cache_large_dataset(data_id):
cache_key = f'large_dataset:{data_id}'
data = optimized_cache.get(cache_key)
if data is None:
# 대용량 데이터 생성
data = generate_large_dataset(data_id)
optimized_cache.set(cache_key, data, timeout=3600)
return data
🔧 운영 환경 최적화
Redis 클러스터 설정
# settings/production.py
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': [
'redis://redis-node1:6379/0',
'redis://redis-node2:6379/0',
'redis://redis-node3:6379/0',
],
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.ShardClient', # 샤딩 클라이언트
'CONNECTION_POOL_KWARGS': {
'max_connections': 50,
'retry_on_timeout': True,
'health_check_interval': 30,
},
'SERIALIZER': 'django_redis.serializers.msgpack.MSGPackSerializer',
'COMPRESSOR': 'django_redis.compressors.lz4.Lz4Compressor',
}
}
}
# Redis Sentinel 설정 (고가용성)
CACHES = {
'default': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://mymaster/0', # Sentinel 마스터 이름
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.SentinelClient',
'CONNECTION_POOL_KWARGS': {
'service_name': 'mymaster',
'sentinels': [
('sentinel1', 26379),
('sentinel2', 26379),
('sentinel3', 26379),
],
},
}
}
}
캐시 계층화 전략
from django.core.cache import caches
class TieredCache:
"""다단계 캐시 구현"""
def __init__(self):
self.l1_cache = caches['l1'] # 로컬 메모리 캐시
self.l2_cache = caches['l2'] # Redis 캐시
def get(self, key, default=None):
"""L1 -> L2 순서로 캐시 조회"""
# L1 캐시에서 먼저 조회
value = self.l1_cache.get(key)
if value is not None:
return value
# L2 캐시에서 조회
value = self.l2_cache.get(key)
if value is not None:
# L1 캐시에 복사 (작은 TTL)
self.l1_cache.set(key, value, timeout=60)
return value
return default
def set(self, key, value, timeout=300):
"""두 레벨 모두에 캐시 설정"""
self.l2_cache.set(key, value, timeout)
# L1은 짧은 시간만 유지
self.l1_cache.set(key, value, timeout=min(60, timeout))
def delete(self, key):
"""두 레벨 모두에서 삭제"""
self.l1_cache.delete(key)
self.l2_cache.delete(key)
# settings.py에서 다중 캐시 설정
CACHES = {
'l1': {
'BACKEND': 'django.core.cache.backends.locmem.LocMemCache',
'LOCATION': 'unique-snowflake',
'OPTIONS': {
'MAX_ENTRIES': 1000,
'CULL_FREQUENCY': 3,
}
},
'l2': {
'BACKEND': 'django_redis.cache.RedisCache',
'LOCATION': 'redis://127.0.0.1:6379/1',
'OPTIONS': {
'CLIENT_CLASS': 'django_redis.client.DefaultClient',
}
}
}
캐시 백업 및 복구
from django.core.management.base import BaseCommand
from django_redis import get_redis_connection
import json
import gzip
class Command(BaseCommand):
help = 'Backup and restore Redis cache data'
def add_arguments(self, parser):
parser.add_argument('action', choices=['backup', 'restore'])
parser.add_argument('--file', required=True, help='Backup file path')
parser.add_argument('--pattern', default='*', help='Key pattern to backup')
def handle(self, *args, **options):
redis_conn = get_redis_connection('default')
if options['action'] == 'backup':
self.backup_cache(redis_conn, options['file'], options['pattern'])
else:
self.restore_cache(redis_conn, options['file'])
def backup_cache(self, redis_conn, file_path, pattern):
"""캐시 데이터 백업"""
keys = redis_conn.keys(pattern)
backup_data = {}
self.stdout.write(f'Backing up {len(keys)} keys...')
for key in keys:
try:
# 키 타입에 따라 다르게 처리
key_type = redis_conn.type(key)
ttl = redis_conn.ttl(key)
if key_type == b'string':
value = redis_conn.get(key)
elif key_type == b'hash':
value = redis_conn.hgetall(key)
elif key_type == b'list':
value = redis_conn.lrange(key, 0, -1)
elif key_type == b'set':
value = list(redis_conn.smembers(key))
elif key_type == b'zset':
value = redis_conn.zrange(key, 0, -1, withscores=True)
else:
continue
backup_data[key.decode()] = {
'type': key_type.decode(),
'value': value,
'ttl': ttl if ttl > 0 else None
}
except Exception as e:
self.stdout.write(f'Error backing up key {key}: {e}')
# 압축하여 저장
with gzip.open(file_path, 'wt', encoding='utf-8') as f:
json.dump(backup_data, f, default=str)
self.stdout.write(
self.style.SUCCESS(f'Backup completed: {file_path}')
)
def restore_cache(self, redis_conn, file_path):
"""캐시 데이터 복구"""
try:
with gzip.open(file_path, 'rt', encoding='utf-8') as f:
backup_data = json.load(f)
except FileNotFoundError:
self.stdout.write(
self.style.ERROR(f'Backup file not found: {file_path}')
)
return
self.stdout.write(f'Restoring {len(backup_data)} keys...')
for key, data in backup_data.items():
try:
key_type = data['type']
value = data['value']
ttl = data['ttl']
if key_type == 'string':
redis_conn.set(key, value)
elif key_type == 'hash':
redis_conn.hmset(key, value)
elif key_type == 'list':
redis_conn.delete(key)
redis_conn.lpush(key, *value)
elif key_type == 'set':
redis_conn.delete(key)
redis_conn.sadd(key, *value)
elif key_type == 'zset':
redis_conn.delete(key)
for member, score in value:
redis_conn.zadd(key, {member: score})
# TTL 설정
if ttl:
redis_conn.expire(key, ttl)
except Exception as e:
self.stdout.write(f'Error restoring key {key}: {e}')
self.stdout.write(
self.style.SUCCESS('Restore completed')
)
🎯 마무리: Redis 캐시 베스트 프랙티스
✅ 구현 체크리스트
기본 설정:
- Django-Redis 패키지 설치 및 설정
- 적절한 직렬화 방식 선택 (JSON vs MessagePack vs Pickle)
- 압축 설정으로 메모리 사용량 최적화
- 연결 풀 크기 적절히 설정
캐시 전략:
- 캐시 레벨별 적용 범위 결정
- 적절한 TTL 설정 (데이터 특성에 따라)
- 캐시 키 네이밍 컨벤션 수립
- 캐시 무효화 전략 구현
성능 최적화:
- 캐시 히트율 모니터링 구현
- 캐시 워밍업 전략 수립
- 분산 락으로 중복 계산 방지
- 계층화된 캐시 구조 고려
운영 환경:
- Redis 클러스터/센티넬 설정 (고가용성)
- 캐시 백업 및 복구 계획 수립
- 모니터링 및 알림 시스템 구축
- 장애 대응 매뉴얼 작성
⚡ 성능 향상 팁
- 적절한 TTL 설정: 데이터 업데이트 빈도에 맞춰 설정
- 캐시 키 설계: 충돌을 피하고 의미 있는 네이밍
- 압축 활용: 큰 데이터는 압축으로 메모리 절약
- 배치 처리: 여러 키를 한 번에 처리하는
get_many,set_many활용 - 모니터링: 정기적인 성능 측정 및 최적화
🔒 주의사항
- 캐시 일관성: 데이터 변경 시 반드시 관련 캐시 무효화
- 메모리 관리: Redis 메모리 사용량 지속적 모니터링
- 장애 대응: 캐시 장애 시에도 서비스가 동작하도록 설계
- 보안: 중요한 데이터는 캐시하지 않거나 암호화
Redis를 활용한 Django 캐시 시스템은 웹 애플리케이션 성능을 극적으로 향상시킬 수 있는 강력한 도구입니다. 이 가이드를 통해 여러분의 Django 애플리케이션이 더욱 빠르고 효율적으로 동작하기를 바랍니다.
📚 추가 학습 자료
💡 팁: 캐시 구현 후에는 반드시 부하 테스트를 통해 실제 성능 향상을 측정해보세요!