Redis 완전 가이드 2025: 캐싱 전략, 데이터 구조, Pub/Sub, 그리고 Redis Stack까지

# 기본 SET/GET
SET user:1:name "Alice"
GET user:1:name

# 숫자 증감 (원자적 연산)
SET page:views 0
INCR page:views          # 1
INCRBY page:views 10     # 11

# 만료 시간 설정
SET session:abc123 "user_data" EX 3600   # 1시간 후 만료
TTL session:abc123                        # 남은 시간 확인

# SET 옵션
SET lock:resource "owner1" NX EX 30      # NX: 키가 없을 때만 설정
SET user:1:name "Bob" XX                 # XX: 키가 있을 때만 업데이트

# 기본 연산
LPUSH queue:emails "email1" "email2" "email3"
RPOP queue:emails                        # "email1" (FIFO 큐)

# 범위 조회
LRANGE queue:emails 0 -1                 # 모든 요소

# 블로킹 팝 (메시지 큐로 활용)
BRPOP queue:emails 30                    # 30초 대기 후 팝

# 트리밍 (최근 N개 유지)
LPUSH notifications:user1 "new_msg"
LTRIM notifications:user1 0 99          # 최근 100개만 유지

# 기본 연산
SADD tags:post:1 "python" "redis" "backend"
SADD tags:post:2 "python" "django" "orm"

# 멤버십 확인
SISMEMBER tags:post:1 "python"           # 1 (true)

# 집합 연산
SINTER tags:post:1 tags:post:2           # "python" (교집합)
SUNION tags:post:1 tags:post:2           # 합집합
SDIFF tags:post:1 tags:post:2            # tags:post:1에만 있는 요소

# 랜덤 추출
SRANDMEMBER tags:post:1 2               # 랜덤 2개

# 리더보드 구현
ZADD leaderboard 1500 "player:alice"
ZADD leaderboard 2300 "player:bob"
ZADD leaderboard 1800 "player:charlie"

# 랭킹 조회 (높은 점수 순)
ZREVRANGE leaderboard 0 2 WITHSCORES
# 1) "player:bob"     2) "2300"
# 3) "player:charlie" 4) "1800"
# 5) "player:alice"   6) "1500"

# 특정 멤버 랭킹 (0부터 시작)
ZREVRANK leaderboard "player:alice"      # 2

# 점수 증가
ZINCRBY leaderboard 500 "player:alice"   # 2000

# 범위 검색
ZRANGEBYSCORE leaderboard 1500 2000 WITHSCORES

# 사용자 프로필 저장
HSET user:1 name "Alice" email "alice@example.com" age "30" role "admin"

# 개별 필드 조회
HGET user:1 name                         # "Alice"

# 전체 조회
HGETALL user:1

# 필드 증감
HINCRBY user:1 age 1                     # 31

# 존재 확인
HEXISTS user:1 email                     # 1 (true)

# 여러 필드 한 번에
HMGET user:1 name email role

# 고유 방문자 수 추정
PFADD visitors:2025-03-23 "user1" "user2" "user3"
PFADD visitors:2025-03-23 "user1" "user4"          # user1은 중복

PFCOUNT visitors:2025-03-23                         # 4

# 여러 날 합산
PFMERGE visitors:week visitors:2025-03-23 visitors:2025-03-24
PFCOUNT visitors:week

# 일일 출석 체크
SETBIT attendance:2025-03-23 1001 1      # 사용자 1001 출석
SETBIT attendance:2025-03-23 1002 1
SETBIT attendance:2025-03-23 1003 0      # 결석

# 출석 여부 확인
GETBIT attendance:2025-03-23 1001        # 1

# 출석 인원 수
BITCOUNT attendance:2025-03-23           # 2

# 연속 출석 (AND 연산)
BITOP AND consecutive attendance:2025-03-22 attendance:2025-03-23
BITCOUNT consecutive

# 위치 추가 (경도, 위도)
GEOADD stores 126.9784 37.5665 "gangnam-store"
GEOADD stores 127.0276 37.4979 "samsung-store"
GEOADD stores 126.9316 37.5563 "hongdae-store"

