redis-patterns

Redis 패턴

일반적인 백엔드 사용 사례에 대한 Redis 모범 사례 빠른 참조.

작동 방식

Redis는 문자열, 해시, 리스트, 셋, 정렬된 셋, 스트림 등을 지원하는 인메모리 데이터 구조 저장소입니다. 단일 인스턴스에서 개별 Redis 명령은 원자적입니다. 여러 단계의 워크플로우는 원자성을 유지하기 위해 Lua 스크립트, MULTI/EXEC 트랜잭션 또는 명시적 동기화가 필요합니다. 데이터는 선택적으로 RDB 스냅샷 또는 AOF 로그를 통해 영속화됩니다. 클라이언트는 TCP를 통해 RESP 프로토콜로 통신하며, 요청별 핸드셰이크 오버헤드를 피하기 위해 연결 풀이 필수적입니다.

활성화 시점

• 애플리케이션에 캐싱 추가
• 속도 제한 또는 스로틀링 구현
• 분산 락 또는 조정 구축
• 세션 또는 토큰 저장소 설정
• 메시징을 위해 Pub/Sub 또는 Redis Streams 사용
• 프로덕션에서 Redis 구성(풀링, 제거, 클러스터링)

데이터 구조 치트 시트

사용 사례	구조	예시 키
간단한 캐시	String	product:123
사용자 세션	Hash	session:abc
리더보드	Sorted Set	scores:weekly
고유 방문자	Set	visitors:2024-01-01
활동 피드	List	feed:user:456
이벤트 스트림	Stream	events:orders
카운터 / 속도 제한	String (INCR)	ratelimit:user:123
블룸 필터 / HLL	HyperLogLog	hll:pageviews

핵심 패턴

캐시 어사이드 (지연 로딩)

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_product(product_id: int):
    cache_key = f"product:{product_id}"
    cached = r.get(cache_key)

    if cached:
        return json.loads(cached)

    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.setex(cache_key, 3600, json.dumps(product))  # TTL: 1시간
    return product

라이트 스루 캐시

def update_product(product_id: int, data: dict):
    # 먼저 DB에 쓰기
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

    # 즉시 캐시 업데이트
    cache_key = f"product:{product_id}"
    r.setex(cache_key, 3600, json.dumps(data))

캐시 무효화

# 태그 기반 무효화 — 관련 키를 셋으로 그룹화
def cache_product(product_id: int, category_id: int, data: dict):
    key = f"product:{product_id}"
    tag = f"tag:category:{category_id}"
    pipe = r.pipeline(transaction=True)
    pipe.setex(key, 3600, json.dumps(data))
    pipe.sadd(tag, key)
    pipe.expire(tag, 3600)
    pipe.execute()

def invalidate_category(category_id: int):
    tag = f"tag:category:{category_id}"
    keys = r.smembers(tag)
    if keys:
        r.delete(*keys)
    r.delete(tag)

세션 저장소

import time
import uuid

def create_session(user_id: int, ttl: int = 86400) -> str:
    session_id = str(uuid.uuid4())
    key = f"session:{session_id}"
    pipe = r.pipeline(transaction=True)
    pipe.hset(key, mapping={
        "user_id": user_id,
        "created_at": int(time.time()),
    })
    pipe.expire(key, ttl)
    pipe.execute()
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    return data if data else None

def delete_session(session_id: str):
    r.delete(f"session:{session_id}")

속도 제한

고정 윈도우 (간단)

def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window}"
    pipe = r.pipeline(transaction=True)
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()
    return count > limit

슬라이딩 윈도우 (Lua — 원자적)

-- sliding_window.lua
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    -- 동일한 밀리초 내 충돌을 피하기 위해 고유 멤버(now + 시퀀스) 사용
    local seq_key = key .. ':seq'
    local seq = redis.call('INCR', seq_key)
    redis.call('EXPIRE', seq_key, math.ceil(window / 1000))
    redis.call('ZADD', key, now, now .. '-' .. seq)
    redis.call('EXPIRE', key, math.ceil(window / 1000))
    return 1
end
return 0

sliding_window = r.register_script(open('sliding_window.lua').read())

def allow_request(user_id: int) -> bool:
    key = f"ratelimit:sliding:{user_id}"
    now = int(time.time() * 1000)
    return bool(sliding_window(keys=[key], args=[now, 60000, 100]))

분산 락

분산 락 (단일 노드 — SET NX PX)

import uuid

def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:
    lock_key = f"lock:{resource}"
    token = str(uuid.uuid4())
    acquired = r.set(lock_key, token, px=ttl_ms, nx=True)
    return token if acquired else None

def release_lock(resource: str, token: str) -> bool:
    release_script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(release_script, 1, f"lock:{resource}", token)
    return bool(result)

# 사용법
token = acquire_lock("order:payment:123")
if token:
    try:
        process_payment()
    finally:
        release_lock("order:payment:123", token)

다중 노드 설정의 경우 전체 Redlock 알고리즘을 구현하는 redlock-py 라이브러리를 사용하세요.

Pub/Sub 및 스트림

Pub/Sub (발송 후 확인)

# 발행자
def publish_event(channel: str, payload: dict):
    r.publish(channel, json.dumps(payload))

# 구독자 (블로킹 — 별도 스레드/프로세스에서 실행)
def subscribe_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            handle(json.loads(message['data']))

Redis Streams (내구성 있는 큐)

# 생산자
def emit(stream: str, event: dict):
    r.xadd(stream, event, maxlen=10000)  # 스트림 길이 제한

