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

# Basic SET/GET
SET user:1:name "Alice"
GET user:1:name

# Atomic increment/decrement
SET page:views 0
INCR page:views          # 1
INCRBY page:views 10     # 11

# TTL setting
SET session:abc123 "user_data" EX 3600   # Expires in 1 hour
TTL session:abc123                        # Check remaining time

# SET options
SET lock:resource "owner1" NX EX 30      # NX: Set only if key doesn't exist
SET user:1:name "Bob" XX                 # XX: Update only if key exists

# Basic operations
LPUSH queue:emails "email1" "email2" "email3"
RPOP queue:emails                        # "email1" (FIFO queue)

# Range query
LRANGE queue:emails 0 -1                 # All elements

# Blocking pop (message queue pattern)
BRPOP queue:emails 30                    # Wait 30 seconds then pop

# Trimming (keep recent N items)
LPUSH notifications:user1 "new_msg"
LTRIM notifications:user1 0 99          # Keep only recent 100

# Basic operations
SADD tags:post:1 "python" "redis" "backend"
SADD tags:post:2 "python" "django" "orm"

# Membership check
SISMEMBER tags:post:1 "python"           # 1 (true)

# Set operations
SINTER tags:post:1 tags:post:2           # "python" (intersection)
SUNION tags:post:1 tags:post:2           # union
SDIFF tags:post:1 tags:post:2            # elements only in post:1

# Random sampling
SRANDMEMBER tags:post:1 2               # Random 2 elements

# Leaderboard implementation
ZADD leaderboard 1500 "player:alice"
ZADD leaderboard 2300 "player:bob"
ZADD leaderboard 1800 "player:charlie"

# Ranking query (descending by score)
ZREVRANGE leaderboard 0 2 WITHSCORES
# 1) "player:bob"     2) "2300"
# 3) "player:charlie" 4) "1800"
# 5) "player:alice"   6) "1500"

# Specific member rank (0-based)
ZREVRANK leaderboard "player:alice"      # 2

# Score increment
ZINCRBY leaderboard 500 "player:alice"   # 2000

# Range search
ZRANGEBYSCORE leaderboard 1500 2000 WITHSCORES

# Store user profile
HSET user:1 name "Alice" email "alice@example.com" age "30" role "admin"

# Get individual field
HGET user:1 name                         # "Alice"

# Get all
HGETALL user:1

# Field increment
HINCRBY user:1 age 1                     # 31

# Existence check
HEXISTS user:1 email                     # 1 (true)

# Multiple fields at once
HMGET user:1 name email role

# Estimate unique visitors
PFADD visitors:2025-03-23 "user1" "user2" "user3"
PFADD visitors:2025-03-23 "user1" "user4"          # user1 is duplicate

PFCOUNT visitors:2025-03-23                         # 4

# Merge multiple days
PFMERGE visitors:week visitors:2025-03-23 visitors:2025-03-24
PFCOUNT visitors:week

# Daily attendance check
SETBIT attendance:2025-03-23 1001 1      # User 1001 present
SETBIT attendance:2025-03-23 1002 1
SETBIT attendance:2025-03-23 1003 0      # Absent

# Check attendance
GETBIT attendance:2025-03-23 1001        # 1

# Count attendees
BITCOUNT attendance:2025-03-23           # 2

# Consecutive attendance (AND operation)
BITOP AND consecutive attendance:2025-03-22 attendance:2025-03-23
BITCOUNT consecutive

# Add locations (longitude, latitude)
GEOADD stores 126.9784 37.5665 "gangnam-store"
GEOADD stores 127.0276 37.4979 "samsung-store"
GEOADD stores 126.9316 37.5563 "hongdae-store"

# Distance calculation
GEODIST stores "gangnam-store" "hongdae-store" km    # ~5.2 km

# Radius search (Redis 6.2+)
GEOSEARCH stores FROMLONLAT 126.9784 37.5665 BYRADIUS 10 km ASC COUNT 5

# Add messages
XADD events * type "order" user_id "123" amount "50000"
XADD events * type "payment" user_id "123" status "completed"

# Read
XRANGE events - + COUNT 10

# Create Consumer Group
XGROUP CREATE events analytics-group $ MKSTREAM

# Read as consumer
XREADGROUP GROUP analytics-group consumer-1 COUNT 5 BLOCK 2000 STREAMS events >

# ACK (processing complete)
XACK events analytics-group "1679000000000-0"

# Check pending messages
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. Check cache
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # 2. Cache miss -> Query DB
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        return None

    # 3. Store in cache
    r.setex(cache_key, 3600, json.dumps(user.to_dict()))
    return user.to_dict()

