Python Memcached 개념 정리

1️⃣ 기본 연결

Memcached 서버와의 기본적인 연결을 설정하는 방법이다.

from pymemcache.client.base import Client
from pymemcache.client.hash import HashClient
from typing import Any, Optional, Dict, List, Tuple
import json
import pickle
import socket

class MemcachedClient:
    def __init__(
        self,
        host: str = 'localhost',
        port: int = 11211,
        connect_timeout: float = 1.0,
        timeout: float = 0.5
    ):
        """Memcached 클라이언트 초기화
        
        Args:
            host: Memcached 서버 호스트
            port: Memcached 서버 포트
            connect_timeout: 연결 시간 제한(초)
            timeout: 작업 시간 제한(초)
        """
        self.client = Client(
            (host, port),
            connect_timeout=connect_timeout,
            timeout=timeout
        )
    
    def ping(self) -> bool:
        """서버 연결 테스트"""
        try:
            return self.client.stats() is not None
        except (socket.timeout, ConnectionRefusedError) as e:
            print(f"서버 연결 실패: {e}")
            return False
        except Exception as e:
            print(f"기타 오류: {e}")
            return False
    
    def close(self):
        """연결 종료"""
        try:
            self.client.close()
            print("Memcached 연결이 정상적으로 종료되었습니다")
        except Exception as e:
            print(f"연결 종료 중 오류 발생: {e}")
    
    def get_stats(self) -> Dict:
        """서버 통계 정보 조회"""
        try:
            return self.client.stats()
        except Exception as e:
            print(f"통계 정보 조회 오류: {e}")
            return {}
    
    def flush_all(self) -> bool:
        """모든 캐시 데이터 삭제"""
        try:
            return self.client.flush_all()
        except Exception as e:
            print(f"캐시 삭제 오류: {e}")
            return False

✅ 특징:

서버 연결 관리
상태 확인
연결 종료 처리
타임아웃 설정
오류 처리
서버 통계 정보 조회

Memcached란?

Memcached는 고성능 분산 메모리 캐싱 시스템으로 웹 애플리케이션의 성능을 향상시키는 데 사용된다.

인메모리 캐시: 데이터베이스 부하를 줄이기 위해 자주 사용되는 데이터를 메모리에 저장
분산 아키텍처: 여러 서버에 걸쳐 확장 가능한 구조
간단한 키-값 저장소: 복잡한 연산 없이 빠른 액세스 제공
LRU(Least Recently Used): 메모리가 부족할 때 가장 오래된 항목부터 제거
원자적 연산: 다중 스레드/프로세스 환경에서 안전한 데이터 액세스

2️⃣ 기본 작업

Memcached에 데이터를 저장하고 조회하는 기본 작업 방법이다.

class MemcachedOperations:
    def __init__(self, client: MemcachedClient):
        self.client = client.client
    
    def set_value(
        self,
        key: str,
        value: Any,
        expire: int = 0,
        noreply: bool = False
    ) -> bool:
        """값 설정
        
        Args:
            key: 저장할 키
            value: 저장할 값
            expire: 만료 시간(초), 0은 만료 없음
            noreply: 서버 응답을 기다리지 않음
        
        Returns:
            bool: 성공 여부
        """
        try:
            return self.client.set(
                key,
                pickle.dumps(value),
                expire=expire,
                noreply=noreply
            )
        except Exception as e:
            print(f"값 설정 오류: {e}")
            return False
    
    def get_value(self, key: str) -> Optional[Any]:
        """값 조회
        
        Args:
            key: 조회할 키
            
        Returns:
            Optional[Any]: 저장된 값 또는 None
        """
        try:
            value = self.client.get(key)
            return pickle.loads(value) if value else None
        except Exception as e:
            print(f"값 조회 오류: {e}")
            return None
    
    def delete(self, key: str, noreply: bool = False) -> bool:
        """값 삭제
        
        Args:
            key: 삭제할 키
            noreply: 서버 응답을 기다리지 않음
            
        Returns:
            bool: 성공 여부
        """
        try:
            return self.client.delete(key, noreply=noreply)
        except Exception as e:
            print(f"값 삭제 오류: {e}")
            return False
    
    def add(
        self,
        key: str,
        value: Any,
        expire: int = 0
    ) -> bool:
        """키가 존재하지 않을 때만 값 추가
        
        Args:
            key: 추가할 키
            value: 추가할 값
            expire: 만료 시간(초)
            
        Returns:
            bool: 성공 여부
        """
        try:
            return self.client.add(
                key,
                pickle.dumps(value),
                expire=expire
            )
        except Exception as e:
            print(f"값 추가 오류: {e}")
            return False
    
    def incr(self, key: str, value: int = 1) -> Optional[int]:
        """정수 값 증가
        
        Args:
            key: 증가시킬 키
            value: 증가량
            
        Returns:
            Optional[int]: 증가 후 값 또는 None
        """
        try:
            return self.client.incr(key, value)
        except Exception as e:
            print(f"값 증가 오류: {e}")
            return None
    
    def decr(self, key: str, value: int = 1) -> Optional[int]:
        """정수 값 감소
        
        Args:
            key: 감소시킬 키
            value: 감소량
            
        Returns:
            Optional[int]: 감소 후 값 또는 None
        """
        try:
            return self.client.decr(key, value)
        except Exception as e:
            print(f"값 감소 오류: {e}")
            return None
    
    def get_multi(self, keys: List[str]) -> Dict[str, Any]:
        """여러 키 값 동시 조회
        
        Args:
            keys: 조회할 키 목록
            
        Returns:
            Dict[str, Any]: 키-값 쌍 딕셔너리
        """
        try:
            values = self.client.get_multi(keys)
            return {k: pickle.loads(v) if v else None for k, v in values.items()}
        except Exception as e:
            print(f"다중 값 조회 오류: {e}")
            return {}

