Python Redis 개념 정리

1️⃣ Redis 기초

Redis는 인메모리 데이터 구조 저장소로, 고성능 캐싱, 메시지 브로커 및 실시간 애플리케이션의 데이터 저장소로 널리 사용된다. Python에서는 redis-py 라이브러리를 통해 쉽게 접근하고 활용할 수 있다.

import redis
from typing import Optional, Any, Dict, List, Union
import json
import time
import logging
from contextlib import contextmanager

# Redis 연결 클래스
class RedisClient:
    """Redis 서버 연결 및 관리를 위한 클라이언트 클래스"""
    
    def __init__(
        self,
        host: str = 'localhost',
        port: int = 6379,
        db: int = 0,
        password: Optional[str] = None,
        socket_timeout: float = 5.0,
        socket_connect_timeout: float = 5.0,
        socket_keepalive: bool = True,
        retry_on_timeout: bool = True,
        health_check_interval: int = 30,
        decode_responses: bool = True
    ):
        """
        Redis 클라이언트 초기화
        
        Args:
            host: Redis 서버 호스트
            port: Redis 서버 포트
            db: 사용할 데이터베이스 번호
            password: 인증 비밀번호
            socket_timeout: 소켓 작업 타임아웃(초)
            socket_connect_timeout: 연결 타임아웃(초)
            socket_keepalive: 소켓 keepalive 설정
            retry_on_timeout: 타임아웃 시 재시도 여부
            health_check_interval: 상태 확인 간격(초)
            decode_responses: 응답 자동 디코딩 여부
        """
        self.logger = logging.getLogger(__name__)
        
        # 연결 설정
        self.connection_params = {
            'host': host,
            'port': port,
            'db': db,
            'password': password,
            'socket_timeout': socket_timeout,
            'socket_connect_timeout': socket_connect_timeout,
            'socket_keepalive': socket_keepalive,
            'retry_on_timeout': retry_on_timeout,
            'health_check_interval': health_check_interval,
            'decode_responses': decode_responses
        }
        
        self.redis = self._create_connection()
    
    def _create_connection(self):
        """Redis 연결 생성"""
        try:
            self.logger.info(f"Redis 서버 {self.connection_params['host']}:{self.connection_params['port']} 연결 시도...")
            connection = redis.Redis(**self.connection_params)
            return connection
        except redis.ConnectionError as e:
            self.logger.error(f"Redis 연결 실패: {e}")
            raise
    
    def ping(self) -> bool:
        """
        Redis 서버 연결 테스트
        
        Returns:
            bool: 연결 성공 여부
        """
        try:
            return self.redis.ping()
        except redis.ConnectionError as e:
            self.logger.error(f"Redis 서버 연결 테스트 실패: {e}")
            return False
    
    def info(self) -> Dict:
        """
        Redis 서버 정보 조회
        
        Returns:
            Dict: Redis 서버 정보
        """
        try:
            return self.redis.info()
        except redis.RedisError as e:
            self.logger.error(f"Redis 서버 정보 조회 실패: {e}")
            return {}
    
    def close(self):
        """연결 종료"""
        try:
            self.redis.close()
            self.logger.info("Redis 연결 종료")
        except Exception as e:
            self.logger.error(f"Redis 연결 종료 중 오류: {e}")
    
    @contextmanager
    def pipeline_context(self, transaction: bool = True):
        """
        파이프라인 컨텍스트 매니저
        
        사용 예:
        ```
        with client.pipeline_context() as pipe:
            pipe.set('key1', 'value1')
            pipe.set('key2', 'value2')
            results = pipe.execute()
        ```
        
        Args:
            transaction: 트랜잭션 사용 여부
            
        Yields:
            Pipeline: Redis 파이프라인 객체
        """
        pipeline = self.redis.pipeline(transaction=transaction)
        try:
            yield pipeline
        finally:
            pipeline.execute()
    
    def flush_db(self, force: bool = False):
        """
        현재 데이터베이스의 모든 키 삭제
        
        Args:
            force: 강제 삭제 여부 (안전장치)
            
        Returns:
            bool: 성공 여부
        """
        if not force:
            self.logger.warning("flush_db 호출 시 force=True 옵션이 필요하다")
            return False
            
        try:
            self.redis.flushdb()
            self.logger.info(f"데이터베이스 {self.connection_params['db']} 비움")
            return True
        except redis.RedisError as e:
            self.logger.error(f"데이터베이스 비우기 실패: {e}")
            return False

✅ 특징:

안정적인 연결 설정 및 관리
다양한 연결 옵션 지원
파이프라인을 통한 성능 최적화
자동 응답 디코딩 설정
타입 힌팅으로 코드 가독성 향상
컨텍스트 매니저를 통한 자원 관리
상세한 로깅으로 연결 문제 추적
서버 상태 확인 (ping, info)
안전한 데이터베이스 초기화 메커니즘

2️⃣ 기본 작업

Redis는 문자열, 해시, 리스트, 집합, 정렬 집합 등 다양한 데이터 구조를 제공한다. 각 데이터 타입에 맞는 효율적인 작업 방법을 통해 다양한 사용 사례를 구현할 수 있다.

