🎪 인메모리 캐싱, 도대체 뭐길래?

🎯 Redis vs Memcached: 누가 더 짱일까?

🔴 Redis: 만능 스위스 아미 나이프

⚪ Memcached: 심플 이즈 베스트

🎮 Redis 실전 활용법

🔧 Redis 설치 및 기본 설정

sudo apt update
sudo apt install redis-server
sudo systemctl start redis-server
sudo systemctl enable redis-server

brew install redis
brew services start redis

docker run --name my-redis -p 6379:6379 -d redis

💻 Python으로 Redis 사용하기

pip install redis

import redis

# Redis 연결
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

# 기본 SET/GET
r.set('username', 'coding_master')
username = r.get('username')
print(username)  # 'coding_master'

# 만료 시간 설정 (초 단위)
r.setex('session_token', 3600, 'abc123xyz')  # 1시간 후 자동 삭제

# 숫자 증가/감소
r.set('page_views', 0)
r.incr('page_views')  # 1 증가
r.incrby('page_views', 10)  # 10 증가
print(r.get('page_views'))  # '11'

# 키 존재 확인
if r.exists('username'):
    print('키가 존재해요!')

# 키 삭제
r.delete('username')

📝 List 자료구조 활용

# 리스트에 요소 추가
r.lpush('recent_searches', '파이썬 강의')  # 왼쪽에 추가
r.lpush('recent_searches', 'Redis 튜토리얼')
r.rpush('recent_searches', '데이터베이스 최적화')  # 오른쪽에 추가

# 리스트 조회
searches = r.lrange('recent_searches', 0, -1)  # 전체 조회
print(searches)  # ['Redis 튜토리얼', '파이썬 강의', '데이터베이스 최적화']

# 리스트 길이
length = r.llen('recent_searches')

# 요소 제거
r.lpop('recent_searches')  # 왼쪽에서 제거
r.rpop('recent_searches')  # 오른쪽에서 제거

# 최근 5개만 유지 (나머지 삭제)
r.ltrim('recent_searches', 0, 4)

🎲 Set 자료구조 활용

# Set에 요소 추가
r.sadd('user:1000:likes', 'post:1', 'post:5', 'post:10')
r.sadd('user:2000:likes', 'post:5', 'post:15', 'post:20')

# Set 조회
likes = r.smembers('user:1000:likes')
print(likes)  # {'post:1', 'post:5', 'post:10'}

# 요소 존재 확인
if r.sismember('user:1000:likes', 'post:5'):
    print('이미 좋아요를 눌렀어요!')

# Set 크기
count = r.scard('user:1000:likes')

# 교집합 (두 사용자가 공통으로 좋아하는 게시물)
common = r.sinter('user:1000:likes', 'user:2000:likes')
print(common)  # {'post:5'}

# 합집합
all_likes = r.sunion('user:1000:likes', 'user:2000:likes')

# 차집합 (user:1000만 좋아하는 게시물)
unique = r.sdiff('user:1000:likes', 'user:2000:likes')

# 요소 제거
r.srem('user:1000:likes', 'post:1')

🗂️ Hash 자료구조 활용

# Hash 설정
r.hset('user:1000', mapping={
    'name': '김코딩',
    'email': 'coding@example.com',
    'age': '28',
    'level': 'expert'
})

# 개별 필드 설정
r.hset('user:1000', 'points', '1500')

# 필드 조회
name = r.hget('user:1000', 'name')
print(name)  # '김코딩'

# 전체 조회
user_data = r.hgetall('user:1000')
print(user_data)
# {'name': '김코딩', 'email': 'coding@example.com', 'age': '28', 'level': 'expert', 'points': '1500'}

# 숫자 필드 증가
r.hincrby('user:1000', 'points', 100)  # 포인트 100 증가

# 필드 존재 확인
if r.hexists('user:1000', 'email'):
    print('이메일이 등록되어 있어요!')