✅ 특징:

데이터 저장/조회
직렬화 처리
에러 처리
원자적 증가/감소
다중 키 조회
조건부 추가

Memcached 작업 타입:

Memcached에서 지원하는 다양한 작업 타입과 특징이다.

작업	설명	특징
set	값 설정	키가 존재하면 덮어씀
get	값 조회	없으면 None 반환
delete	값 삭제	성공 여부 반환
add	새 값 추가	키가 존재하면 실패
replace	값 대체	키가 존재해야 성공
incr	값 증가	정수 값만 가능
decr	값 감소	정수 값만 가능
append	값 뒤에 추가	기존 값 필요
prepend	값 앞에 추가	기존 값 필요
get_multi	다중 키 조회	네트워크 최적화

3️⃣ 캐싱 구현

함수 결과를 효율적으로 캐싱하는 데코레이터 패턴 구현 방법이다.

from functools import wraps
import hashlib
import time
import inspect
import json

class MemcachedCache:
    def __init__(self, client: MemcachedClient):
        self.ops = MemcachedOperations(client)
    
    def _generate_cache_key(self, func, args, kwargs):
        """함수 호출에 대한 고유 캐시 키 생성"""
        # 함수 이름과 모듈 포함
        key_parts = [func.__module__, func.__name__]
        
        # 위치 인자 추가
        for arg in args:
            key_parts.append(str(arg))
        
        # 키워드 인자 추가 (정렬하여 일관성 보장)
        sorted_kwargs = sorted(kwargs.items())
        for k, v in sorted_kwargs:
            key_parts.append(f"{k}={v}")
        
        # 키 문자열 생성 및 해시
        key_str = ":".join(key_parts)
        return f"cache:{hashlib.md5(key_str.encode()).hexdigest()}"
    
    def cache_decorator(self, expire: int = 300):
        """함수 결과를 캐시하는 데코레이터
        
        Args:
            expire: 캐시 만료 시간(초)
            
        Returns:
            Callable: 데코레이터 함수
        """
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                # 캐시 키 생성
                key = self._generate_cache_key(func, args, kwargs)
                
                # 캐시된 결과 확인
                cached_result = self.ops.get_value(key)
                if cached_result is not None:
                    return cached_result
                
                # 캐시 미스: 함수 실행 및 결과 캐싱
                start_time = time.time()
                result = func(*args, **kwargs)
                exec_time = time.time() - start_time
                
                # 결과 캐싱
                self.ops.set_value(key, result, expire)
                
                print(f"캐시 저장: {key} (실행 시간: {exec_time:.4f}초)")
                return result
            return wrapper
        return decorator
    
    def invalidate_cache(self, func, *args, **kwargs):
        """특정 함수 호출에 대한 캐시 무효화
        
        Args:
            func: 캐시를 무효화할 함수
            *args, **kwargs: 함수 인자
            
        Returns:
            bool: 성공 여부
        """
        key = self._generate_cache_key(func, args, kwargs)
        return self.ops.delete(key)
    
    def memoize(self, expire: int = 300):
        """메서드 결과를 메모이제이션하는 데코레이터
        
        Args:
            expire: 캐시 만료 시간(초)
            
        Returns:
            Callable: 데코레이터 함수
        """
        return self.cache_decorator(expire)