class RedisOperations:
    """Redis 기본 데이터 타입별 작업 클래스"""
    
    def __init__(self, client: RedisClient):
        """
        Redis 작업 클래스 초기화
        
        Args:
            client: Redis 클라이언트 인스턴스
        """
        self.redis = client.redis
        self.logger = logging.getLogger(__name__)
    
    # 문자열 작업
    def set_value(
        self, 
        key: str, 
        value: str, 
        expiry: Optional[int] = None,
        nx: bool = False,
        xx: bool = False
    ) -> bool:
        """
        키-값 저장
        
        Args:
            key: 키
            value: 값
            expiry: 만료 시간(초)
            nx: 키가 존재하지 않을 때만 설정
            xx: 키가 이미 존재할 때만 설정
            
        Returns:
            bool: 설정 성공 여부
        """
        try:
            return self.redis.set(key, value, ex=expiry, nx=nx, xx=xx)
        except redis.RedisError as e:
            self.logger.error(f"값 설정 실패 - 키: {key}, 오류: {e}")
            return False
    
    def get_value(self, key: str) -> Optional[str]:
        """
        값 조회
        
        Args:
            key: 키
            
        Returns:
            Optional[str]: 저장된 값 또는 None
        """
        try:
            return self.redis.get(key)
        except redis.RedisError as e:
            self.logger.error(f"값 조회 실패 - 키: {key}, 오류: {e}")
            return None
            
    def increment(self, key: str, amount: int = a1) -> Optional[int]:
        """
        숫자 값 증가
        
        Args:
            key: 키
            amount: 증가량
            
        Returns:
            Optional[int]: 증가 후 값 또는 None
        """
        try:
            return self.redis.incrby(key, amount)
        except redis.RedisError as e:
            self.logger.error(f"증분 실패 - 키: {key}, 오류: {e}")
            return None
    
    def set_with_ttl(self, key: str, value: str, ttl: int) -> bool:
        """
        TTL(Time-To-Live)과 함께 값 설정
        
        Args:
            key: 키
            value: 값
            ttl: 만료 시간(초)
            
        Returns:
            bool: 설정 성공 여부
        """
        try:
            return self.redis.setex(key, ttl, value)
        except redis.RedisError as e:
            self.logger.error(f"TTL과 함께 값 설정 실패 - 키: {key}, 오류: {e}")
            return False
    
    def set_json(self, key: str, data: Dict, expiry: Optional[int] = None) -> bool:
        """
        JSON 데이터 저장
        
        Args:
            key: 키
            data: JSON으로 직렬화할 데이터
            expiry: 만료 시간(초)
            
        Returns:
            bool: 설정 성공 여부
        """
        try:
            json_data = json.dumps(data)
            return self.set_value(key, json_data, expiry)
        except (redis.RedisError, TypeError) as e:
            self.logger.error(f"JSON 데이터 저장 실패 - 키: {key}, 오류: {e}")
            return False
    
    def get_json(self, key: str) -> Optional[Dict]:
        """
        JSON 데이터 조회
        
        Args:
            key: 키
            
        Returns:
            Optional[Dict]: 역직렬화된 데이터 또는 None
        """
        try:
            json_data = self.get_value(key)
            if json_data:
                return json.loads(json_data)
            return None
        except (redis.RedisError, json.JSONDecodeError) as e:
            self.logger.error(f"JSON 데이터 조회 실패 - 키: {key}, 오류: {e}")
            return None
    
    # 해시 작업
    def set_hash(self, name: str, mapping: Dict) -> bool:
        """
        해시 저장
        
        Args:
            name: 해시 이름
            mapping: 필드-값 매핑
            
        Returns:
            bool: 설정 성공 여부
        """
        try:
            self.redis.hset(name, mapping=mapping)
            return True
        except redis.RedisError as e:
            self.logger.error(f"해시 저장 실패 - 이름: {name}, 오류: {e}")
            return False
    
    def get_hash(self, name: str) -> Dict:
        """
        해시 전체 조회
        
        Args:
            name: 해시 이름
            
        Returns:
            Dict: 해시의 모든 필드-값 쌍
        """
        try:
            return self.redis.hgetall(name)
        except redis.RedisError as e:
            self.logger.error(f"해시 조회 실패 - 이름: {name}, 오류: {e}")
            return {}
    
    def get_hash_field(self, name: str, field: str) -> Optional[str]:
        """
        해시 필드 값 조회
        
        Args:
            name: 해시 이름
            field: 필드 이름
            
        Returns:
            Optional[str]: 필드 값 또는 None
        """
        try:
            return self.redis.hget(name, field)
        except redis.RedisError as e:
            self.logger.error(f"해시 필드 조회 실패 - 이름: {name}, 필드: {field}, 오류: {e}")
            return None
    
    def increment_hash_field(self, name: str, field: str, amount: int = 1) -> Optional[int]:
        """
        해시 필드 값 증가
        
        Args:
            name: 해시 이름
            field: 필드 이름
            amount: 증가량
            
        Returns:
            Optional[int]: 증가 후 값 또는 None
        """
        try:
            return self.redis.hincrby(name, field, amount)
        except redis.RedisError as e:
            self.logger.error(f"해시 필드 증분 실패 - 이름: {name}, 필드: {field}, 오류: {e}")
            return None
    
    # 리스트 작업
    def push_to_list(self, name: str, *values: str) -> Optional[int]:
        """
        리스트에 값 추가 (오른쪽)
        
        Args:
            name: 리스트 이름
            *values: 추가할 값들
            
        Returns:
            Optional[int]: 추가 후 리스트 길이 또는 None
        """
        try:
            return self.redis.rpush(name, *values)
        except redis.RedisError as e:
            self.logger.error(f"리스트 추가 실패 - 이름: {name}, 오류: {e}")
            return None
    
    def push_to_list_left(self, name: str, *values: str) -> Optional[int]:
        """
        리스트에 값 추가 (왼쪽)
        
        Args:
            name: 리스트 이름
            *values: 추가할 값들
            
        Returns:
            Optional[int]: 추가 후 리스트 길이 또는 None
        """
        try:
            return self.redis.lpush(name, *values)
        except redis.RedisError as e:
            self.logger.error(f"리스트 왼쪽 추가 실패 - 이름: {name}, 오류: {e}")
            return None
    
    def pop_from_list(self, name: str) -> Optional[str]:
        """
        리스트에서 값 꺼내기 (오른쪽)
        
        Args:
            name: 리스트 이름
            
        Returns:
            Optional[str]: 꺼낸 값 또는 None
        """
        try:
            return self.redis.rpop(name)
        except redis.RedisError as e:
            self.logger.error(f"리스트 꺼내기 실패 - 이름: {name}, 오류: {e}")
            return None
    
    def pop_from_list_left(self, name: str) -> Optional[str]:
        """
        리스트에서 값 꺼내기 (왼쪽)
        
        Args:
            name: 리스트 이름
            
        Returns:
            Optional[str]: 꺼낸 값 또는 None
        """
        try:
            return self.redis.lpop(name)
        except redis.RedisError as e:
            self.logger.error(f"리스트 왼쪽 꺼내기 실패 - 이름: {name}, 오류: {e}")
            return None
    
    def get_list(self, name: str, start: int = 0, end: int = -1) -> List[str]:
        """
        리스트 범위 조회
        
        Args:
            name: 리스트 이름
            start: 시작 인덱스
            end: 종료 인덱스 (-1은 끝까지)
            
        Returns:
            List[str]: 리스트 요소들
        """
        try:
            return self.redis.lrange(name, start, end)
        except redis.RedisError as e:
            self.logger.error(f"리스트 조회 실패 - 이름: {name}, 오류: {e}")
            return []
    
    def get_list_length(self, name: str) -> int:
        """
        리스트 길이 조회
        
        Args:
            name: 리스트 이름
            
        Returns:
            int: 리스트 길이
        """
        try:
            return self.redis.llen(name)
        except redis.RedisError as e:
            self.logger.error(f"리스트 길이 조회 실패 - 이름: {name}, 오류: {e}")
            return 0
    
    # 집합 작업
    def add_to_set(self, name: str, *values: str) -> Optional[int]:
        """
        집합에 값 추가
        
        Args:
            name: 집합 이름
            *values: 추가할 값들
            
        Returns:
            Optional[int]: 새로 추가된 값의 수 또는 None
        """
        try:
            return self.redis.sadd(name, *values)
        except redis.RedisError as e:
            self.logger.error(f"집합 추가 실패 - 이름: {name}, 오류: {e}")
            return None
    
    def get_set_members(self, name: str) -> List[str]:
        """
        집합 멤버 조회
        
        Args:
            name: 집합 이름
            
        Returns:
            List[str]: 집합 멤버 목록
        """
        try:
            return list(self.redis.smembers(name))
        except redis.RedisError as e:
            self.logger.error(f"집합 멤버 조회 실패 - 이름: {name}, 오류: {e}")
            return []
    
    def is_member(self, name: str, value: str) -> bool:
        """
        값이 집합의 멤버인지 확인
        
        Args:
            name: 집합 이름
            value: 확인할 값
            
        Returns:
            bool: 멤버 여부
        """
        try:
            return self.redis.sismember(name, value)
        except redis.RedisError as e:
            self.logger.error(f"집합 멤버 확인 실패 - 이름: {name}, 값: {value}, 오류: {e}")
            return False
    
    def remove_from_set(self, name: str, *values: str) -> Optional[int]:
        """
        집합에서 값 제거
        
        Args:
            name: 집합 이름
            *values: 제거할 값들
            
        Returns:
            Optional[int]: 제거된 값의 수 또는 None
        """
        try:
            return self.redis.srem(name, *values)
        except redis.RedisError as e:
            self.logger.error(f"집합 제거 실패 - 이름: {name}, 오류: {e}")
            return None
    
    # 정렬된 집합 작업
    def add_to_sorted_set(self, name: str, mapping: Dict[str, float]) -> Optional[int]:
        """
        정렬된 집합에 값 추가
        
        Args:
            name: 정렬된 집합 이름
            mapping: 값-점수 매핑
            
        Returns:
            Optional[int]: 새로 추가된 값의 수 또는 None
        """
        try:
            return self.redis.zadd(name, mapping)
        except redis.RedisError as e:
            self.logger.error(f"정렬된 집합 추가 실패 - 이름: {name}, 오류: {e}")
            return None
    
    def get_sorted_set_range(self, name: str, start: int = 0, end: int = -1, withscores: bool = False) -> Union[List[str], List[tuple]]:
        """
        정렬된 집합 범위 조회 (점수 오름차순)
        
        Args:
            name: 정렬된 집합 이름
            start: 시작 인덱스
            end: 종료 인덱스 (-1은 끝까지)
            withscores: 점수 포함 여부
            
        Returns:
            Union[List[str], List[tuple]]: 값 목록 또는 (값, 점수) 튜플 목록
        """
        try:
            return self.redis.zrange(name, start, end, withscores=withscores)
        except redis.RedisError as e:
            self.logger.error(f"정렬된 집합 범위 조회 실패 - 이름: {name}, 오류: {e}")
            return []
    
    def get_sorted_set_rev_range(self, name: str, start: int = 0, end: int = -1, withscores: bool = False) -> Union[List[str], List[tuple]]:
        """
        정렬된 집합 범위 조회 (점수 내림차순)
        
        Args:
            name: 정렬된 집합 이름
            start: 시작 인덱스
            end: 종료 인덱스 (-1은 끝까지)
            withscores: 점수 포함 여부
            
        Returns:
            Union[List[str], List[tuple]]: 값 목록 또는 (값, 점수) 튜플 목록
        """
        try:
            return self.redis.zrevrange(name, start, end, withscores=withscores)
        except redis.RedisError as e:
            self.logger.error(f"정렬된 집합 역순 범위 조회 실패 - 이름: {name}, 오류: {e}")
            return []
    
    def increment_score(self, name: str, value: str, amount: float = 1.0) -> Optional[float]:
        """
        정렬된 집합 멤버의 점수 증가
        
        Args:
            name: 정렬된 집합 이름
            value: 값
            amount: 증가량
            
        Returns:
            Optional[float]: 새 점수 또는 None
        """
        try:
            return self.redis.zincrby(name, amount, value)
        except redis.RedisError as e:
            self.logger.error(f"정렬된 집합 점수 증가 실패 - 이름: {name}, 값: {value}, 오류: {e}")
            return None
    
    # 키 관리
    def delete_key(self, *names: str) -> int:
        """
        키 삭제
        
        Args:
            *names: 삭제할 키 이름들
            
        Returns:
            int: 삭제된 키의 수
        """
        try:
            return self.redis.delete(*names)
        except redis.RedisError as e:
            self.logger.error(f"키 삭제 실패 - 키: {names}, 오류: {e}")
            return 0
    
    def set_expiry(self, name: str, time_seconds: int) -> bool:
        """
        키 만료 시간 설정
        
        Args:
            name: 키 이름
            time_seconds: 만료 시간(초)
            
        Returns:
            bool: 설정 성공 여부
        """
        try:
            return self.redis.expire(name, time_seconds)
        except redis.RedisError as e:
            self.logger.error(f"만료 시간 설정 실패 - 키: {name}, 오류: {e}")
            return False
    
    def get_ttl(self, name: str) -> int:
        """
        키 만료 시간 조회
        
        Args:
            name: 키 이름
            
        Returns:
            int: 남은 시간(초), -1은 영구적, -2는 키 없음
        """
        try:
            return self.redis.ttl(name)
        except redis.RedisError as e:
            self.logger.error(f"TTL 조회 실패 - 키: {name}, 오류: {e}")
            return -2
    
    def exists(self, *names: str) -> int:
        """
        키 존재 여부 확인
        
        Args:
            *names: 확인할 키 이름들
            
        Returns:
            int: 존재하는 키의 수
        """
        try:
            return self.redis.exists(*names)
        except redis.RedisError as e:
            self.logger.error(f"키 존재 확인 실패 - 키: {names}, 오류: {e}")
            return 0

