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

# TTL設定
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            # 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 139.7671 35.6812 "tokyo-store"
GEOADD stores 135.5023 34.6937 "osaka-store"
GEOADD stores 136.9066 35.1815 "nagoya-store"

# 距離計算
GEODIST stores "tokyo-store" "osaka-store" km

# 半径検索（Redis 6.2+）
GEOSEARCH stores FROMLONLAT 139.7671 35.6812 BYRADIUS 500 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.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)

    # 分散ロックで1つのリクエストだけが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

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}")

# JSONドキュメント保存
JSON.SET user:1 $ '{"name":"Alice","age":30,"address":{"city":"Tokyo","zip":"100-0001"},"tags":["python","redis"]}'

# パスベースのクエリ
JSON.GET user:1 $.name                   # "Alice"
JSON.GET user:1 $.address.city           # "Tokyo"

# 部分更新
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

# アトミックな読取-変更-書込
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])

-- ウィンドウ外の古いリクエストを削除
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 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"):
    # クリティカルセクション — 1つのプロセスだけが実行
    process_order(123)

ノード構成例：
  Node A: スロット 0-5460
  Node B: スロット 5461-10922
  Node C: スロット 10923-16383

各ノードに1つのレプリカを追加すると：
  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

# ハッシュタグ使用 — 中括弧内の部分でスロット決定
SET {user:1}:profile "data"
SET {user:1}:settings "data"
SET {user:1}:sessions "data"

# この3つのキーは同じスロットに配置 -> マルチキー操作が可能

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)

# 使用例
lb = Leaderboard("weekly")
lb.add_score("player:alice", 1500)
lb.increment_score("player:alice", 200)
print(lb.get_top(10))

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 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);
    }
}

import Redis from 'ioredis';

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

// 基本キャッシング
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;
}

// パイプライン（バッチ処理）
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);
}

import redis.asyncio as aioredis
import json
import functools

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

# キャッシュデコレータ
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)