# 소비자 그룹 — 최소 한 번 전달 보장
try:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
except Exception:
    pass  # 그룹이 이미 존재함

def consume(stream: str, group: str, consumer: str):
    while True:
        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)
        for _, entries in (messages or []):
            for msg_id, data in entries:
                process(data)
                r.xack(stream, group, msg_id)

전달 보장, 소비자 그룹 또는 재생이 필요한 경우 Pub/Sub보다 Streams를 선호하세요.

키 설계

명명 규칙

# 패턴: resource:id:field
user:123:profile
order:456:status
cache:product:789

# 패턴: namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# 패턴: resource:date (시간 범위 키)
stats:pageviews:2024-01-01

TTL 전략

데이터 유형	권장 TTL
사용자 세션	24시간 (86400)
API 응답 캐시	5~15분
속도 제한 윈도우	윈도우 크기와 일치
단기 토큰	5~10분
리더보드	1시간~24시간
정적/참조 데이터	1시간~1주일

항상 TTL을 설정하세요. TTL이 없는 키는 무기한 축적되어 메모리 압력을 유발합니다.

연결 관리

연결 풀링

from redis import ConnectionPool, Redis

pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=20,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)

r = Redis(connection_pool=pool)

클러스터 모드

from redis.cluster import RedisCluster

r = RedisCluster(
    startup_nodes=[{"host": "redis-1", "port": 6379}],
    decode_responses=True,
    skip_full_coverage_check=True,
)

센티널 (고가용성)

from redis.sentinel import Sentinel

sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379)],
    socket_timeout=0.5,
)
master = sentinel.master_for('mymaster', decode_responses=True)
replica = sentinel.slave_for('mymaster', decode_responses=True)

제거 정책

정책	동작	최적 사용 사례
noeviction	가득 차면 쓰기 오류	큐 / 중요한 데이터
allkeys-lru	가장 최근에 사용되지 않은 항목 제거	일반 캐시
volatile-lru	TTL이 있는 키 중에서만 LRU	혼합 데이터 저장소
allkeys-lfu	가장 자주 사용되지 않은 항목 제거	편향된 액세스 패턴
volatile-ttl	가장 빨리 만료될 항목 제거	장수명 데이터 우선

redis.conf에서 설정: maxmemory-policy allkeys-lru

안티패턴

안티패턴	문제점	해결책
TTL이 없는 키	메모리가 무한히 증가	항상 TTL 설정
프로덕션에서 KEYS * 사용	서버 차단 (O(N))	SCAN 커서 사용
대용량 Blob 저장 (>100KB)	느린 직렬화, 메모리 압력	참조 저장 후 객체 저장소에서 가져오기
모든 용도로 단일 Redis 사용	캐시와 큐 간 격리 부족	별도 DB 또는 인스턴스 사용
연결 풀 제한 무시	부하 시 연결 소진	워크로드에 맞게 풀 크기 조정
캐시 미스 스탬피드 처리 안 함	콜드 스타트 시 떼 로드	락 또는 확률적 조기 만료 사용
생각 없이 FLUSHALL 사용	전체 인스턴스 초기화	키 패턴별로 삭제 범위 지정

캐시 미스 스탬피드 방지

import threading

_locks: dict[str, threading.Lock] = {}
_locks_mutex = threading.Lock()

def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    with _locks_mutex:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]
    with lock:
        cached = r.get(key)  # 락 획득 후 재확인
        if cached:
            return json.loads(cached)
        value = fetch_fn()
        r.setex(key, ttl, json.dumps(value))
        return value

참고: 다중 프로세스 배포의 경우, 프로세스 내 락을 위의 분산 락 섹션에 있는 acquire_lock/release_lock으로 대체하세요.

예제

Django/Flask API 엔드포인트에 캐싱 추가: setex와 5분 TTL을 사용한 캐시 어사이드를 사용합니다. 요청 매개변수를 키로 지정합니다.

사용자별 API 속도 제한: 트래픽이 적은 엔드포인트에는 pipeline(transaction=True)을 사용한 고정 윈도우를, 정확한 사용자별 스로틀링에는 슬라이딩 윈도우 Lua를 사용합니다.

작업자 간 백그라운드 작업 조정: 예상 작업 기간을 초과하는 TTL로 acquire_lock을 사용합니다. 항상 finally 블록에서 락을 해제합니다.

여러 구독자에게 알림 전파: 발송 후 확인에는 Pub/Sub을 사용합니다. 늦은 소비자를 위한 전달 보장 또는 재생이 필요하면 Streams로 전환합니다.

빠른 참조

패턴	사용 시점
캐시 어사이드	읽기 중심, 약간의 지연 허용
라이트 스루	강력한 일관성 필요
분산 락	리소스에 대한 동시 액세스 방지
슬라이딩 윈도우 속도 제한	정확한 사용자별 스로틀링
Redis Streams	소비자 그룹이 있는 내구성 있는 이벤트 큐
Pub/Sub	전달 보장이 필요 없는 브로드캐스트
정렬된 셋 리더보드	순위 점수 매기기, 페이지네이션
HyperLogLog	낮은 메모리로 대략적인 고유 개수

관련 항목

• 스킬: postgres-patterns — 관계형 데이터 패턴
• 스킬: backend-patterns — API 및 서비스 계층 패턴
• 스킬: database-migrations — 스키마 버전 관리
• 스킬: django-patterns — Django 캐시 프레임워크 통합
• 에이전트: database-reviewer — 전체 데이터베이스 검토 워크플로우