✅ 특징:

모든 Redis 데이터 타입 지원 (문자열, 해시, 리스트, 집합, 정렬 집합)
각 데이터 타입별 전용 메서드
일관된 오류 처리와 로깅
JSON 데이터 직렬화/역직렬화 지원
TTL(Time-To-Live) 설정 및 관리
원자적 증분 연산 지원
다양한 리스트 작업 (양방향 추가/제거)
집합 연산 지원
정렬된 집합 관리 (순위, 점수 기반)
키 관리 기능 (삭제, 만료 설정, 존재 확인)

3️⃣ 캐싱

Redis는 빠른 접근 속도와 TTL 기능을 통해 애플리케이션의 성능을 향상시키는 효과적인 캐싱 시스템으로 활용된다. 함수 결과나 데이터베이스 쿼리 결과 등을 캐싱하여 반복적인 계산이나 IO 작업을 최소화할 수 있다.

from functools import wraps
import pickle
import hashlib
from datetime import timedelta
import inspect
import time

class RedisCache:
    """Redis를 활용한 캐싱 구현 클래스"""
    
    def __init__(self, client: RedisClient, prefix: str = "cache"):
        """
        Redis 캐시 초기화
        
        Args:
            client: Redis 클라이언트 인스턴스
            prefix: 캐시 키 접두사
        """
        self.redis = client.redis
        self.prefix = prefix
        self.logger = logging.getLogger(__name__)
    
    def _make_cache_key(self, func_name: str, args: tuple, kwargs: dict) -> str:
        """
        함수 호출에 대한 캐시 키 생성
        
        Args:
            func_name: 함수 이름
            args: 위치 인자
            kwargs: 키워드 인자
            
        Returns:
            str: 생성된 캐시 키
        """
        # 인자를 정렬하여 일관된 키 생성
        key_parts = [func_name, str(args)]
        
        if kwargs:
            sorted_items = sorted(kwargs.items())
            key_parts.append(str(sorted_items))
        
        # SHA-256 해시 생성
        key_str = "::".join(key_parts)
        hashed = hashlib.sha256(key_str.encode()).hexdigest()
        
        return f"{self.prefix}:{func_name}:{hashed}"
    
    def cache_decorator(self, expiry: int = 3600):
        """
        함수 결과를 캐시하는 데코레이터
        
        Args:
            expiry: 캐시 만료 시간(초)
            
        Returns:
            callable: 데코레이터 함수
        """
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 함수 이름 가져오기
                func_name = func.__name__
                
                # 캐시 키 생성
                cache_key = self._make_cache_key(func_name, args, kwargs)
                
                # 캐시된 결과 확인
                cached_result = self.redis.get(cache_key)
                if cached_result:
                    self.logger.debug(f"캐시 히트: {func_name}")
                    try:
                        return pickle.loads(cached_result)
                    except pickle.PickleError as e:
                        self.logger.error(f"캐시 역직렬화 실패: {e}")
                
                # 캐시 미스 - 함수 실행
                self.logger.debug(f"캐시 미스: {func_name}")
                start_time = time.time()
                result = func(*args, **kwargs)
                execution_time = time.time() - start_time
                
                # 결과 캐싱
                try:
                    serialized = pickle.dumps(result)
                    self.redis.setex(cache_key, expiry, serialized)
                    self.logger.debug(
                        f"캐시 저장: {func_name}, 실행 시간: {execution_time:.4f}s, 크기: {len(serialized)} 바이트"
                    )
                except (pickle.PickleError, redis.RedisError) as e:
                    self.logger.error(f"캐시 저장 실패: {e}")
                
                return result
            return wrapper
        return decorator
    
    def invalidate_cache(self, pattern: str = "*") -> int:
        """
        패턴에 맞는 캐시 무효화
        
        Args:
            pattern: 캐시 키 패턴
            
        Returns:
            int: 삭제된 키의 수
        """
        try:
            # 패턴에 맞는 키 찾기
            cache_pattern = f"{self.prefix}:{pattern}"
            keys = self.redis.keys(cache_pattern)
            
            if not keys:
                return 0
                
            # 키 삭제
            deleted = self.redis.delete(*keys)
            self.logger.info(f"{deleted}개의 캐시 항목 삭제됨: {cache_pattern}")
            return deleted
        except redis.RedisError as e:
            self.logger.error(f"캐시 무효화 실패: {e}")
            return 0
    
    def memoize(self, expiry: int = 3600):
        """
        함수 결과를 메모이제이션하는 데코레이터
        (동일 인자에 대해 함수 내부에서 한 번만 실행)
        
        Args:
            expiry: 캐시 만료 시간(초)
            
        Returns:
            callable: 데코레이터 함수
        """
        return self.cache_decorator(expiry)
    
    def cached_property(self, expiry: int = 3600):
        """
        클래스 프로퍼티 결과를 캐시하는 데코레이터
        
        Args:
            expiry: 캐시 만료 시간(초)
            
        Returns:
            callable: 데코레이터 함수
        """
        def decorator(method):
            @property
            @wraps(method)
            def wrapper(self):
                # 인스턴스 정보를 포함한 캐시 키 생성
                instance_id = id(self)
                method_name = method.__name__
                class_name = self.__class__.__name__
                
                cache_key = f"{self.prefix}:{class_name}:{instance_id}:{method_name}"
                
                # 캐시 확인
                cached_result = self.redis.get(cache_key)
                if cached_result:
                    try:
                        return pickle.loads(cached_result)
                    except pickle.PickleError:
                        pass
                
                # 메서드 실행
                result = method(self)
                
                # 결과 캐싱
                try:
                    self.redis.setex(cache_key, expiry, pickle.dumps(result))
                except (pickle.PickleError, redis.RedisError):
                    pass
                
                return result
            return wrapper
        return decorator

    def cache_context(self, key: str, expiry: int = 3600):
        """
        컨텍스트 매니저를 통한 캐싱
        
        Args:
            key: 캐시 키
            expiry: 캐시 만료 시간(초)
            
        Returns:
            contextmanager: 컨텍스트 매니저
        """
        @contextmanager
        def cache_cm():
            cache_key = f"{self.prefix}:{key}"
            
            # 캐시 확인
            cached_data = self.redis.get(cache_key)
            if cached_data:
                try:
                    yield pickle.loads(cached_data), True  # (데이터, 캐시 히트)
                    return
                except pickle.PickleError:
                    pass
            
            # 캐시 미스
            result_container = []
            
            try:
                yield result_container, False  # (결과 컨테이너, 캐시 미스)
                
                # 결과가 컨테이너에 저장되었는지 확인
                if result_container:
                    # 첫 번째 항목을 캐시
                    try:
                        self.redis.setex(cache_key, expiry, pickle.dumps(result_container[0]))
                    except (pickle.PickleError, redis.RedisError) as e:
                        self.logger.error(f"컨텍스트 캐시 저장 실패: {e}")
            except Exception as e:
                self.logger.error(f"캐시 컨텍스트 오류: {e}")
                raise
        
        return cache_cm()