# 필드 삭제
r.hdel('user:1000', 'age')

# 모든 필드 이름 조회
fields = r.hkeys('user:1000')

# 모든 값 조회
values = r.hvals('user:1000')

🏆 Sorted Set 활용 (랭킹 시스템)

# 점수와 함께 추가
r.zadd('game_leaderboard', {
    'player1': 1500,
    'player2': 2300,
    'player3': 1800,
    'player4': 2100,
    'player5': 1950
})

# 점수 증가
r.zincrby('game_leaderboard', 200, 'player1')  # player1 점수 200 증가

# 상위 3명 조회 (점수 높은 순)
top3 = r.zrevrange('game_leaderboard', 0, 2, withscores=True)
print(top3)
# [('player2', 2300.0), ('player4', 2100.0), ('player5', 1950.0)]

# 하위 3명 조회 (점수 낮은 순)
bottom3 = r.zrange('game_leaderboard', 0, 2, withscores=True)

# 특정 플레이어 순위 조회 (0부터 시작)
rank = r.zrevrank('game_leaderboard', 'player1')
print(f'player1의 순위: {rank + 1}위')

# 특정 플레이어 점수 조회
score = r.zscore('game_leaderboard', 'player1')

# 점수 범위로 조회 (1800~2000점 사이)
mid_range = r.zrangebyscore('game_leaderboard', 1800, 2000, withscores=True)

# 전체 플레이어 수
total = r.zcard('game_leaderboard')

# 플레이어 제거
r.zrem('game_leaderboard', 'player3')

🔐 세션 관리의 모든 것

🤔 세션이 뭐예요?

🏪 세션 저장소 선택하기

🔨 Flask에서 Redis 세션 구현하기

from flask import Flask, session, redirect, url_for, request
from flask_session import Session
import redis

app = Flask(__name__)

# Flask 세션 설정
app.config['SECRET_KEY'] = 'your-secret-key-here'
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_PERMANENT'] = False
app.config['SESSION_USE_SIGNER'] = True
app.config['SESSION_KEY_PREFIX'] = 'session:'
app.config['SESSION_REDIS'] = redis.Redis(
    host='localhost',
    port=6379,
    db=0
)

Session(app)

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    
    # 실제로는 DB에서 사용자 확인
    if username == 'admin' and password == 'password123':
        # 세션에 사용자 정보 저장
        session['user_id'] = 1000
        session['username'] = username
        session['is_admin'] = True
        session.permanent = True  # 영구 세션 (설정된 시간까지)
        
        return redirect(url_for('dashboard'))
    
    return '로그인 실패!', 401

@app.route('/dashboard')
def dashboard():
    # 세션에서 사용자 정보 조회
    if 'user_id' not in session:
        return redirect(url_for('login'))
    
    username = session.get('username')
    return f'환영합니다, {username}님!'

@app.route('/logout')
def logout():
    # 세션 삭제
    session.clear()
    return redirect(url_for('login'))

if __name__ == '__main__':
    app.run(debug=True)

🎯 Django에서 Redis 세션 구현하기

# settings.py

# Redis 세션 백엔드 설정
SESSION_ENGINE = 'django.contrib.sessions.backends.cache'
SESSION_CACHE_ALIAS = 'default'

# Redis 캐시 설정
CACHES = {
    'default': {
        'BACKEND': 'django_redis.cache.RedisCache',
        'LOCATION': 'redis://127.0.0.1:6379/1',
        'OPTIONS': {
            'CLIENT_CLASS': 'django_redis.client.DefaultClient',
            'PASSWORD': 'your-redis-password',  # 비밀번호가 있다면
        },
        'KEY_PREFIX': 'myapp',
        'TIMEOUT': 3600,  # 1시간
    }
}

# 세션 설정
SESSION_COOKIE_AGE = 3600  # 1시간
SESSION_COOKIE_SECURE = True  # HTTPS에서만 전송
SESSION_COOKIE_HTTPONLY = True  # JavaScript 접근 차단
SESSION_COOKIE_SAMESITE = 'Lax'  # CSRF 방어

