Redis 基本觀念完整指南

目錄

• 什麼是 Redis？
• 基本原理
• 核心資料結構
• 持久化機制
• 實戰範例
• 對比分析
• 最佳實踐
• 常見問題
• 總結

什麼是 Redis？

Redis (Remote Dictionary Server) 是一個開源的記憶體資料結構儲存系統，可以用作資料庫、快取和訊息代理。

為什麼需要 Redis？

沒有 Redis 的問題：

• 傳統資料庫讀取速度慢（磁碟 I/O）
• 無法快速處理高併發讀取請求
• 缺乏靈活的資料結構支援
• 分散式快取管理困難

有了 Redis 的好處：

• ✅ 極快的讀寫速度（記憶體操作，微秒級）
• ✅ 豐富的資料結構（String、List、Set、Hash、Sorted Set 等）
• ✅ 支援資料持久化（RDB、AOF）
• ✅ 內建複製、哨兵、叢集功能
• ✅ 原子性操作保證

核心特點

Redis = 記憶體儲存 + 多種資料結構 + 持久化 + 高可用

關鍵特性：

• 速度快：所有資料存在記憶體中
• 資料結構豐富：不只是 key-value
• 原子性：所有操作都是原子性的
• 持久化：可以將記憶體資料保存到磁碟
• 高可用：支援主從複製和故障轉移

基本原理

原理 1：記憶體優先架構

說明：Redis 將所有資料存在記憶體中，提供極快的存取速度。

視覺化：

客戶端請求
    ↓
Redis Server (記憶體)
    ├─ 資料讀寫（微秒級）
    └─ 定期/即時持久化到磁碟
         ↓
    RDB/AOF 檔案

效能對比：

記憶體操作：~0.1 毫秒
SSD 磁碟：  ~1-10 毫秒
HDD 磁碟：  ~10-100 毫秒

原理 2：單執行緒模型

說明：Redis 使用單執行緒處理請求，避免執行緒切換開銷和鎖競爭。

運作方式：

事件循環（Event Loop）
    ↓
處理網路 I/O（非阻塞）
    ↓
執行命令（記憶體操作，極快）
    ↓
返回結果

為什麼單執行緒還這麼快：

• 所有操作在記憶體中完成
• 避免執行緒上下文切換
• 沒有鎖競爭問題
• 使用多路複用 I/O（epoll）

原理 3：鍵值對映射

說明：Redis 使用雜湊表存儲 key-value 對，提供 O(1) 的查找效率。

結構：

雜湊表
├─ key1 → value (String)
├─ key2 → value (List)
├─ key3 → value (Hash)
└─ key4 → value (Set)

核心資料結構

資料結構 1：String（字串）

定義：最基本的資料類型，可以存儲字串、數字或二進制資料。

作用：

• 存儲簡單的 key-value
• 計數器功能
• 分散式鎖
• Session 存儲

常用命令：

# 設定值
SET key value

# 取得值
GET key

# 遞增/遞減
INCR counter
DECR counter

# 設定過期時間（秒）
SETEX key 60 value

# 只有 key 不存在時才設定
SETNX lock 1

範例：

# 存儲使用者 session
SET session:user:123 "user_data" EX 3600

# 頁面瀏覽計數
INCR page:views:homepage

# 分散式鎖
SETNX lock:order:456 1 EX 10

資料結構 2：List（列表）

定義：有序的字串列表，支援從兩端插入和彈出。

作用：

• 訊息佇列
• 最新消息列表
• 時間軸功能

常用命令：

# 從左側插入
LPUSH list value1 value2

# 從右側插入
RPUSH list value3

# 從左側彈出
LPOP list

# 從右側彈出
RPOP list

# 獲取範圍內元素
LRANGE list 0 -1

# 獲取列表長度
LLEN list

範例：

# 最新文章列表（最多保留 100 篇）
LPUSH articles:latest "article:123"
LTRIM articles:latest 0 99

# 簡單訊息佇列
LPUSH queue:tasks "task1"
RPOP queue:tasks

資料結構 3：Hash（雜湊表）

定義：欄位-值對的集合，適合存儲物件。

作用：

• 存儲物件（使用者資訊、商品資訊）
• 減少記憶體使用
• 部分欄位更新

常用命令：

# 設定單個欄位
HSET user:1 name "Alice"

# 設定多個欄位
HMSET user:1 name "Alice" age 30 city "Taipei"

# 獲取單個欄位
HGET user:1 name

# 獲取所有欄位
HGETALL user:1

# 遞增數字欄位
HINCRBY user:1 login_count 1

# 刪除欄位
HDEL user:1 city

範例：