✅ 특징:

데코레이터 패턴을 통한 간편한 캐싱
안정적인 캐시 키 생성 (SHA-256 해시 사용)
직렬화를 통한 복잡한 객체 캐싱
패턴 기반 캐시 무효화
메모이제이션 지원
프로퍼티 캐싱 기능
컨텍스트 매니저를 통한 유연한 캐싱
실행 시간 측정 및 로깅
예외 처리를 통한 안정성 확보
일관된 키 접두사 관리

4️⃣ 세션 관리

Redis의 빠른 처리 속도와 TTL 기능은 웹 애플리케이션의 세션 관리에 이상적이다. 인증 정보와 사용자 상태를 안전하게 저장하고 관리할 수 있으며, 분산 환경에서도 일관된 세션 접근이 가능하다.

import uuid
from datetime import datetime, timedelta
import json
import secrets
import time

class RedisSession:
    """Redis를 활용한 세션 관리 클래스"""
    
    def __init__(
        self, 
        client: RedisClient, 
        prefix: str = "session",
        default_expiry: int = 86400  # 24시간
    ):
        """
        Redis 세션 관리자 초기화
        
        Args:
            client: Redis 클라이언트 인스턴스
            prefix: 세션 키 접두사
            default_expiry: 기본 세션 만료 시간(초)
        """
        self.redis = client.redis
        self.prefix = prefix
        self.default_expiry = default_expiry
        self.logger = logging.getLogger(__name__)
    
    def _get_session_key(self, session_id: str) -> str:
        """
        세션 ID로부터 세션 키 생성
        
        Args:
            session_id: 세션 ID
            
        Returns:
            str: 세션 키
        """
        return f"{self.prefix}:{session_id}"
    
    def create_session(
        self, 
        user_id: str, 
        data: dict,
        expiry: Optional[int] = None
    ) -> str:
        """
        새 세션 생성
        
        Args:
            user_id: 사용자 ID
            data: 세션에 저장할 데이터
            expiry: 세션 만료 시간(초), None이면 기본값 사용
            
        Returns:
            str: 생성된 세션 ID
        """
        # 세션 ID 생성 (UUID v4)
        session_id = str(uuid.uuid4())
        session_key = self._get_session_key(session_id)
        
        # 세션 데이터 준비
        current_time = datetime.now().isoformat()
        session_data = {
            "user_id": user_id,
            "created_at": current_time,
            "last_accessed": current_time,
            "ip_address": data.get("ip_address", "unknown"),
            "user_agent": data.get("user_agent", "unknown"),
            **data
        }
        
        # TTL 설정
        ttl = expiry if expiry is not None else self.default_expiry
        
        try:
            # 세션 저장
            self.redis.hset(session_key, mapping=session_data)
            self.redis.expire(session_key, timedelta(seconds=ttl))
            
            # 사용자 ID로 세션 인덱싱 (한 사용자가 여러 세션을 가질 수 있음)
            user_sessions_key = f"user_sessions:{user_id}"
            self.redis.sadd(user_sessions_key, session_id)
            
            self.logger.info(f"세션 생성: {session_id} (사용자: {user_id})")
            return session_id
            
        except redis.RedisError as e:
            self.logger.error(f"세션 생성 실패: {e}")
            raise
    
    def get_session(self, session_id: str, update_access: bool = True) -> Optional[dict]:
        """
        세션 정보 조회
        
        Args:
            session_id: 세션 ID
            update_access: 접근 시간 업데이트 여부
            
        Returns:
            Optional[dict]: 세션 데이터 또는 None
        """
        if not session_id:
            return None
            
        session_key = self._get_session_key(session_id)
        
        try:
            # 세션 데이터 조회
            session_data = self.redis.hgetall(session_key)
            
            if not session_data:
                return None
                
            # 마지막 접근 시간 업데이트
            if update_access:
                self.redis.hset(session_key, "last_accessed", datetime.now().isoformat())
                
            return session_data
            
        except redis.RedisError as e:
            self.logger.error(f"세션 조회 실패: {e}")
            return None
    
    def update_session(self, session_id: str, data: dict) -> bool:
        """
        세션 데이터 업데이트
        
        Args:
            session_id: 세션 ID
            data: 업데이트할 데이터
            
        Returns:
            bool: 업데이트 성공 여부
        """
        session_key = self._get_session_key(session_id)
        
        try:
            # 세션 존재 확인
            if not self.redis.exists(session_key):
                return False
                
            # 업데이트할 데이터에 마지막 접근 시간 추가
            update_data = {
                **data,
                "last_accessed": datetime.now().isoformat()
            }
            
            # 세션 업데이트
            self.redis.hset(session_key, mapping=update_data)
            return True
            
        except redis.RedisError as e:
            self.logger.error(f"세션 업데이트 실패: {e}")
            return False
    
    def delete_session(self, session_id: str) -> bool:
        """
        세션 삭제
        
        Args:
            session_id: 세션 ID
            
        Returns:
            bool: 삭제 성공 여부
        """
        session_key = self._get_session_key(session_id)
        
        try:
            # 세션 데이터에서 사용자 ID 가져오기
            user_id = self.redis.hget(session_key, "user_id")
            
            # 세션 삭제
            if self.redis.delete(session_key):
                # 사용자 세션 인덱스에서도 삭제
                if user_id:
                    self.redis.srem(f"user_sessions:{user_id}", session_id)
                
                self.logger.info(f"세션 삭제: {session_id}")
                return True
            return False
            
        except redis.RedisError as e:
            self.logger.error(f"세션 삭제 실패: {e}")
            return False
    
    def extend_session(self, session_id: str, extension: int) -> bool:
        """
        세션 만료 시간 연장
        
        Args:
            session_id: 세션 ID
            extension: 연장할 시간(초)
            
        Returns:
            bool: 연장 성공 여부
        """
        session_key = self._get_session_key(session_id)
        
        try:
            # 현재 남은 시간에 연장 시간 추가
            return self.redis.expire(session_key, timedelta(seconds=extension))
        except redis.RedisError as e:
            self.logger.error(f"세션 연장 실패: {e}")
            return False
    
    def get_user_sessions(self, user_id: str) -> List[str]:
        """
        사용자의 모든 세션 ID 조회
        
        Args:
            user_id: 사용자 ID
            
        Returns:
            List[str]: 세션 ID 목록
        """
        try:
            user_sessions_key = f"user_sessions:{user_id}"
            return list(self.redis.smembers(user_sessions_key))
        except redis.RedisError as e:
            self.logger.error(f"사용자 세션 조회 실패: {e}")
            return []
    
    def invalidate_user_sessions(self, user_id: str, keep_current: Optional[str] = None) -> int:
        """
        사용자의 모든 세션 무효화
        
        Args:
            user_id: 사용자 ID
            keep_current: 유지할 현재 세션 ID (선택 사항)
            
        Returns:
            int: 무효화된 세션 수
        """
        try:
            # 사용자의 모든 세션 ID 가져오기
            sessions = self.get_user_sessions(user_id)
            if not sessions:
                return 0
                
            # 현재 세션을 제외한 모든 세션 삭제
            deleted_count = 0
            for session_id in sessions:
                if keep_current and session_id == keep_current:
                    continue
                    
                if self.delete_session(session_id):
                    deleted_count += 1
            
            self.logger.info(f"사용자 {user_id}의 세션 {deleted_count}개 무효화")
            return deleted_count
            
        except redis.RedisError as e:
            self.logger.error(f"사용자 세션 무효화 실패: {e}")
            return 0
    
    def create_csrf_token(self, session_id: str) -> Optional[str]:
        """
        CSRF 토큰 생성 및 세션에 저장
        
        Args:
            session_id: 세션 ID
            
        Returns:
            Optional[str]: CSRF 토큰 또는 None
        """
        try:
            # 랜덤 토큰 생성
            csrf_token = secrets.token_hex(32)
            
            # 세션에 토큰 저장
            if self.update_session(session_id, {"csrf_token": csrf_token}):
                return csrf_token
            return None
            
        except Exception as e:
            self.logger.error(f"CSRF 토큰 생성 실패: {e}")
            return None
    
    def validate_csrf_token(self, session_id: str, token: str) -> bool:
        """
        CSRF 토큰 검증
        
        Args:
            session_id: 세션 ID
            token: 검증할 CSRF 토큰
            
        Returns:
            bool: 유효 여부
        """
        try:
            # 세션에서 토큰 가져오기
            session = self.get_session(session_id, update_access=False)
            if not session:
                return False
                
            # 토큰 비교 (시간 독립적 비교)
            stored_token = session.get("csrf_token")
            if not stored_token:
                return False
                
            return secrets.compare_digest(stored_token, token)
            
        except Exception as e:
            self.logger.error(f"CSRF 토큰 검증 실패: {e}")
            return False