# 거리 계산
GEODIST stores "gangnam-store" "hongdae-store" km    # ~5.2 km

# 반경 검색 (Redis 6.2+)
GEOSEARCH stores FROMLONLAT 126.9784 37.5665 BYRADIUS 10 km ASC COUNT 5

# 메시지 추가
XADD events * type "order" user_id "123" amount "50000"
XADD events * type "payment" user_id "123" status "completed"

# 읽기
XRANGE events - + COUNT 10

# Consumer Group 생성
XGROUP CREATE events analytics-group $ MKSTREAM

# Consumer로 읽기
XREADGROUP GROUP analytics-group consumer-1 COUNT 5 BLOCK 2000 STREAMS events >

# ACK (처리 완료)
XACK events analytics-group "1679000000000-0"

# 미처리 메시지 확인
XPENDING events analytics-group

import redis
import json

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

def get_user(user_id: int) -> dict:
    cache_key = f"user:{user_id}"

    # 1. 캐시 확인
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 2. 캐시 미스 -> DB 조회
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        return None

    # 3. 캐시에 저장
    r.setex(cache_key, 3600, json.dumps(user.to_dict()))
    return user.to_dict()

def update_user(user_id: int, data: dict):
    # DB 업데이트
    db.query(User).filter(User.id == user_id).update(data)
    db.commit()

    # 캐시 무효화
    r.delete(f"user:{user_id}")

def save_user(user_id: int, data: dict):
    cache_key = f"user:{user_id}"

    # DB와 캐시 동시 업데이트
    db.query(User).filter(User.id == user_id).update(data)
    db.commit()

    r.setex(cache_key, 3600, json.dumps(data))

def save_user_async(user_id: int, data: dict):
    cache_key = f"user:{user_id}"

    # 캐시에 먼저 쓰기
    r.setex(cache_key, 3600, json.dumps(data))

    # 비동기로 DB에 반영 (Celery 등)
    sync_to_db_task.delay(user_id, data)

SET product:123 "data" EX 300           # 5분 후 만료

def update_product(product_id: int, data: dict):
    db.update(product_id, data)

    # 관련 캐시 모두 삭제
    r.delete(f"product:{product_id}")
    r.delete(f"product_list:category:{data['category_id']}")
    r.delete("product_list:featured")

def get_product_list(category_id: int) -> list:
    version = r.get(f"product_version:{category_id}") or "1"
    cache_key = f"products:cat:{category_id}:v:{version}"

    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    products = db.query(Product).filter(Product.category_id == category_id).all()
    r.setex(cache_key, 3600, json.dumps([p.to_dict() for p in products]))
    return [p.to_dict() for p in products]

def invalidate_category(category_id: int):
    # 버전을 증가시켜 기존 캐시를 무효화
    r.incr(f"product_version:{category_id}")

import random

def get_with_jitter(key: str, ttl: int = 3600) -> dict:
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    # 분산 잠금으로 하나의 요청만 DB 조회
    lock_key = f"lock:{key}"
    if r.set(lock_key, "1", nx=True, ex=10):
        try:
            data = fetch_from_db(key)
            # TTL에 랜덤 지터 추가
            jitter = random.randint(0, 300)
            r.setex(key, ttl + jitter, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)
    else:
        # 다른 프로세스가 갱신 중 — 짧은 대기 후 재시도
        import time
        time.sleep(0.1)
        return get_with_jitter(key, ttl)

# 구독자
SUBSCRIBE notifications:user:123

# 발행자
PUBLISH notifications:user:123 "You have a new message!"

# 패턴 구독
PSUBSCRIBE notifications:*

# Python 구독자
import redis

r = redis.Redis()
pubsub = r.pubsub()
pubsub.subscribe("notifications:user:123")

for message in pubsub.listen():
    if message["type"] == "message":
        print(f"Received: {message['data']}")

import redis
import time

r = redis.Redis(decode_responses=True)

# 이벤트 발행
def publish_event(stream: str, event_type: str, data: dict):
    r.xadd(stream, {"type": event_type, **data}, maxlen=10000)