# 存儲使用者資訊
HMSET user:123 name "Bob" email "bob@example.com" points 1000

# 更新積分
HINCRBY user:123 points 50

# 獲取使用者資訊
HGETALL user:123

資料結構 4：Set（集合）

定義：無序且唯一的字串集合。

作用：

• 標籤系統
• 共同好友
• 唯一性統計（UV）
• 去重

常用命令：

# 添加元素
SADD set value1 value2

# 獲取所有成員
SMEMBERS set

# 判斷是否存在
SISMEMBER set value1

# 移除元素
SREM set value1

# 集合運算
SINTER set1 set2      # 交集
SUNION set1 set2      # 聯集
SDIFF set1 set2       # 差集

# 獲取集合大小
SCARD set

範例：

# 文章標籤
SADD article:123:tags "Redis" "Database" "NoSQL"

# 使用者關注列表
SADD user:alice:following user:bob user:charlie

# 共同關注（交集）
SINTER user:alice:following user:bob:following

# 唯一訪客統計
SADD page:homepage:visitors user:123 user:456
SCARD page:homepage:visitors

資料結構 5：Sorted Set（有序集合）

定義：帶分數的有序集合，成員唯一，按分數排序。

作用：

• 排行榜
• 延遲佇列
• 時間序列資料
• 優先級佇列

常用命令：

# 添加成員（帶分數）
ZADD leaderboard 100 "player1" 200 "player2"

# 獲取排名範圍（按分數從小到大）
ZRANGE leaderboard 0 9

# 獲取排名範圍（按分數從大到小）
ZREVRANGE leaderboard 0 9 WITHSCORES

# 獲取成員分數
ZSCORE leaderboard "player1"

# 增加分數
ZINCRBY leaderboard 50 "player1"

# 獲取排名
ZRANK leaderboard "player1"

# 按分數範圍查詢
ZRANGEBYSCORE leaderboard 100 200

範例：

# 遊戲排行榜
ZADD game:leaderboard 9500 "Alice" 8200 "Bob" 9800 "Charlie"
ZREVRANGE game:leaderboard 0 9 WITHSCORES  # Top 10

# 延遲任務佇列（使用時間戳作為分數）
ZADD delayed:tasks 1700000000 "task1" 1700000060 "task2"
ZRANGEBYSCORE delayed:tasks 0 [current_timestamp]  # 獲取到期任務

持久化機制

機制 1：RDB（快照）

說明：在指定時間間隔內將記憶體中的資料快照寫入磁碟。

優點：

• 檔案小，適合備份
• 恢復速度快
• 對效能影響小

缺點：

• 可能丟失最後一次快照後的資料
• Fork 子程序時可能造成短暫停頓

配置：

# redis.conf
save 900 1      # 900 秒內至少 1 次寫入
save 300 10     # 300 秒內至少 10 次寫入
save 60 10000   # 60 秒內至少 10000 次寫入

手動觸發：

SAVE      # 阻塞式保存
BGSAVE    # 背景保存（推薦）

機制 2：AOF（Append Only File）

說明：記錄每個寫操作，伺服器重啟時重新執行命令來恢復資料。

優點：

• 資料更安全，最多丟失 1 秒資料
• 檔案可讀，易於修復
• 自動重寫壓縮檔案

缺點：

• 檔案通常比 RDB 大
• 恢復速度較慢
• 對效能影響較大

配置：

# redis.conf
appendonly yes
appendfsync everysec   # 每秒同步一次（推薦）
# appendfsync always   # 每次寫入都同步（最安全但慢）
# appendfsync no       # 由作業系統決定（最快但不安全）

持久化策略選擇

實戰範例

範例 1：實作分散式 Session

情境：多台 Web 伺服器共享使用者 session

實作：

import redis
from flask import Flask, session

app = Flask(__name__)
redis_client = redis.Redis(host='localhost', port=6379, decode_responses=True)

@app.route('/login', methods=['POST'])
def login():
    user_id = authenticate_user()  # 驗證使用者
    session_id = generate_session_id()

    # 將 session 存入 Redis，過期時間 1 小時
    redis_client.setex(
        f'session:{session_id}',
        3600,
        json.dumps({'user_id': user_id, 'login_time': time.time()})
    )

    return {'session_id': session_id}

@app.route('/profile')
def profile():
    session_id = request.headers.get('Session-ID')
    session_data = redis_client.get(f'session:{session_id}')

    if not session_data:
        return {'error': 'Session expired'}, 401

    user_data = json.loads(session_data)
    return {'user_id': user_data['user_id']}

範例 2：實作排行榜系統

情境：即時更新的遊戲排行榜