# 사용 예제
def create_user_session(redis_session, user_info):
    """사용자 로그인 후 세션 생성 예제"""
    user_id = user_info["id"]
    ip_address = user_info.get("ip_address", "127.0.0.1")
    user_agent = user_info.get("user_agent", "Unknown")
    
    # 세션 데이터 준비
    session_data = {
        "username": user_info["username"],
        "email": user_info["email"],
        "role": user_info["role"],
        "ip_address": ip_address,
        "user_agent": user_agent,
        "login_time": int(time.time())
    }
    
    # 세션 생성
    session_id = redis_session.create_session(user_id, session_data)
    
    # CSRF 토큰 생성
    csrf_token = redis_session.create_csrf_token(session_id)
    
    return {
        "session_id": session_id,
        "csrf_token": csrf_token
    }

✅ 특징:

안전한 세션 ID 생성 (UUID 사용)
유연한 세션 데이터 저장
자동 만료 시간 관리
세션 접근 시간 추적
사용자별 세션 인덱싱
세션 데이터 업데이트 기능
세션 연장 지원
사용자별 세션 관리
CSRF 보호 기능 내장
안전한 토큰 비교 (타이밍 어택 방지)

5️⃣ 고급 패턴

Redis는 단순한 키-값 저장소를 넘어 다양한 고급 패턴을 구현할 수 있다. 메시지 큐, 분산 락, 속도 제한, 리더보드 등의 기능을 효율적으로 구현하여 복잡한 애플리케이션 요구사항을 해결할 수 있다.