✅ 특징:

데코레이터 패턴
키 생성 로직
만료 시간 관리
캐시 무효화
실행 시간 측정
메모이제이션

캐싱 활용 사례:

Memcached 캐싱이 효과적인 일반적인 사용 사례이다.

데이터베이스 쿼리 결과: 자주 실행되는 쿼리의 결과 캐싱
API 응답: 외부 API 호출 결과 캐싱으로 API 제한 및 지연 시간 감소
세션 데이터: 웹 애플리케이션의 사용자 세션 정보 저장
페이지 렌더링: 동적 웹 페이지의 HTML 출력 캐싱
계산 비용이 높은 작업: 복잡한 연산 결과 캐싱
접근 제어 토큰: 임시 액세스 토큰 및 인증 정보 저장

4️⃣ 세션 관리

Memcached를 사용한 웹 애플리케이션 세션 관리 구현 방법이다.

import uuid
from datetime import datetime, timedelta
import json
from typing import Optional, Dict, Any

class MemcachedSession:
    def __init__(self, client: MemcachedClient):
        self.ops = MemcachedOperations(client)
        self.session_prefix = "session:"
    
    def create_session(
        self,
        user_id: str,
        data: dict,
        expire: int = 3600
    ) -> Optional[str]:
        """새 세션 생성
        
        Args:
            user_id: 사용자 ID
            data: 세션에 저장할 데이터
            expire: 세션 만료 시간(초)
            
        Returns:
            Optional[str]: 생성된 세션 ID 또는 None
        """
        session_id = str(uuid.uuid4())
        session_key = f"{self.session_prefix}{session_id}"
        
        # 세션 데이터 구성
        session_data = {
            "user_id": user_id,
            "created_at": datetime.now().isoformat(),
            "last_accessed": datetime.now().isoformat(),
            "expire_at": (datetime.now() + timedelta(seconds=expire)).isoformat(),
            "data": data
        }
        
        # 세션 저장
        if self.ops.set_value(session_key, session_data, expire):
            return session_id
        return None
    
    def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
        """세션 조회
        
        Args:
            session_id: 세션 ID
            
        Returns:
            Optional[Dict[str, Any]]: 세션 데이터 또는 None
        """
        session_key = f"{self.session_prefix}{session_id}"
        session_data = self.ops.get_value(session_key)
        
        if session_data:
            # 마지막 접근 시간 업데이트
            session_data["last_accessed"] = datetime.now().isoformat()
            self.ops.set_value(session_key, session_data)
        
        return session_data
    
    def update_session(
        self,
        session_id: str,
        data: Dict[str, Any],
        extend_expire: bool = True,
        expire: int = 3600
    ) -> bool:
        """세션 데이터 업데이트
        
        Args:
            session_id: 세션 ID
            data: 업데이트할 데이터
            extend_expire: 만료 시간 연장 여부
            expire: 새 만료 시간(초)
            
        Returns:
            bool: 성공 여부
        """
        session_key = f"{self.session_prefix}{session_id}"
        session_data = self.ops.get_value(session_key)
        
        if not session_data:
            return False
        
        # 세션 데이터 업데이트
        session_data["data"].update(data)
        session_data["last_accessed"] = datetime.now().isoformat()
        
        if extend_expire:
            session_data["expire_at"] = (datetime.now() + timedelta(seconds=expire)).isoformat()
            return self.ops.set_value(session_key, session_data, expire)
        else:
            return self.ops.set_value(session_key, session_data)
    
    def delete_session(self, session_id: str) -> bool:
        """세션 삭제
        
        Args:
            session_id: 세션 ID
            
        Returns:
            bool: 성공 여부
        """
        session_key = f"{self.session_prefix}{session_id}"
        return self.ops.delete(session_key)
    
    def is_valid_session(self, session_id: str) -> bool:
        """세션 유효성 확인
        
        Args:
            session_id: 세션 ID
            
        Returns:
            bool: 유효 여부
        """
        session_data = self.get_session(session_id)
        return session_data is not None

✅ 특징:

세션 식별자 생성
만료 시간 설정
세션 데이터 관리
자동 갱신
유효성 검증
사용자 연결

5️⃣ 분산 캐시

여러 Memcached 서버를 관리하는 분산 캐시 구현 방법이다.

from typing import List, Tuple, Dict, Any, Optional
import hashlib