實作：

import redis

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

def update_score(player_id, score):
    """更新玩家分數"""
    redis_client.zadd('game:leaderboard', {player_id: score})

def get_top_players(n=10):
    """獲取前 N 名玩家"""
    return redis_client.zrevrange('game:leaderboard', 0, n-1, withscores=True)

def get_player_rank(player_id):
    """獲取玩家排名（從 1 開始）"""
    rank = redis_client.zrevrank('game:leaderboard', player_id)
    return rank + 1 if rank is not None else None

def get_around_players(player_id, range=5):
    """獲取玩家周圍的排名"""
    rank = redis_client.zrevrank('game:leaderboard', player_id)
    if rank is None:
        return []

    start = max(0, rank - range)
    end = rank + range
    return redis_client.zrevrange('game:leaderboard', start, end, withscores=True)

# 使用範例
update_score('player123', 9500)
update_score('player456', 8200)

print(get_top_players(10))  # 前 10 名
print(get_player_rank('player123'))  # 玩家排名

範例 3：實作快取系統

情境：減輕資料庫查詢壓力

實作：

import redis
import json
from functools import wraps

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

def cache_result(expire=3600):
    """快取裝飾器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成快取鍵
            cache_key = f"cache:{func.__name__}:{args}:{kwargs}"

            # 檢查快取
            cached = redis_client.get(cache_key)
            if cached:
                return json.loads(cached)

            # 執行函數
            result = func(*args, **kwargs)

            # 存入快取
            redis_client.setex(cache_key, expire, json.dumps(result))

            return result
        return wrapper
    return decorator

@cache_result(expire=600)
def get_user_profile(user_id):
    """獲取使用者資料（從資料庫）"""
    # 模擬資料庫查詢
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")
    return user

# 快取預熱
def warm_up_cache(user_ids):
    """預先載入熱門資料到快取"""
    for user_id in user_ids:
        user = get_user_from_db(user_id)
        redis_client.setex(
            f'user:{user_id}',
            3600,
            json.dumps(user)
        )

# 快取失效
def invalidate_cache(user_id):
    """刪除快取"""
    redis_client.delete(f'user:{user_id}')

範例 4：實作計數器與限流

情境：API 請求限流

實作：

import redis
import time

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

def rate_limit(user_id, max_requests=100, window=60):
    """
    限流檢查：每分鐘最多 100 次請求

    Args:
        user_id: 使用者 ID
        max_requests: 最大請求數
        window: 時間窗口（秒）

    Returns:
        bool: 是否允許請求
    """
    key = f'rate_limit:{user_id}'
    current = redis_client.get(key)

    if current is None:
        # 第一次請求
        redis_client.setex(key, window, 1)
        return True

    if int(current) >= max_requests:
        return False

    # 遞增計數
    redis_client.incr(key)
    return True

# 使用範例
@app.route('/api/data')
def api_endpoint():
    user_id = get_current_user_id()

    if not rate_limit(user_id, max_requests=100, window=60):
        return {'error': 'Rate limit exceeded'}, 429

    # 處理請求
    return {'data': 'success'}

# 進階：滑動窗口限流
def sliding_window_rate_limit(user_id, max_requests=100, window=60):
    """使用 Sorted Set 實現滑動窗口限流"""
    key = f'rate_limit_sliding:{user_id}'
    now = time.time()

    # 移除過期的請求記錄
    redis_client.zremrangebyscore(key, 0, now - window)

    # 獲取當前窗口內的請求數
    current_requests = redis_client.zcard(key)

    if current_requests >= max_requests:
        return False

    # 記錄新請求
    redis_client.zadd(key, {str(now): now})
    redis_client.expire(key, window)

    return True

對比分析

Redis vs Memcached

何時選擇 Redis：

• 需要豐富的資料結構
• 需要資料持久化
• 需要主從複製或叢集
• 需要事務支援

何時選擇 Memcached：

• 純快取用途
• 追求極致效能
• 資料結構簡單

Redis vs 傳統關聯式資料庫

最佳實踐

1. 合理設定過期時間

# ❌ 不好的做法：沒有設定過期時間
SET user:123 "data"

# ✅ 好的做法：設定適當的過期時間
SETEX user:123 3600 "data"

# ✅ 好的做法：不同資料設定不同過期時間
SETEX session:abc 1800 "session_data"     # 30 分鐘
SETEX cache:product:456 86400 "product"   # 1 天
SETEX temp:otp:789 300 "123456"           # 5 分鐘

建議：

• Session 資料：15-30 分鐘
• 快取資料：根據更新頻率設定（1 小時至 1 天）
• 臨時資料：越短越好
• 避免永久資料，除非必要

2. 使用適當的資料結構

# ❌ 不好的做法：用 String 存儲物件
SET user:123 '{"name":"Alice","age":30,"city":"Taipei"}'
# 更新單個欄位需要：GET → 解析 → 修改 → SET

# ✅ 好的做法：用 Hash 存儲物件
HMSET user:123 name "Alice" age 30 city "Taipei"
HINCRBY user:123 age 1  # 直接更新單個欄位

# ❌ 不好的做法：用 String 實現排行榜
SET rank:player1 100
SET rank:player2 200
# 獲取前 10 名需要複雜邏輯

# ✅ 好的做法：用 Sorted Set
ZADD leaderboard 100 "player1" 200 "player2"
ZREVRANGE leaderboard 0 9  # 直接獲取前 10 名

3. 避免大 Key

# ❌ 不好的做法：單個 List 存儲大量資料
LPUSH logs "log1" "log2" ... "log100000"  # 10 萬條日誌

# ✅ 好的做法：分片存儲
LPUSH logs:2024-11-18 "log1" "log2" ...
LPUSH logs:2024-11-19 "log3" "log4" ...

# ❌ 不好的做法：單個 Hash 存儲所有使用者
HMSET users user1 "data1" user2 "data2" ...

# ✅ 好的做法：獨立 Key
HMSET user:1 name "Alice" age 30
HMSET user:2 name "Bob" age 25

大 Key 的問題：

• 操作阻塞時間長
• 網路傳輸慢
• 記憶體分配壓力大
• 刪除時可能造成阻塞

4. 批次操作提升效能

# ❌ 不好的做法：多次單獨操作
SET key1 value1
SET key2 value2
SET key3 value3

# ✅ 好的做法：使用 Pipeline
MULTI
SET key1 value1
SET key2 value2
SET key3 value3
EXEC

# ✅ 更好的做法：使用 MSET
MSET key1 value1 key2 value2 key3 value3

# Python 範例
pipe = redis_client.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
pipe.execute()

5. 監控記憶體使用

# 查看記憶體使用情況
INFO memory

# 查看所有 Key 的記憶體佔用
redis-cli --bigkeys

# 設定最大記憶體和淘汰策略
# redis.conf
maxmemory 2gb
maxmemory-policy allkeys-lru  # LRU 淘汰最少使用的 key

淘汰策略：

• noeviction：不淘汰，記憶體滿時返回錯誤
• allkeys-lru：所有 key 中淘汰最少使用的（推薦）
• volatile-lru：只在設定過期時間的 key 中淘汰
• allkeys-random：隨機淘汰
• volatile-ttl：淘汰即將過期的 key

6. 合理使用持久化

# ✅ 推薦配置：RDB + AOF
# redis.conf

# RDB 配置
save 900 1
save 300 10
save 60 10000

# AOF 配置
appendonly yes
appendfsync everysec  # 每秒同步，兼顧效能和安全

建議：

• 開發環境：不持久化（最快）
• 測試環境：僅 RDB
• 生產環境（快取）：RDB
• 生產環境（資料庫）：RDB + AOF

7. 使用連線池

# ❌ 不好的做法：每次建立新連線
def get_data(key):
    client = redis.Redis(host='localhost', port=6379)
    return client.get(key)

# ✅ 好的做法：使用連線池
from redis import ConnectionPool

pool = ConnectionPool(
    host='localhost',
    port=6379,
    max_connections=50,
    decode_responses=True
)

redis_client = redis.Redis(connection_pool=pool)

def get_data(key):
    return redis_client.get(key)

8. 避免阻塞操作

# ❌ 危險的操作：在生產環境使用
KEYS *           # 掃描所有 key，會阻塞
FLUSHALL        # 清空所有資料庫，會阻塞
FLUSHDB         # 清空當前資料庫，會阻塞

# ✅ 安全的替代方案
SCAN 0 MATCH user:* COUNT 100  # 漸進式掃描
DEL key1 key2 key3             # 批次刪除指定 key

常見問題

Q1: Redis 為什麼這麼快？

A:

1. 記憶體操作：所有資料在記憶體中，避免磁碟 I/O
2. 單執行緒模型：避免執行緒切換和鎖競爭
3. 高效資料結構：針對不同場景優化的資料結構
4. 非阻塞 I/O：使用 epoll 等多路複用技術
5. 簡單協議：RESP 協議解析開銷小

Q2: Redis 單執行緒如何處理併發？

A: Redis 使用事件驅動模型和非阻塞 I/O：

事件循環
├─ 接收多個客戶端連線（非阻塞）
├─ 依序處理每個請求（記憶體操作極快）
└─ 返回結果給對應客戶端

因為記憶體操作極快（微秒級），單執行緒已足夠處理大量併發請求。

補充：Redis 6.0 後引入多執行緒，但只用於網路 I/O，命令執行仍是單執行緒。

Q3: 如何保證 Redis 和資料庫的資料一致性？

A: 常見策略

1. Cache Aside Pattern（旁路快取）

def get_user(user_id):
    # 1. 先查 Redis
    user = redis_client.get(f'user:{user_id}')
    if user:
        return json.loads(user)

    # 2. 查資料庫
    user = db.query(f"SELECT * FROM users WHERE id = {user_id}")

    # 3. 寫入 Redis
    redis_client.setex(f'user:{user_id}', 3600, json.dumps(user))

    return user

def update_user(user_id, data):
    # 1. 先更新資料庫
    db.update(f"UPDATE users SET ... WHERE id = {user_id}")

    # 2. 刪除快取（而非更新）
    redis_client.delete(f'user:{user_id}')

2. Read/Write Through Pattern

• 應用程式只與快取互動
• 快取層負責與資料庫同步

3. Write Behind Pattern（非同步寫入）

• 先寫快取，非同步寫資料庫
• 適合寫入密集場景

Q4: Redis 記憶體滿了怎麼辦？

A: 幾種解決方案

1. 設定淘汰策略

maxmemory 2gb
maxmemory-policy allkeys-lru

2. 設定過期時間

SETEX key 3600 value  # 1 小時後自動過期

3. 定期清理

# 清理過期的臨時資料
def cleanup_expired_keys():
    cursor = 0
    while True:
        cursor, keys = redis_client.scan(cursor, match='temp:*', count=100)
        for key in keys:
            redis_client.delete(key)
        if cursor == 0:
            break

4. 擴展記憶體或使用叢集

• 增加伺服器記憶體
• 使用 Redis Cluster 分散資料

Q5: 如何防止快取穿透、雪崩和擊穿？

A:

快取穿透（查詢不存在的資料）

# 解決方案：快取空值
def get_user(user_id):
    user = redis_client.get(f'user:{user_id}')
    if user == 'null':  # 空值標記
        return None
    if user:
        return json.loads(user)

    user = db.query(...)
    if user is None:
        redis_client.setex(f'user:{user_id}', 300, 'null')  # 快取空值 5 分鐘
        return None

    redis_client.setex(f'user:{user_id}', 3600, json.dumps(user))
    return user

快取雪崩（大量 key 同時過期）

# 解決方案：過期時間加上隨機值
import random

expire_time = 3600 + random.randint(0, 300)  # 3600-3900 秒
redis_client.setex(key, expire_time, value)

快取擊穿（熱點 key 過期）

# 解決方案：使用鎖或設定永不過期
import redis.lock

def get_hot_data(key):
    data = redis_client.get(key)
    if data:
        return data

    # 使用分散式鎖
    lock = redis.lock.Lock(redis_client, f'lock:{key}', timeout=10)
    if lock.acquire(blocking=False):
        try:
            data = load_from_db()
            redis_client.setex(key, 3600, data)
            return data
        finally:
            lock.release()
    else:
        # 等待其他執行緒載入資料
        time.sleep(0.1)
        return redis_client.get(key)

總結

核心要點

Redis = 記憶體儲存 + 豐富資料結構 + 高效能 + 持久化

關鍵特性：
  ├─ 速度：微秒級操作
  ├─ 資料結構：String, List, Hash, Set, Sorted Set
  ├─ 持久化：RDB + AOF
  └─ 高可用：複製、哨兵、叢集

使用決策

快速參考

常用命令速查：

# String
SET key value
GET key
INCR counter

# Hash
HMSET user:1 name "Alice" age 30
HGETALL user:1

# List
LPUSH list value
RPOP list

# Set
SADD set value
SMEMBERS set

# Sorted Set
ZADD leaderboard 100 "player1"
ZREVRANGE leaderboard 0 9

效能基準：

• 讀寫速度：10-20 萬 QPS（單機）
• 延遲：微秒級（記憶體操作）
• 記憶體：建議單機 < 20GB
• 連線數：建議 < 10000

記憶口訣

Redis 三大核心：

1. 快 - 記憶體操作，微秒級
2. 活 - 多種資料結構，靈活應用
3. 穩 - 持久化機制，資料可靠

選擇 Redis 的三個問題：

1. 需要快速讀寫？ → 是
2. 需要複雜資料結構？ → 是
3. 資料量適中（< TB）？ → 是

建立日期：2025-11-18 最後更新：2025-11-18