import time
import random
import uuid
from datetime import datetime
from typing import Optional, Dict, List, Tuple, Callable, Any

class RedisAdvancedPatterns:
    """Redis를 활용한 고급 패턴 구현 클래스"""
    
    def __init__(self, client: RedisClient):
        """
        Redis 고급 패턴 초기화
        
        Args:
            client: Redis 클라이언트 인스턴스
        """
        self.redis = client.redis
        self.logger = logging.getLogger(__name__)
    
    # 메시지 큐 패턴
    def enqueue(self, queue_name: str, message: Dict[str, Any]) -> bool:
        """
        메시지 큐에 메시지 추가
        
        Args:
            queue_name: 큐 이름
            message: 메시지 데이터
            
        Returns:
            bool: 성공 여부
        """
        try:
            # 메시지에 메타데이터 추가
            full_message = {
                "id": str(uuid.uuid4()),
                "timestamp": datetime.now().isoformat(),
                "data": message
            }
            
            # 큐에 메시지 추가 (오른쪽에 추가)
            self.redis.rpush(queue_name, json.dumps(full_message))
            return True
        except (redis.RedisError, json.JSONDecodeError) as e:
            self.logger.error(f"메시지 큐 추가 실패: {e}")
            return False
    
    def dequeue(self, queue_name: str, timeout: int = 0) -> Optional[Dict]:
        """
        메시지 큐에서 메시지 가져오기 (블로킹)
        
        Args:
            queue_name: 큐 이름
            timeout: 타임아웃(초), 0은 무한 대기
            
        Returns:
            Optional[Dict]: 메시지 또는 None
        """
        try:
            # 왼쪽에서 메시지 가져오기 (BLPOP: 블로킹 왼쪽 팝)
            result = self.redis.blpop(queue_name, timeout)
            
            # 타임아웃된 경우
            if not result:
                return None
                
            # 결과는 (큐 이름, 메시지) 형태
            _, message_data = result
            
            # JSON 역직렬화
            return json.loads(message_data)
        except (redis.RedisError, json.JSONDecodeError) as e:
            self.logger.error(f"메시지 큐 가져오기 실패: {e}")
            return None
    
    def queue_length(self, queue_name: str) -> int:
        """
        큐 길이 조회
        
        Args:
            queue_name: 큐 이름
            
        Returns:
            int: 큐 길이
        """
        try:
            return self.redis.llen(queue_name)
        except redis.RedisError as e:
            self.logger.error(f"큐 길이 조회 실패: {e}")
            return 0
    
    # 분산 락 패턴
    def acquire_lock(
        self, 
        lock_name: str, 
        owner: str = None,
        expiry: int = 10,
        retry_count: int = 3,
        retry_delay: float = 0.2
    ) -> Optional[str]:
        """
        분산 락 획득
        
        Args:
            lock_name: 락 이름
            owner: 락 소유자 식별자
            expiry: 락 만료 시간(초)
            retry_count: 재시도 횟수
            retry_delay: 재시도 간격(초)
            
        Returns:
            Optional[str]: 락 식별자 또는 None
        """
        # 락 키
        lock_key = f"lock:{lock_name}"
        
        # 락 소유자 식별자
        owner_id = owner or str(uuid.uuid4())
        
        # 재시도 루프
        for attempt in range(retry_count + 1):
            # 락 시도 (NX: 키가 없을 때만 설정)
            success = self.redis.set(lock_key, owner_id, ex=expiry, nx=True)
            
            if success:
                self.logger.debug(f"락 획득: {lock_name} (소유자: {owner_id})")
                return owner_id
                
            # 마지막 시도가 아니면 대기 후 재시도
            if attempt < retry_count:
                # 무작위 지터 추가 (경합 해결)
                jitter = random.uniform(0, 0.1)
                time.sleep(retry_delay + jitter)
        
        self.logger.debug(f"락 획득 실패: {lock_name}")
        return None
    
    def release_lock(self, lock_name: str, owner_id: str) -> bool:
        """
        분산 락 해제
        
        Args:
            lock_name: 락 이름
            owner_id: 락 소유자 식별자
            
        Returns:
            bool: 성공 여부
        """
        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
        """
        
        try:
            # 스크립트 실행
            result = self.redis.eval(script, 1, lock_key, owner_id)
            success = result == 1
            
            if success:
                self.logger.debug(f"락 해제: {lock_name} (소유자: {owner_id})")
            else:
                self.logger.debug(f"락 해제 실패: {lock_name} (소유자 불일치 또는 만료)")
                
            return success
        except redis.RedisError as e:
            self.logger.error(f"락 해제 오류: {e}")
            return False
    
    @contextmanager
    def distributed_lock(
        self, 
        lock_name: str, 
        expiry: int = 10,
        retry_count: int = 3,
        retry_delay: float = 0.2
    ):
        """
        분산 락 컨텍스트 매니저
        
        Args:
            lock_name: 락 이름
            expiry: 락 만료 시간(초)
            retry_count: 재시도 횟수
            retry_delay: 재시도 간격(초)
            
        Raises:
            RuntimeError: 락 획득 실패 시
        """
        owner_id = None
        
        try:
            # 락 획득
            owner_id = self.acquire_lock(
                lock_name, 
                expiry=expiry,
                retry_count=retry_count,
                retry_delay=retry_delay
            )
            
            if not owner_id:
                raise RuntimeError(f"락 획득 실패: {lock_name}")
                
            # 락 획득 성공, 컨텍스트 진입
            yield
            
        finally:
            # 락 해제 (소유자가 있는 경우에만)
            if owner_id:
                self.release_lock(lock_name, owner_id)
    
    # 속도 제한 패턴
    def rate_limit(
        self, 
        key: str, 
        limit: int, 
        period: int
    ) -> Tuple[bool, int, int]:
        """
        속도 제한 체크 (슬라이딩 윈도우 알고리즘)
        
        Args:
            key: 제한할 키 (예: IP 또는 사용자 ID)
            limit: 기간 내 허용 요청 수
            period: 기간(초)
            
        Returns:
            Tuple[bool, int, int]: (허용 여부, 현재 요청 수, 남은 초)
        """
        # 속도 제한 키 및 현재 시간
        rate_key = f"ratelimit:{key}"
        now = int(time.time())
        
        # 파이프라인을 통한 원자적 연산
        pipe = self.redis.pipeline()
        
        try:
            # 만료된 요청 제거 (현재 시간 - 기간보다 이전 요청)
            pipe.zremrangebyscore(rate_key, 0, now - period)
            
            # 새 요청 추가
            pipe.zadd(rate_key, {str(now): now})
            
            # 현재 요청 수 조회
            pipe.zcard(rate_key)
            
            # 키 만료 설정 (기간 + 1초)
            pipe.expire(rate_key, period + 1)
            
            # 파이프라인 실행
            _, _, request_count, _ = pipe.execute()
            
            # 요청 허용 여부 결정
            allowed = request_count <= limit
            
            # TTL 확인
            ttl = self.redis.ttl(rate_key)
            
            return allowed, request_count, ttl
            
        except redis.RedisError as e:
            self.logger.error(f"속도 제한 체크 실패: {e}")
            return False, 0, 0
    
    # 리더보드 패턴
    def add_to_leaderboard(self, leaderboard: str, user: str, score: float) -> int:
        """
        리더보드에 점수 추가/갱신
        
        Args:
            leaderboard: 리더보드 이름
            user: 사용자 식별자
            score: 점수
            
        Returns:
            int: 새로 추가된 경우 1, 갱신된 경우 0
        """
        try:
            return self.redis.zadd(leaderboard, {user: score})
        except redis.RedisError as e:
            self.logger.error(f"리더보드 추가 실패: {e}")
            return 0
    
    def get_leaderboard(
        self, 
        leaderboard: str, 
        start: int = 0, 
        end: int = 9, 
        desc: bool = True
    ) -> List[Tuple[str, float]]:
        """
        리더보드 순위 조회
        
        Args:
            leaderboard: 리더보드 이름
            start: 시작 인덱스 (0부터 시작)
            end: 종료 인덱스
            desc: 내림차순 여부
            
        Returns:
            List[Tuple[str, float]]: (사용자, 점수) 목록
        """
        try:
            # 내림차순 또는 오름차순
            if desc:
                result = self.redis.zrevrange(leaderboard, start, end, withscores=True)
            else:
                result = self.redis.zrange(leaderboard, start, end, withscores=True)
                
            return result
        except redis.RedisError as e:
            self.logger.error(f"리더보드 조회 실패: {e}")
            return []
    
    def get_rank(self, leaderboard: str, user: str, desc: bool = True) -> Optional[int]:
        """
        리더보드 순위 조회
        
        Args:
            leaderboard: 리더보드 이름
            user: 사용자 식별자
            desc: 내림차순 여부
            
        Returns:
            Optional[int]: 0부터 시작하는 순위 또는 None
        """
        try:
            # 내림차순 또는 오름차순
            if desc:
                rank = self.redis.zrevrank(leaderboard, user)
            else:
                rank = self.redis.zrank(leaderboard, user)
                
            return rank
        except redis.RedisError as e:
            self.logger.error(f"순위 조회 실패: {e}")
            return None
    
    def get_score(self, leaderboard: str, user: str) -> Optional[float]:
        """
        리더보드 점수 조회
        
        Args:
            leaderboard: 리더보드 이름
            user: 사용자 식별자
            
        Returns:
            Optional[float]: 점수 또는 None
        """
        try:
            return self.redis.zscore(leaderboard, user)
        except redis.RedisError as e:
            self.logger.error(f"점수 조회 실패: {e}")
            return None

✅ 특징:

메시지 큐 구현 (생산자-소비자 패턴)
분산 락을 통한 동시성 제어
슬라이딩 윈도우 속도 제한
리더보드 및 순위 시스템
Lua 스크립트를 통한 원자적 연산
파이프라인을 활용한 성능 최적화
컨텍스트 매니저 지원
분산 환경 지원
효율적인 만료 시간 관리
견고한 오류 처리

6️⃣ 클러스터링과 고가용성

Redis는 단일 서버 구성을 넘어 클러스터링과 복제 기능을 통해 확장성과 고가용성을 제공한다. 대규모 데이터와 트래픽 처리, 장애 조치 및 데이터 지속성을 위한 다양한 구성이 가능하다.

from rediscluster import RedisCluster
import redis.sentinel

class RedisHighAvailability:
    """Redis 클러스터링 및 고가용성 구성 클래스"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.cluster = None
        self.master = None
    
    def connect_cluster(self, startup_nodes, decode_responses=True):
        """Redis 클러스터 연결"""
        try:
            self.cluster = RedisCluster(
                startup_nodes=startup_nodes,
                decode_responses=decode_responses,
                skip_full_coverage_check=True
            )
            self.logger.info("Redis 클러스터 연결 성공")
            return True
        except Exception as e:
            self.logger.error(f"Redis 클러스터 연결 실패: {e}")
            return False
    
    def connect_sentinel(self, sentinel_list, master_name, password=None):
        """Redis Sentinel 연결 (마스터-슬레이브 구성)"""
        try:
            sentinel = redis.sentinel.Sentinel(
                sentinel_list,
                socket_timeout=1.0,
                password=password
            )
            
            # 마스터 연결
            self.master = sentinel.master_for(
                master_name,
                socket_timeout=0.5,
                password=password,
                decode_responses=True
            )
            
            # 슬레이브 연결 (읽기 전용)
            self.slave = sentinel.slave_for(
                master_name,
                socket_timeout=0.5,
                password=password,
                decode_responses=True
            )
            
            self.logger.info(f"Redis Sentinel 연결 성공 (마스터: {master_name})")
            return True
        except Exception as e:
            self.logger.error(f"Redis Sentinel 연결 실패: {e}")
            return False
    
    def get_cluster_info(self):
        """클러스터 정보 조회"""
        if not self.cluster:
            return None
            
        try:
            nodes = self.cluster.cluster_nodes()
            slots = self.cluster.cluster_slots()
            info = self.cluster.cluster_info()
            
            return {
                "nodes": nodes,
                "slots": slots,
                "info": info
            }
        except Exception as e:
            self.logger.error(f"클러스터 정보 조회 실패: {e}")
            return None
    
    def write_to_master(self, key, value):
        """마스터에 데이터 쓰기"""
        if not self.master:
            return False
            
        try:
            return self.master.set(key, value)
        except Exception as e:
            self.logger.error(f"마스터 쓰기 실패: {e}")
            return False
    
    def read_from_slave(self, key):
        """슬레이브에서 데이터 읽기"""
        if not hasattr(self, 'slave') or not self.slave:
            return None
            
        try:
            return self.slave.get(key)
        except Exception as e:
            self.logger.error(f"슬레이브 읽기 실패: {e}")
            return None