class DistributedCache:
    def __init__(self, servers: List[Tuple[str, int]]):
        """여러 Memcached 서버 연결
        
        Args:
            servers: (호스트, 포트) 형태의 서버 목록
        """
        self.clients = [
            MemcachedClient(host, port)
            for host, port in servers
        ]
        self.n_servers = len(servers)
        self.server_info = servers
    
    def _get_client(self, key: str) -> MemcachedOperations:
        """해시 기반으로 서버 선택
        
        Args:
            key: 키
            
        Returns:
            MemcachedOperations: 선택된 서버의 작업 객체
        """
        hash_value = int(hashlib.md5(key.encode()).hexdigest(), 16)
        server_index = hash_value % self.n_servers
        return MemcachedOperations(self.clients[server_index])
    
    def set(self, key: str, value: Any, expire: int = 0) -> bool:
        """분산 환경에서 값 설정
        
        Args:
            key: 키
            value: 값
            expire: 만료 시간(초)
            
        Returns:
            bool: 성공 여부
        """
        client = self._get_client(key)
        return client.set_value(key, value, expire)
    
    def get(self, key: str) -> Optional[Any]:
        """분산 환경에서 값 조회
        
        Args:
            key: 키
            
        Returns:
            Optional[Any]: 값 또는 None
        """
        client = self._get_client(key)
        return client.get_value(key)
    
    def delete(self, key: str) -> bool:
        """분산 환경에서 값 삭제
        
        Args:
            key: 키
            
        Returns:
            bool: 성공 여부
        """
        client = self._get_client(key)
        return client.delete(key)
    
    def get_stats_all(self) -> Dict[Tuple[str, int], Dict]:
        """모든 서버의 통계 정보 조회
        
        Returns:
            Dict: 서버별 통계 정보
        """
        stats = {}
        for i, client in enumerate(self.clients):
            stats[self.server_info[i]] = client.get_stats()
        return stats
    
    def close_all(self):
        """모든 서버 연결 종료"""
        for client in self.clients:
            client.close()

✅ 특징:

다중 서버 관리
해시 기반 분산
확장성 지원
서버 선택 알고리즘
통합 인터페이스
리소스 관리

분산 캐시 알고리즘:

Memcached에서 사용되는 주요 분산 알고리즘과 특징이다.

일관된 해싱(Consistent Hashing): 서버 추가/제거 시 최소한의 키 재배치
모듈러 해싱(Modular Hashing): 키 해시를 서버 수로 나눈 나머지로 서버 결정
가중치 기반 배포: 서버 용량에 비례하여 키 배포
레플리카(Replica): 동일한 키를 여러 서버에 복제하여 가용성 향상
샤딩(Sharding): 데이터를 분할하여 여러 서버에 분산

주요 팁

✅ 모범 사례:

키 네이밍 전략 수립: 관련 데이터를 그룹화하는 일관된 키 접두사 사용
만료 시간 적절히 설정: 데이터 유형과 갱신 빈도에 맞게 설정
메모리 사용량 모니터링: 과도한 메모리 사용으로 인한 항목 제거 방지
분산 처리 고려: 일관된 해싱을 통한 효율적인 서버 확장
에러 처리 구현: 네트워크 오류와 서버 다운에 대한 대비
성능 최적화: 직렬화/역직렬화 비용 최소화 및 배치 작업 활용
백업 전략 수립: 주요 데이터에 대한 백업 메커니즘 구현
보안 설정 확인: 방화벽 설정 및 인증 메커니즘 활성화
대용량 객체 분할: 큰 객체는 여러 조각으로 나누어 저장
핫키 문제 해결: 자주 접근하는 키의 부하 분산 전략 수립

KR_Memcached - somaz94/python-study GitHub Wiki

Python Memcached 개념 정리

1️⃣ 기본 연결

Memcached란?

2️⃣ 기본 작업

Memcached 작업 타입:

3️⃣ 캐싱 구현

캐싱 활용 사례:

4️⃣ 세션 관리

5️⃣ 분산 캐시

분산 캐시 알고리즘:

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_Memcached - somaz94/python-study GitHub Wiki

Python Memcached 개념 정리

1️⃣ 기본 연결

Memcached란?

2️⃣ 기본 작업

Memcached 작업 타입:

3️⃣ 캐싱 구현

캐싱 활용 사례:

4️⃣ 세션 관리

5️⃣ 분산 캐시

분산 캐시 알고리즘:

주요 팁

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

⚠️ GitHub.com Fallback ⚠️