# Consumer Group 기반 처리
def consume_events(stream: str, group: str, consumer: str):
    # 그룹 생성 (이미 존재하면 무시)
    try:
        r.xgroup_create(stream, group, id="0", mkstream=True)
    except redis.ResponseError:
        pass

    while True:
        messages = r.xreadgroup(
            group, consumer,
            {stream: ">"},
            count=10,
            block=5000,
        )

        for stream_name, entries in messages:
            for msg_id, fields in entries:
                try:
                    process_event(fields)
                    r.xack(stream_name, group, msg_id)
                except Exception as e:
                    print(f"Error processing {msg_id}: {e}")

def process_event(fields: dict):
    event_type = fields.get("type")
    if event_type == "order_created":
        handle_order(fields)
    elif event_type == "payment_completed":
        handle_payment(fields)

# JSON 문서 저장
JSON.SET user:1 $ '{"name":"Alice","age":30,"address":{"city":"Seoul","zip":"06000"},"tags":["python","redis"]}'

# 경로 기반 조회
JSON.GET user:1 $.name                   # "Alice"
JSON.GET user:1 $.address.city           # "Seoul"

# 부분 업데이트
JSON.SET user:1 $.age 31
JSON.ARRAPPEND user:1 $.tags '"fastapi"'

# 숫자 증감
JSON.NUMINCRBY user:1 $.age 1

# 인덱스 생성
FT.CREATE idx:products
  ON JSON
  PREFIX 1 product:
  SCHEMA
    $.name AS name TEXT WEIGHT 5.0
    $.description AS description TEXT
    $.price AS price NUMERIC SORTABLE
    $.category AS category TAG

# 문서 추가
JSON.SET product:1 $ '{"name":"Redis in Action","description":"Complete guide to Redis","price":45000,"category":"book"}'
JSON.SET product:2 $ '{"name":"Python Cookbook","description":"Python recipes and patterns","price":38000,"category":"book"}'

# 검색
FT.SEARCH idx:products "Redis guide"
FT.SEARCH idx:products "@category:{book} @price:[30000 50000]"
FT.SEARCH idx:products "@name:(Python)" SORTBY price ASC

# 시계열 데이터 생성
TS.CREATE sensor:temperature:1 RETENTION 86400000 LABELS sensor_id 1 type temperature

# 데이터 추가
TS.ADD sensor:temperature:1 * 23.5
TS.ADD sensor:temperature:1 * 24.1
TS.ADD sensor:temperature:1 * 22.8

# 범위 조회
TS.RANGE sensor:temperature:1 - + COUNT 10

# 집계 (5분 평균)
TS.RANGE sensor:temperature:1 - + AGGREGATION avg 300000

# 다운샘플링 규칙
TS.CREATERULE sensor:temperature:1 sensor:temperature:1:avg AGGREGATION avg 300000

# 원자적 읽기-수정-쓰기
EVAL "
  local current = redis.call('GET', KEYS[1])
  if current then
    local new_val = tonumber(current) + tonumber(ARGV[1])
    redis.call('SET', KEYS[1], new_val)
    return new_val
  end
  return nil
" 1 counter 5

# Python + Lua Rate Limiter
RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

-- 윈도우 밖의 오래된 요청 제거
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)

-- 현재 요청 수 확인
local count = redis.call('ZCARD', key)

if count < limit then
    -- 허용: 새 요청 추가
    redis.call('ZADD', key, now, now .. ':' .. math.random())
    redis.call('EXPIRE', key, window)
    return 1
else
    -- 거부
    return 0