class RedisPubSub:
    """Redis를 활용한 발행/구독 구현 클래스"""
    
    def __init__(self, client):
        self.redis = client.redis
        self.pubsub = self.redis.pubsub()
        self.logger = logging.getLogger(__name__)
        self.thread = None
    
    def subscribe(self, channels):
        """채널 구독"""
        try:
            self.pubsub.subscribe(*channels)
            self.logger.info(f"채널 구독: {channels}")
            return True
        except Exception as e:
            self.logger.error(f"채널 구독 실패: {e}")
            return False
    
    def pattern_subscribe(self, patterns):
        """패턴 구독"""
        try:
            self.pubsub.psubscribe(*patterns)
            self.logger.info(f"패턴 구독: {patterns}")
            return True
        except Exception as e:
            self.logger.error(f"패턴 구독 실패: {e}")
            return False
    
    def publish(self, channel, message):
        """메시지 발행"""
        try:
            receivers = self.redis.publish(channel, json.dumps(message))
            self.logger.debug(f"메시지 발행: {channel}, 수신자: {receivers}")
            return receivers
        except Exception as e:
            self.logger.error(f"메시지 발행 실패: {e}")
            return 0
    
    def start_listening(self, callback, sleep_time=0.1):
        """백그라운드에서 메시지 수신 시작"""
        self.thread = self.pubsub.run_in_thread(
            sleep_time=sleep_time,
            callback=callback
        )
        return self.thread
    
    def stop_listening(self):
        """메시지 수신 중지"""
        if self.thread:
            self.thread.stop()
            self.thread = None
    
    def get_message(self, ignore_subscribe_messages=True):
        """단일 메시지 수신"""
        return self.pubsub.get_message(
            ignore_subscribe_messages=ignore_subscribe_messages
        )
    
    def close(self):
        """구독 종료 및 자원 정리"""
        self.stop_listening()
        self.pubsub.unsubscribe()
        self.pubsub.punsubscribe()
        self.pubsub.close()