from django.shortcuts import render, redirect
from django.contrib.auth import authenticate, login, logout
from django.contrib.auth.decorators import login_required

def user_login(request):
    if request.method == 'POST':
        username = request.POST.get('username')
        password = request.POST.get('password')
        
        user = authenticate(request, username=username, password=password)
        
        if user is not None:
            login(request, user)
            
            # 세션에 추가 정보 저장
            request.session['last_login_ip'] = request.META.get('REMOTE_ADDR')
            request.session['login_count'] = request.session.get('login_count', 0) + 1
            
            return redirect('dashboard')
        else:
            return render(request, 'login.html', {'error': '로그인 실패'})
    
    return render(request, 'login.html')

@login_required
def dashboard(request):
    # 세션 정보 조회
    login_count = request.session.get('login_count', 0)
    last_ip = request.session.get('last_login_ip', 'Unknown')
    
    context = {
        'username': request.user.username,
        'login_count': login_count,
        'last_ip': last_ip,
    }
    
    return render(request, 'dashboard.html', context)

def user_logout(request):
    logout(request)
    return redirect('login')

🛡️ 세션 보안 베스트 프랙티스

⚡ 세션 성능 최적화

import json
import zlib
import redis

r = redis.Redis()

def save_compressed_session(session_id, data):
    # JSON 직렬화 후 압축
    json_data = json.dumps(data)
    compressed = zlib.compress(json_data.encode('utf-8'))
    r.setex(f'session:{session_id}', 3600, compressed)

def load_compressed_session(session_id):
    # 압축 해제 후 역직렬화
    compressed = r.get(f'session:{session_id}')
    if compressed:
        json_data = zlib.decompress(compressed).decode('utf-8')
        return json.loads(json_data)
    return None

import redis

# 연결 풀 생성
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=50,
    decode_responses=True
)

# 연결 풀 사용
r = redis.Redis(connection_pool=pool)

# 비효율적인 방법
r.set('key1', 'value1')
r.set('key2', 'value2')
r.set('key3', 'value3')

# 효율적인 방법 (파이프라이닝)
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
pipe.execute()  # 한 번에 실행!

import time
from datetime import datetime, timedelta

def cleanup_old_sessions():
    """30일 이상 된 세션 정리"""
    cursor = 0
    cutoff_time = (datetime.now() - timedelta(days=30)).timestamp()
    
    while True:
        cursor, keys = r.scan(cursor, match='session:*', count=100)
        
        for key in keys:
            # 세션 생성 시간 확인
            session_data = r.hgetall(key)
            if session_data:
                created_at = float(session_data.get('created_at', 0))
                if created_at < cutoff_time:
                    r.delete(key)
                    print(f'Deleted old session: {key}')
        
        if cursor == 0:
            break
        
        time.sleep(0.1)  # CPU 부하 방지

🎨 고급 활용 패턴

🔄 캐시 무효화 전략

# 10분 후 자동 만료
r.setex('user:1000:profile', 600, json.dumps(user_data))

# 또는
r.set('product:500', json.dumps(product_data), ex=3600)  # 1시간

def update_user_profile(user_id, new_data):
    # 1. DB 업데이트
    db.update_user(user_id, new_data)
    
    # 2. 캐시 업데이트
    cache_key = f'user:{user_id}:profile'
    r.setex(cache_key, 3600, json.dumps(new_data))
    
    return True

def get_user_profile(user_id):
    cache_key = f'user:{user_id}:profile'
    
    # 1. 캐시 확인
    cached_data = r.get(cache_key)
    if cached_data:
        return json.loads(cached_data)
    
    # 2. 캐시 미스 - DB 조회
    user_data = db.get_user(user_id)
    
    # 3. 캐시에 저장
    if user_data:
        r.setex(cache_key, 3600, json.dumps(user_data))
    
    return user_data