def update_user(user_id: int, data: dict):
    db.query(User).filter(User.id == user_id).update(data)
    db.commit()

    # Invalidate cache
    r.delete(f"user:{user_id}")

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

    # Update DB and cache simultaneously
    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}"

    # Write to cache first
    r.setex(cache_key, 3600, json.dumps(data))

    # Async DB persistence (Celery, etc.)
    sync_to_db_task.delay(user_id, data)

SET product:123 "data" EX 300           # Expires in 5 minutes

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

    # Delete all related caches
    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):
    # Increment version to invalidate existing caches
    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)

    # Distributed lock — only one request queries DB
    lock_key = f"lock:{key}"
    if r.set(lock_key, "1", nx=True, ex=10):
        try:
            data = fetch_from_db(key)
            # Add random jitter to TTL
            jitter = random.randint(0, 300)
            r.setex(key, ttl + jitter, json.dumps(data))
            return data
        finally:
            r.delete(lock_key)
    else:
        # Another process is refreshing — wait briefly and retry
        import time
        time.sleep(0.1)
        return get_with_jitter(key, ttl)

# Subscriber
SUBSCRIBE notifications:user:123

# Publisher
PUBLISH notifications:user:123 "You have a new message!"

# Pattern subscription
PSUBSCRIBE notifications:*

# Python subscriber
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)

# Publish event
def publish_event(stream: str, event_type: str, data: dict):
    r.xadd(stream, {"type": event_type, **data}, maxlen=10000)

# Consumer Group processing
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)

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

# Path-based query
JSON.GET user:1 $.name                   # "Alice"
JSON.GET user:1 $.address.city           # "Seoul"

# Partial update
JSON.SET user:1 $.age 31
JSON.ARRAPPEND user:1 $.tags '"fastapi"'

# Numeric increment
JSON.NUMINCRBY user:1 $.age 1

# Create index
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

# Add documents
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"}'

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

# Create time series
TS.CREATE sensor:temperature:1 RETENTION 86400000 LABELS sensor_id 1 type temperature

# Add data
TS.ADD sensor:temperature:1 * 23.5
TS.ADD sensor:temperature:1 * 24.1
TS.ADD sensor:temperature:1 * 22.8

# Range query
TS.RANGE sensor:temperature:1 - + COUNT 10

# Aggregation (5-minute average)
TS.RANGE sensor:temperature:1 - + AGGREGATION avg 300000

# Downsampling rule
TS.CREATERULE sensor:temperature:1 sensor:temperature:1:avg AGGREGATION avg 300000

# Atomic read-modify-write
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

RATE_LIMIT_SCRIPT = """
local key = KEYS[1]
local limit = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local now = tonumber(ARGV[3])

-- Remove old requests outside window
redis.call('ZREMRANGEBYSCORE', key, 0, now - window)

-- Check current request count
local count = redis.call('ZCARD', key)

if count < limit then
    -- Allow: add new request
    redis.call('ZADD', key, now, now .. ':' .. math.random())
    redis.call('EXPIRE', key, window)
    return 1
else
    -- Deny
    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:
        # Atomic check + delete with Lua script
        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()

# Usage
r = redis.Redis()
with DistributedLock(r, "order:process:123"):
    # Critical section — only one process executes
    process_order(123)

Node configuration example:
  Node A: slots 0-5460
  Node B: slots 5461-10922
  Node C: slots 10923-16383

With one replica per node:
  Node A -> Replica A'
  Node B -> Replica B'
  Node C -> Replica C'

# Create cluster (6 nodes: 3 masters + 3 replicas)
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

# Check cluster status
redis-cli -c -p 7001 cluster info
redis-cli -c -p 7001 cluster nodes

# Manual failover
redis-cli -c -p 7004 cluster failover

# Resharding (move slots)
redis-cli --cluster reshard 127.0.0.1:7001

# Add node
redis-cli --cluster add-node 127.0.0.1:7007 127.0.0.1:7001

# Remove node
redis-cli --cluster del-node 127.0.0.1:7001 <node-id>

# Hash tags — slot determined by content in curly braces
SET {user:1}:profile "data"
SET {user:1}:settings "data"
SET {user:1}:sessions "data"

# These three keys are in the same slot -> multi-key operations possible

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 hours
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    if not data:
        return None
    # Refresh TTL on access
    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)

# Usage
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:
                # On failure, add to retry queue
                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);
    }

    public void addScore(String leaderboard, String playerId, double score) {
        redisTemplate.opsForZSet().add("lb:" + leaderboard, playerId, score);
    }
}

import Redis from 'ioredis';

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

// Basic caching
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 (batch processing)
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

# Async Redis client
pool = aioredis.ConnectionPool.from_url(
    "redis://localhost:6379",
    max_connections=20,
    decode_responses=True,
)
r = aioredis.Redis(connection_pool=pool)

# Pipeline
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

# Cache decorator
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)