end
"""

import redis
import time

r = redis.Redis()
rate_limit_sha = r.script_load(RATE_LIMIT_SCRIPT)

def is_allowed(user_id: str, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}"
    now = int(time.time() * 1000)
    result = r.evalsha(rate_limit_sha, 1, key, limit, window * 1000, now)
    return bool(result)

import redis
import time
import uuid

class DistributedLock:
    def __init__(self, redis_client: redis.Redis, resource: str, ttl: int = 10):
        self.redis = redis_client
        self.resource = f"lock:{resource}"
        self.ttl = ttl
        self.token = str(uuid.uuid4())

    def acquire(self) -> bool:
        return bool(self.redis.set(
            self.resource, self.token,
            nx=True, ex=self.ttl,
        ))

    def release(self) -> bool:
        # Lua 스크립트로 원자적 확인 + 삭제
        script = """
        if redis.call('GET', KEYS[1]) == ARGV[1] then
            return redis.call('DEL', KEYS[1])
        end
        return 0
        """
        return bool(self.redis.eval(script, 1, self.resource, self.token))

    def __enter__(self):
        if not self.acquire():
            raise Exception("Could not acquire lock")
        return self

    def __exit__(self, *args):
        self.release()

# 사용
r = redis.Redis()
with DistributedLock(r, "order:process:123"):
    # 임계 영역 — 하나의 프로세스만 실행
    process_order(123)

노드 구성 예시:
  Node A: 슬롯 0-5460
  Node B: 슬롯 5461-10922
  Node C: 슬롯 10923-16383

각 노드에 1개의 복제본(Replica)을 추가하면:
  Node A -> Replica A'
  Node B -> Replica B'
  Node C -> Replica C'

# 클러스터 생성 (6노드: 3 마스터 + 3 레플리카)
redis-cli --cluster create \
  127.0.0.1:7001 127.0.0.1:7002 127.0.0.1:7003 \
  127.0.0.1:7004 127.0.0.1:7005 127.0.0.1:7006 \
  --cluster-replicas 1

# 클러스터 상태 확인
redis-cli -c -p 7001 cluster info
redis-cli -c -p 7001 cluster nodes

# 수동 페일오버
redis-cli -c -p 7004 cluster failover

# 리샤딩 (슬롯 이동)
redis-cli --cluster reshard 127.0.0.1:7001

# 노드 추가
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7001

# 노드 제거
redis-cli --cluster del-node 127.0.0.1:7001 <node-id>

# 해시 태그 사용 — 중괄호 안의 부분으로 슬롯 결정
SET {user:1}:profile "data"
SET {user:1}:settings "data"
SET {user:1}:sessions "data"

# 이 세 키는 같은 슬롯에 배치됨 -> MULTI 키 연산 가능

import redis
import json
import uuid

r = redis.Redis(decode_responses=True)

def create_session(user_id: int, data: dict) -> str:
    session_id = str(uuid.uuid4())
    session_data = {
        "user_id": str(user_id),
        "created_at": str(time.time()),
        **data,
    }
    r.hset(f"session:{session_id}", mapping=session_data)
    r.expire(f"session:{session_id}", 86400)  # 24시간
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    if not data:
        return None
    # 접근 시 TTL 갱신
    r.expire(f"session:{session_id}", 86400)
    return data

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

def check_rate_limit(user_id: str, limit: int = 100, window: int = 60) -> bool:
    key = f"rate:{user_id}:{int(time.time()) // window}"

    pipe = r.pipeline()
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()

    return count <= limit

class Leaderboard:
    def __init__(self, name: str):
        self.key = f"leaderboard:{name}"

    def add_score(self, player_id: str, score: float):
        r.zadd(self.key, {player_id: score})

    def increment_score(self, player_id: str, delta: float):
        r.zincrby(self.key, delta, player_id)

    def get_rank(self, player_id: str) -> int | None:
        rank = r.zrevrank(self.key, player_id)
        return rank + 1 if rank is not None else None

    def get_top(self, count: int = 10) -> list[tuple[str, float]]:
        return r.zrevrange(self.key, 0, count - 1, withscores=True)

    def get_around(self, player_id: str, count: int = 5) -> list:
        rank = r.zrevrank(self.key, player_id)
        if rank is None:
            return []
        start = max(0, rank - count)
        end = rank + count
        return r.zrevrange(self.key, start, end, withscores=True)

# 사용
lb = Leaderboard("weekly")
lb.add_score("player:alice", 1500)
lb.increment_score("player:alice", 200)
print(lb.get_top(10))
print(lb.get_rank("player:alice"))

import json
import time

def enqueue_job(queue: str, job_data: dict):
    job = {
        "id": str(uuid.uuid4()),
        "data": job_data,
        "created_at": time.time(),
    }
    r.lpush(f"queue:{queue}", json.dumps(job))

def dequeue_job(queue: str, timeout: int = 30) -> dict | None:
    result = r.brpop(f"queue:{queue}", timeout=timeout)
    if result:
        _, job_json = result
        return json.loads(job_json)
    return None

def worker(queue: str):
    while True:
        job = dequeue_job(queue)
        if job:
            try:
                process_job(job["data"])
            except Exception as e:
                # 실패 시 재시도 큐에 추가
                r.lpush(f"queue:{queue}:failed", json.dumps(job))

@Configuration
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        return template;
    }
}

@Service
public class UserCacheService {
    private final RedisTemplate<String, Object> redisTemplate;
    private final Duration TTL = Duration.ofHours(1);

    public UserCacheService(RedisTemplate<String, Object> redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    public void cacheUser(String userId, UserDto user) {
        String key = "user:" + userId;
        redisTemplate.opsForValue().set(key, user, TTL);
    }

    public UserDto getCachedUser(String userId) {
        String key = "user:" + userId;
        return (UserDto) redisTemplate.opsForValue().get(key);
    }

    // Sorted Set 리더보드
    public void addScore(String leaderboard, String playerId, double score) {
        redisTemplate.opsForZSet().add("lb:" + leaderboard, playerId, score);
    }

    public Set<ZSetOperations.TypedTuple<Object>> getTopPlayers(String leaderboard, int count) {
        return redisTemplate.opsForZSet().reverseRangeWithScores("lb:" + leaderboard, 0, count - 1);
    }
}

import Redis from 'ioredis';

const redis = new Redis({
  host: 'localhost',
  port: 6379,
  retryStrategy: (times) => Math.min(times * 50, 2000),
  maxRetriesPerRequest: 3,
});

// 기본 캐싱
async function getUser(userId) {
  const cacheKey = `user:${userId}`;

  const cached = await redis.get(cacheKey);
  if (cached) return JSON.parse(cached);

  const user = await db.findUser(userId);
  if (user) {
    await redis.setex(cacheKey, 3600, JSON.stringify(user));
  }
  return user;
}

// Pipeline (배치 처리)
async function getMultipleUsers(userIds) {
  const pipeline = redis.pipeline();
  userIds.forEach(id => pipeline.get(`user:${id}`));
  const results = await pipeline.exec();
  return results.map(([err, val]) => val ? JSON.parse(val) : null);
}

// Pub/Sub
const subscriber = new Redis();
subscriber.subscribe('notifications', (err, count) => {
  console.log(`Subscribed to ${count} channels`);
});

subscriber.on('message', (channel, message) => {
  console.log(`Received on ${channel}: ${message}`);
});

// 발행
await redis.publish('notifications', JSON.stringify({ type: 'alert', message: 'Server update' }));

import redis.asyncio as aioredis
import json

# 비동기 Redis 클라이언트
pool = aioredis.ConnectionPool.from_url(
    "redis://localhost:6379",
    max_connections=20,
    decode_responses=True,
)
r = aioredis.Redis(connection_pool=pool)

# 파이프라인
async def batch_operations():
    async with r.pipeline(transaction=True) as pipe:
        pipe.set("key1", "value1")
        pipe.set("key2", "value2")
        pipe.get("key1")
        results = await pipe.execute()
        return results

# Pub/Sub (비동기)
async def subscribe_handler():
    pubsub = r.pubsub()
    await pubsub.subscribe("events")

    async for message in pubsub.listen():
        if message["type"] == "message":
            data = json.loads(message["data"])
            await process_event(data)

# 캐시 데코레이터
import functools

def redis_cache(ttl: int = 3600):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            cache_key = f"cache:{func.__name__}:{args}:{kwargs}"
            cached = await r.get(cache_key)
            if cached:
                return json.loads(cached)
            result = await func(*args, **kwargs)
            await r.setex(cache_key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@redis_cache(ttl=600)
async def get_product_list(category_id: int):
    return await db.get_products(category_id)