def update_user_profile(user_id, new_data):
    # 1. DB 업데이트
    db.update_user(user_id, new_data)
    
    # 2. 캐시 무효화
    cache_key = f'user:{user_id}:profile'
    r.delete(cache_key)
    
    return True

def cache_with_tags(key, data, tags, ttl=3600):
    """태그와 함께 캐시 저장"""
    # 데이터 저장
    r.setex(key, ttl, json.dumps(data))
    
    # 각 태그에 키 추가
    for tag in tags:
        r.sadd(f'tag:{tag}', key)
        r.expire(f'tag:{tag}', ttl)

def invalidate_by_tag(tag):
    """특정 태그의 모든 캐시 무효화"""
    tag_key = f'tag:{tag}'
    keys = r.smembers(tag_key)
    
    if keys:
        r.delete(*keys)  # 모든 키 삭제
        r.delete(tag_key)  # 태그도 삭제
    
    return len(keys)

# 사용 예시
user_data = {'id': 1000, 'name': '김코딩'}
cache_with_tags(
    'user:1000:profile',
    user_data,
    tags=['user:1000', 'users', 'profiles'],
    ttl=3600
)

# 사용자 1000의 모든 캐시 무효화
invalidate_by_tag('user:1000')

🔒 분산 락(Distributed Lock)

import time
import uuid