# 클러스터 설정 예제
def setup_redis_cluster():
    """Redis 클러스터 설정 및 연결 예제"""
    startup_nodes = [
        {"host": "redis-node1", "port": 7000},
        {"host": "redis-node2", "port": 7001},
        {"host": "redis-node3", "port": 7002}
    ]
    
    ha_manager = RedisHighAvailability()
    
    if ha_manager.connect_cluster(startup_nodes):
        print("클러스터 연결 성공")
        
        # 클러스터 정보 출력
        cluster_info = ha_manager.get_cluster_info()
        if cluster_info:
            print(f"클러스터 상태: {cluster_info['info'].get('cluster_state')}")
            print(f"노드 수: {len(cluster_info['nodes'])}")
    else:
        print("클러스터 연결 실패")

✅ 특징:

다양한 고가용성 구성 지원
클러스터 모드: 수평 확장 및 데이터 분산
Sentinel 모드: 자동 장애 조치 및 마스터-슬레이브 구성
클러스터 상태 모니터링 기능
읽기/쓰기 분리로 부하 분산 가능
발행/구독(Pub/Sub) 패턴 구현
실시간 메시지 처리 및 이벤트 기반 아키텍처
멀티채널 및 패턴 기반 구독
백그라운드 스레드로 비동기 메시지 처리
자원 안전한 정리 및 종료 처리

주요 팁

✅ 모범 사례:

키 설계: 의미 있는 접두사와 구분자를 사용하여 체계적인 키 구조 설계

user:1000:profile - 사용자 프로필
user:1000:posts - 사용자 게시물
post:5000:comments - 게시물 댓글

메모리 최적화: 메모리 사용량 최소화를 위한 데이터 구조 선택
- 작은 해시는 문자열보다 효율적 (ziplist 인코딩)
- 숫자는 정수로 저장 (int 인코딩)
- 필요한 경우에만 TTL 설정

성능 모니터링: 주요 지표 추적으로 성능 문제 조기 발견

# 메모리, 클라이언트 연결, 명령 수 등 모니터링
info = redis.info()

# 느린 로그 설정 및 확인
redis.config_set('slowlog-log-slower-than', 10000)  # 10ms
slow_logs = redis.slowlog_get(10)

백업 전략: 데이터 손실 방지를 위한 RDB 및 AOF 설정

# redis.conf 설정 예
save 900 1
appendonly yes
appendfsync everysec

보안 강화: 인증, 명령어 제한, 네트워크 접근 제어

# redis.conf 보안 설정
requirepass StrongPassword
rename-command FLUSHALL ""  # 위험 명령어 비활성화
bind 127.0.0.1  # 로컬 접근만 허용

효율적인 파이프라인 사용: 네트워크 지연 감소를 위한 명령 배치 처리

# 대량 데이터 처리 시 파이프라인 활용
with redis.pipeline() as pipe:
    for i in range(10000):
        pipe.set(f"key:{i}", f"value:{i}")
    pipe.execute()

클라이언트 연결 관리: 커넥션 풀 사용으로 자원 효율화

pool = redis.ConnectionPool(max_connections=10)
client = redis.Redis(connection_pool=pool)

Lua 스크립트 활용: 복잡한 작업의 원자적 보장

# Lua 스크립트로 원자적 연산 수행
script = """
local current = redis.call('GET', KEYS[1])
if current then
    return redis.call('SET', KEYS[1], tonumber(current) + tonumber(ARGV[1]))
end
return nil
"""
redis.eval(script, 1, 'counter', 5)

전문 기능 활용: Redis 모듈 및 확장 기능 활용
- RedisSearch: 전문 검색 엔진
- RedisTimeSeries: 시계열 데이터 처리
- RedisGears: 서버측 데이터 처리 엔진
- RedisAI: 머신러닝 모델 호스팅
확장성 계획: 처음부터 확장성을 고려한 설계
- 샤딩 키 식별
- 클러스터 호환 명령어 사용
- 다중 데이터베이스 대신 키 접두사 사용


<br/>

---

KR_Redis - somaz94/python-study GitHub Wiki

Python Redis 개념 정리

1️⃣ Redis 기초

2️⃣ 기본 작업

3️⃣ 캐싱

4️⃣ 세션 관리

5️⃣ 고급 패턴

6️⃣ 클러스터링과 고가용성

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_Redis - somaz94/python-study GitHub Wiki

Python Redis 개념 정리

1️⃣ Redis 기초

2️⃣ 기본 작업

3️⃣ 캐싱

4️⃣ 세션 관리

5️⃣ 고급 패턴

6️⃣ 클러스터링과 고가용성

주요 팁

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️