class RedisLock:
    def __init__(self, redis_client, lock_name, timeout=10):
        self.redis = redis_client
        self.lock_name = f'lock:{lock_name}'
        self.timeout = timeout
        self.identifier = str(uuid.uuid4())
    
    def acquire(self, blocking=True, timeout=None):
        """락 획득"""
        end_time = time.time() + (timeout or self.timeout)
        
        while True:
            # SET NX (존재하지 않을 때만 설정)
            if self.redis.set(self.lock_name, self.identifier, 
                            nx=True, ex=self.timeout):
                return True
            
            if not blocking:
                return False
            
            if time.time() > end_time:
                return False
            
            time.sleep(0.001)  # 1ms 대기
    
    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
        """
        self.redis.eval(lua_script, 1, self.lock_name, self.identifier)
    
    def __enter__(self):
        self.acquire()
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        self.release()

# 사용 예시
def process_payment(order_id):
    lock = RedisLock(r, f'order:{order_id}')
    
    with lock:
        # 이 블록 안에서는 다른 서버가 접근할 수 없어요!
        order = db.get_order(order_id)
        
        if order.status == 'pending':
            # 결제 처리
            payment_result = process_payment_gateway(order)
            
            if payment_result.success:
                order.status = 'paid'
                db.save_order(order)
                
                # 캐시 무효화
                r.delete(f'order:{order_id}')
        
        return order

📊 Rate Limiting (속도 제한)

def rate_limit_fixed_window(user_id, limit=100, window=60):
    """
    고정 윈도우 방식
    - limit: 허용 횟수
    - window: 시간 윈도우 (초)
    """
    key = f'rate_limit:{user_id}:{int(time.time() // window)}'
    
    current = r.incr(key)
    
    if current == 1:
        r.expire(key, window)
    
    if current > limit:
        return False, 0
    
    return True, limit - current

def rate_limit_sliding_window(user_id, limit=100, window=60):
    """
    슬라이딩 윈도우 방식 (더 정확함)
    """
    key = f'rate_limit:{user_id}'
    now = time.time()
    
    # 파이프라인 사용
    pipe = r.pipeline()
    
    # 오래된 요청 제거
    pipe.zremrangebyscore(key, 0, now - window)
    
    # 현재 요청 추가
    pipe.zadd(key, {str(now): now})
    
    # 현재 요청 수 조회
    pipe.zcard(key)
    
    # 만료 시간 설정
    pipe.expire(key, window)
    
    results = pipe.execute()
    request_count = results[2]
    
    if request_count > limit:
        return False, 0
    
    return True, limit - request_count

# 사용 예시
@app.route('/api/data')
def get_data():
    user_id = session.get('user_id')
    
    allowed, remaining = rate_limit_sliding_window(user_id, limit=100, window=60)
    
    if not allowed:
        return {
            'error': 'Rate limit exceeded',
            'retry_after': 60
        }, 429
    
    # 정상 처리
    return {
        'data': 'some data',
        'rate_limit_remaining': remaining
    }

🎯 Pub/Sub으로 실시간 알림

# publisher.py (알림 발송)
import redis
import json

r = redis.Redis()

def send_notification(user_id, message):
    """특정 사용자에게 알림 발송"""
    channel = f'notifications:user:{user_id}'
    
    notification = {
        'type': 'message',
        'title': '새 메시지',
        'body': message,
        'timestamp': time.time()
    }
    
    r.publish(channel, json.dumps(notification))

def broadcast_notification(message):
    """모든 사용자에게 알림 발송"""
    notification = {
        'type': 'announcement',
        'title': '공지사항',
        'body': message,
        'timestamp': time.time()
    }
    
    r.publish('notifications:broadcast', json.dumps(notification))

# subscriber.py (알림 수신)
import redis
import json

r = redis.Redis()

def listen_notifications(user_id):
    """특정 사용자의 알림 수신"""
    pubsub = r.pubsub()
    
    # 개인 채널과 브로드캐스트 채널 구독
    pubsub.subscribe(
        f'notifications:user:{user_id}',
        'notifications:broadcast'
    )
    
    print(f'사용자 {user_id}의 알림 수신 시작...')
    
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            print(f"[{data['type']}] {data['title']}: {data['body']}")

# 웹소켓과 연동
from flask import Flask
from flask_socketio import SocketIO, emit
import threading

app = Flask(__name__)
socketio = SocketIO(app)

def redis_listener():
    """Redis Pub/Sub 리스너 (백그라운드 스레드)"""
    pubsub = r.pubsub()
    pubsub.subscribe('notifications:broadcast')
    
    for message in pubsub.listen():
        if message['type'] == 'message':
            data = json.loads(message['data'])
            # 모든 웹소켓 클라이언트에게 전송
            socketio.emit('notification', data, broadcast=True)

@socketio.on('connect')
def handle_connect():
    user_id = session.get('user_id')
    print(f'사용자 {user_id} 연결됨')

@socketio.on('disconnect')
def handle_disconnect():
    print('사용자 연결 해제됨')

if __name__ == '__main__':
    # 백그라운드 리스너 시작
    listener_thread = threading.Thread(target=redis_listener, daemon=True)
    listener_thread.start()
    
    socketio.run(app, debug=True)

🏗️ Redis 클러스터와 고가용성

🔄 Master-Slave 복제

port 6379
bind 0.0.0.0
requirepass master_password
masterauth master_password

# 영속성 설정
save 900 1
save 300 10
save 60 10000

appendonly yes
appendfilename "appendonly.aof"

port 6380
bind 0.0.0.0
requirepass slave_password

# Master 정보
replicaof 192.168.1.100 6379
masterauth master_password

# 읽기 전용
replica-read-only yes

🛡️ Redis Sentinel (자동 장애 조치)

# sentinel.conf
port 26379
sentinel monitor mymaster 192.168.1.100 6379 2
sentinel auth-pass mymaster master_password
sentinel down-after-milliseconds mymaster 5000
sentinel parallel-syncs mymaster 1
sentinel failover-timeout mymaster 10000

from redis.sentinel import Sentinel

# Sentinel 연결
sentinel = Sentinel([
    ('192.168.1.101', 26379),
    ('192.168.1.102', 26379),
    ('192.168.1.103', 26379)
], socket_timeout=0.1)

# Master 연결 (쓰기)
master = sentinel.master_for('mymaster', 
                            socket_timeout=0.1,
                            password='master_password')

# Slave 연결 (읽기)
slave = sentinel.slave_for('mymaster',
                          socket_timeout=0.1,
                          password='master_password')

# 사용
master.set('key', 'value')  # Master에 쓰기
value = slave.get('key')    # Slave에서 읽기

🌐 Redis Cluster (샤딩)

# Python에서 Redis Cluster 사용
from redis.cluster import RedisCluster

# 클러스터 연결
rc = RedisCluster(
    startup_nodes=[
        {"host": "192.168.1.101", "port": "7000"},
        {"host": "192.168.1.102", "port": "7001"},
        {"host": "192.168.1.103", "port": "7002"},
    ],
    decode_responses=True,
    skip_full_coverage_check=True
)

# 일반 Redis처럼 사용
rc.set('key', 'value')
value = rc.get('key')

# 클러스터 정보 조회
cluster_info = rc.cluster_info()
print(cluster_info)

🔍 모니터링과 디버깅

📊 주요 모니터링 지표

# Python으로 모니터링
import redis
import time

r = redis.Redis()

def monitor_redis():
    """Redis 상태 모니터링"""
    info = r.info()
    
    # 메모리 정보
    used_memory_mb = info['used_memory'] / 1024 / 1024
    max_memory_mb = info.get('maxmemory', 0) / 1024 / 1024
    memory_usage = (used_memory_mb / max_memory_mb * 100) if max_memory_mb > 0 else 0
    
    print(f"메모리 사용: {used_memory_mb:.2f}MB / {max_memory_mb:.2f}MB ({memory_usage:.1f}%)")
    
    # 성능 정보
    ops_per_sec = info['instantaneous_ops_per_sec']
    print(f"초당 명령어: {ops_per_sec}")
    
    # 캐시 히트율
    keyspace_hits = info.get('keyspace_hits', 0)
    keyspace_misses = info.get('keyspace_misses', 0)
    total = keyspace_hits + keyspace_misses
    hit_rate = (keyspace_hits / total * 100) if total > 0 else 0
    print(f"캐시 히트율: {hit_rate:.2f}%")
    
    # 연결 정보
    connected_clients = info['connected_clients']
    print(f"연결된 클라이언트: {connected_clients}")
    
    # 경고 체크
    if memory_usage > 80:
        print("⚠️ 경고: 메모리 사용량이 80%를 초과했어요!")
    
    if hit_rate < 80:
        print("⚠️ 경고: 캐시 히트율이 낮아요!")
    
    if connected_clients > 1000:
        print("⚠️ 경고: 연결 수가 많아요!")

# 주기적 모니터링
while True:
    monitor_redis()
    print("-" * 50)
    time.sleep(10)

🐛 디버깅 명령어

# Redis CLI에서 실행

# 1. 느린 쿼리 확인
SLOWLOG GET 10

# 2. 현재 실행 중인 명령어 확인
CLIENT LIST

# 3. 메모리 사용량 분석
MEMORY STATS
MEMORY DOCTOR

# 4. 키 분석
SCAN 0 MATCH user:* COUNT 100
DBSIZE

# 5. 특정 키의 메모리 사용량
MEMORY USAGE user:1000:profile

# 6. 키 만료 시간 확인
TTL session:abc123

# 7. 키 타입 확인
TYPE user:1000:likes

# 8. 실시간 모니터링
MONITOR  # 주의: 프로덕션에서는 성능 영향 있음!

🛠️ 일반적인 문제와 해결법

# redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru  # LRU 방식으로 자동 삭제

💰 비용 최적화 전략

# 메모리 효율적인 설정
# redis.conf

# Hash 최적화
hash-max-ziplist-entries 512
hash-max-ziplist-value 64

# List 최적화
list-max-ziplist-size -2

# Set 최적화
set-max-intset-entries 512

# Sorted Set 최적화
zset-max-ziplist-entries 128
zset-max-ziplist-value 64

🎓 실전 프로젝트: 완전한 세션 관리 시스템

# session_manager.py
import redis
import json
import hashlib
import secrets
from datetime import datetime, timedelta

class SessionManager:
    def __init__(self, redis_client, session_timeout=3600):
        self.redis = redis_client
        self.session_timeout = session_timeout
        self.session_prefix = 'session:'
    
    def create_session(self, user_id, user_data=None):
        """새 세션 생성"""
        # 안전한 세션 ID 생성
        session_id = secrets.token_urlsafe(32)
        
        # 세션 데이터 구성
        session_data = {
            'user_id': user_id,
            'created_at': datetime.now().isoformat(),
            'last_activity': datetime.now().isoformat(),
            'ip_address': user_data.get('ip_address') if user_data else None,
            'user_agent': user_data.get('user_agent') if user_data else None,
        }
        
        # 추가 데이터 병합
        if user_data:
            session_data.update(user_data)
        
        # Redis에 저장
        session_key = f'{self.session_prefix}{session_id}'
        self.redis.hmset(session_key, session_data)
        self.redis.expire(session_key, self.session_timeout)
        
        # 사용자별 세션 목록에 추가
        user_sessions_key = f'user_sessions:{user_id}'
        self.redis.sadd(user_sessions_key, session_id)
        self.redis.expire(user_sessions_key, self.session_timeout)
        
        return session_id
    
    def get_session(self, session_id):
        """세션 조회"""
        session_key = f'{self.session_prefix}{session_id}'
        session_data = self.redis.hgetall(session_key)
        
        if not session_data:
            return None
        
        # 마지막 활동 시간 업데이트
        self.redis.hset(session_key, 'last_activity', 
                       datetime.now().isoformat())
        self.redis.expire(session_key, self.session_timeout)
        
        return session_data
    
    def update_session(self, session_id, data):
        """세션 업데이트"""
        session_key = f'{self.session_prefix}{session_id}'
        
        if not self.redis.exists(session_key):
            return False
        
        self.redis.hmset(session_key, data)
        self.redis.hset(session_key, 'last_activity', 
                       datetime.now().isoformat())
        self.redis.expire(session_key, self.session_timeout)
        
        return True
    
    def delete_session(self, session_id):
        """세션 삭제"""
        session_key = f'{self.session_prefix}{session_id}'
        session_data = self.redis.hgetall(session_key)
        
        if session_data:
            user_id = session_data.get('user_id')
            
            # 세션 삭제
            self.redis.delete(session_key)
            
            # 사용자 세션 목록에서 제거
            if user_id:
                user_sessions_key = f'user_sessions:{user_id}'
                self.redis.srem(user_sessions_key, session_id)
            
            return True
        
        return False
    
    def delete_user_sessions(self, user_id):
        """특정 사용자의 모든 세션 삭제"""
        user_sessions_key = f'user_sessions:{user_id}'
        session_ids = self.redis.smembers(user_sessions_key)
        
        for session_id in session_ids:
            self.delete_session(session_id)
        
        self.redis.delete(user_sessions_key)
        
        return len(session_ids)
    
    def get_user_sessions(self, user_id):
        """사용자의 모든 활성 세션 조회"""
        user_sessions_key = f'user_sessions:{user_id}'
        session_ids = self.redis.smembers(user_sessions_key)
        
        sessions = []
        for session_id in session_ids:
            session_data = self.get_session(session_id)
            if session_data:
                session_data['session_id'] = session_id
                sessions.append(session_data)
        
        return sessions
    
    def validate_session(self, session_id, ip_address=None, user_agent=None):
        """세션 유효성 검증"""
        session_data = self.get_session(session_id)
        
        if not session_data:
            return False, 'Session not found'
        
        # IP 주소 검증 (선택적)
        if ip_address and session_data.get('ip_address'):
            if session_data['ip_address'] != ip_address:
                return False, 'IP address mismatch'
        
        # User-Agent 검증 (선택적)
        if user_agent and session_data.get('user_agent'):
            if session_data['user_agent'] != user_agent:
                return False, 'User agent mismatch'
        
        return True, session_data
    
    def cleanup_expired_sessions(self):
        """만료된 세션 정리"""
        cursor = 0
        cleaned = 0
        
        while True:
            cursor, keys = self.redis.scan(
                cursor, 
                match=f'{self.session_prefix}*',
                count=100
            )
            
            for key in keys:
                if not self.redis.exists(key):
                    cleaned += 1
            
            if cursor == 0:
                break
        
        return cleaned

# Flask 애플리케이션 예제
from flask import Flask, request, jsonify, make_response

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, decode_responses=True)
session_manager = SessionManager(r, session_timeout=3600)

@app.route('/api/login', methods=['POST'])
def login():
    """로그인"""
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    # 실제로는 DB에서 사용자 확인
    # 여기서는 간단히 처리
    if username == 'admin' and password == 'password123':
        user_id = 1000
        
        # 세션 생성
        session_id = session_manager.create_session(
            user_id,
            {
                'username': username,
                'ip_address': request.remote_addr,
                'user_agent': request.headers.get('User-Agent'),
                'role': 'admin'
            }
        )
        
        # 쿠키 설정
        response = make_response(jsonify({
            'success': True,
            'message': '로그인 성공!'
        }))
        
        response.set_cookie(
            'session_id',
            session_id,
            max_age=3600,
            httponly=True,
            secure=True,  # HTTPS에서만
            samesite='Lax'
        )
        
        return response
    
    return jsonify({'success': False, 'message': '로그인 실패'}), 401

@app.route('/api/profile')
def get_profile():
    """프로필 조회 (인증 필요)"""
    session_id = request.cookies.get('session_id')
    
    if not session_id:
        return jsonify({'error': 'Not authenticated'}), 401
    
    # 세션 검증
    valid, result = session_manager.validate_session(
        session_id,
        ip_address=request.remote_addr,
        user_agent=request.headers.get('User-Agent')
    )
    
    if not valid:
        return jsonify({'error': result}), 401
    
    # 사용자 정보 반환
    return jsonify({
        'user_id': result['user_id'],
        'username': result['username'],
        'role': result['role']
    })

@app.route('/api/sessions')
def list_sessions():
    """내 세션 목록 조회"""
    session_id = request.cookies.get('session_id')
    
    if not session_id:
        return jsonify({'error': 'Not authenticated'}), 401
    
    session_data = session_manager.get_session(session_id)
    if not session_data:
        return jsonify({'error': 'Invalid session'}), 401
    
    user_id = session_data['user_id']
    sessions = session_manager.get_user_sessions(user_id)
    
    return jsonify({'sessions': sessions})

@app.route('/api/logout', methods=['POST'])
def logout():
    """로그아웃"""
    session_id = request.cookies.get('session_id')
    
    if session_id:
        session_manager.delete_session(session_id)
    
    response = make_response(jsonify({'success': True}))
    response.delete_cookie('session_id')
    
    return response

@app.route('/api/logout-all', methods=['POST'])
def logout_all():
    """모든 기기에서 로그아웃"""
    session_id = request.cookies.get('session_id')
    
    if not session_id:
        return jsonify({'error': 'Not authenticated'}), 401
    
    session_data = session_manager.get_session(session_id)
    if not session_data:
        return jsonify({'error': 'Invalid session'}), 401
    
    user_id = session_data['user_id']
    count = session_manager.delete_user_sessions(user_id)
    
    response = make_response(jsonify({
        'success': True,
        'message': f'{count}개의 세션이 종료되었습니다.'
    }))
    response.delete_cookie('session_id')
    
    return response

if __name__ == '__main__':
    app.run(debug=True)

🌟 마무리하며