現代化開發利器：Python Redis 非同步、叢集與進階模式深度剖析

在現代應用程式開發中，數據的存取速度與系統效能是決定使用者體驗的關鍵。Redis (REmote DIctionary Server) 作為一款開源、高效能、基於記憶體的鍵值（Key-Value）儲存數據庫，正是為瞭解決這些挑戰而生。它不僅僅是一個簡單的快取系統，更是一個強大的「資料結構伺服器」，原生支援字符串、雜湊、列表、集合、有序集合等多種資料類型，使其在快取、訊息佇列、即時分析、排行榜、工作階段儲存等場景中大放異彩。

Python 作為目前最受歡迎的程式語言之一，擁有龐大而活躍的生態系統。當 Python 遇上 Redis，開發者可以透過 redis-py 這個官方推薦的客戶端函式庫，輕鬆地將 Redis 的強大功能整合到應用程式中。無論是建構高效能的 Web 服務、處理複雜的數據管道，還是實現即時的應用功能，Python for redis 的結合都能提供優雅而強大的解決方案。

本篇文章將作為一份全方位的實戰指南，從環境建置、基礎連線開始，深入探討 Redis 的核心資料結構在 Python 中的詳細操作，並涵蓋連接池、管道（Pipeline）、非同步處理等進階主題。無論您是初次接觸 Redis 的新手，還是希望深化 Python Redis 應用的開發者，都能在此找到詳盡且具實用性的內容。

環境準備與安裝

在開始撰寫程式碼之前，我們需要確保 Redis 伺服器已啟動，並且 Python 環境中已安裝好必要的客戶端套件。

安裝並啟動 Redis 伺服器

對於開發與測試，使用 Docker 來啟動 Redis 伺服器是最快捷便利的方式。只需一行指令，即可擁有一個乾淨獨立的 Redis 環境。

# 啟動一個名為 my-redis 的 Redis 容器，並將容器的 6379 連接埠映射到主機
docker run --name my-redis -p 6379:6379 -d redis

若有密碼驗證需求，可以在啟動時加上 requirepass 參數：

docker run --name my-redis -p 6379:6379 -d redis redis-server --requirepass "your_strong_password"

安裝 Python Redis 客戶端

redis-py 是 Python 連接 Redis 的標準函式庫，它提供了一套完整的 API。我們可以使用 pip 來安裝它。

pip install redis

值得注意的是，從 redis-py 4.2.0 版本開始，知名的非同步函式庫 aioredis 已經被整合進來，我們可以透過 redis.asyncio 模組來使用非同步功能，無需額外安裝。

對於需要與 Redis Cluster 互動的場景，則需要安裝專門的叢集客戶端：

pip install redis-py-cluster

連線到 Redis：基礎與進階模式

成功連線是所有操作的第一步。redis-py 提供了多種靈活的連線方式，以適應不同的應用場景。

基本連線

最直接的方式是實例化 redis.Redis client 類別。

import redis

# 建立 Redis 連線
# decode_responses=True 讓取回的值自動從 bytes 解碼為 utf-8 字串，強烈建議設定
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,  # 預設使用 0 號資料庫
    password=None, # 如果 Redis 有設定密碼，請填寫
    decode_responses=True
)

# 測試連線是否成功
try:
    response = r.ping()
    print(f"成功連線到 Redis！回應: {response}")
except redis.exceptions.ConnectionError as e:
    print(f"無法連線到 Redis: {e}")

# 進行簡單操作
r.set('mykey', 'Hello, Redis!')
value = r.get('mykey')
print(f"從 Redis 取回的值: {value}")

# 關閉連線 (在短腳本中非必要，但良好習慣)
r.close()

使用連線池 (Connection Pool)

在 Web 應用或長駐服務中，每次操作都建立和釋放 TCP 連線會帶來巨大的效能開銷。連接池（Connection Pool）正是為瞭解決這個問題而設計的。它會預先建立並管理一組連線，應用程式在需要時直接從池中獲取，用畢後歸還，實現了連線的複用。

redis-py 預設在內部使用連線池。若要多個 Redis 實例共享一個 pool redis 連線池，可以手動建立並傳入。

import redis

# 建立連接池
pool = redis.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True
)

# 建立兩個 Redis 實例，它們共享同一個連線池
r1 = redis.Redis(connection_pool=pool)
r2 = redis.Redis(connection_pool=pool)

r1.set('shared_key', 'This is from r1')
value_from_r2 = r2.get('shared_key')

print(f"r2 讀取由 r1 寫入的值: {value_from_r2}")

# 應用程式結束時，可以斷開連線池
pool.disconnect()

對於多執行緒環境，建議使用執行緒安全的 BlockingConnectionPool。

非同步連線 (Asyncio)

現代 Python 應用大量使用 asyncio 進行非同步 I/O 操作以提升併發效能。redis-py 提供了完整的非同步支援。

import asyncio
import redis.asyncio as redis

async def main():
    # 建立非同步 Redis 連線
    r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

    # 所有操作都需要使用 await
    await r.set('async_key', 'Hello, Async Redis!')
    value = await r.get('async_key')
    print(f"非同步操作取回的值: {value}")

    # 關閉連線
    await r.close()

if __name__ == "__main__":
    asyncio.run(main())

特殊連線模式

哨兵模式 (Sentinel): 用於高可用架構，當主節點 (master) 故障時，Sentinel 會自動將從節點 (slave) 提升為新的主節點。

python

from redis.sentinel import Sentinel sentinel_nodes = [(‘localhost’, 26379)]
sentinel = Sentinel(sentinel_nodes, socket_timeout=0.1, decode_responses=True)

獲取主節點的寫入連線

master = sentinel.master_for(‘mymaster’, socket_timeout=0.1)

獲取一個從節點的讀取連線

slave = sentinel.slave_for(‘mymaster’, socket_timeout=0.1) master.set(‘high_availability’, ‘ok’)
print(slave.get(‘high_availability’))

叢集模式 (Cluster): 用於數據分片，將數據分散到多個 Redis 節點，以支援大規模數據儲存和高吞吐量。

python

from rediscluster import StrictRedisCluster startup_nodes = [{“host”: “localhost”, “port”: “7000”}]

decode_responses=True 同樣適用

rc = StrictRedisCluster(startup_nodes=startup_nodes, decode_responses=True) rc.set(“cluster_key”,

“Hello, Cluster!”)

print(rc.get(“cluster_key”))

核心資料結構與 Python 操作詳解

掌握 Redis 的核心在於理解其豐富的資料結構。以下將詳細介紹五種主要資料類型在 redis-py 中的使用方法。

字串 (String)

String 是 Redis 最基本的資料類型，可以儲存任何形式的資料，如字符串、整數、浮點數，甚至是序列化後的物件。其最大容量為 512MB。

常用操作：

Python 方法	Redis 指令	說明
r.set(name, value, ex, px, nx, xx)	SET	執行設置操作。ex: 秒過期, px: 毫秒過期, nx: 不存在才設定 (nx false 為預設行為), xx: 存在才設定 (xx false 為預設行為)。
r.get(name)	GET	獲取 name對應值。
r.mset({key1: val1, …})	MSET	批量設置值。
r.mget([key1, key2, …])	MGET	批次獲取多個鍵值。
r.incr(name, amount=1)	INCRBY	將值原子性地增加指定整數，常用於計數器。
r.incrbyfloat(name, amount=1.0)	INCRBYFLOAT	原子性地增加指定浮點數。
r.decr(name, amount=1)	DECRBY	原子性地減少指定整數。
r.append(name, value)	APPEND	在現有值的末尾追加內容。
r.strlen(name)	STRLEN	獲取值的位元組長度。

程式碼範例：

# 基本設定與獲取
r.set('user:1:name', 'Alice')
print(r.get('user:1:name'))  # 輸出: Alice

# 設置值並帶有 60 秒過期時間
r.set('session:token', 'xyz123', ex=60)

# 只有當 'counter' 不存在時才設定
r.set('counter', 0, nx=True)

# 頁面點擊計數器應用
page_id = 'product:123'
r.incr(f"page:views:{page_id}")
r.incr(f"page:views:{page_id}")
print(f"頁面 {page_id} 的點擊數: {r.get(f'page:views:{page_id}')}") # 輸出: 2

# 批次操作
r.mset({'user:2:name': 'Bob', 'user:3:name': 'Charlie'})
names = r.mget(['user:1:name', 'user:2:name', 'user:3:name'])
print(names) # 輸出: ['Alice', 'Bob', 'Charlie']

雜湊 (Hash)

Hash 非常適合用來儲存物件。你可以將一個物件（如一個使用者）的多個屬性（如姓名、年齡、信箱）儲存在一個 Redis key 中，而不是為每個屬性建立一個 key。

常用操作：

Python 方法	Redis 指令	說明
r.hset(name, key, value)	HSET	設定 hash 中一個欄位的值。
r.hget(name, key)	HGET	獲取 hash 中一個欄位的值。
r.hmset(name, mapping)	HMSET	批次設定 hash 中的多個欄位。
r.hmget(name, keys)	HMGET	批次獲取 hash 中的多個欄位。
r.hgetall(name)	HGETALL	獲取 hash 中所有的欄位和值。
r.hkeys(name)	HKEYS	獲取 hash 中所有的欄位名。
r.hvals(name)	HVALS	獲取 hash 中所有的值。
r.hlen(name)	HLEN	獲取 hash 的欄位數量，即 key count。
r.hincrby(name, key, amount=1)	HINCRBY	對 hash 中指定欄位的值進行原子性增減。
r.hdel(name, *keys)	HDEL	刪除 hash 中鍵值的一個或多個欄位。

程式碼範例：

user_id = 'user:101'
user_info = {
    'name': 'David',
    'email': '[email protected]',
    'reputation': 50
}

# 使用 hmset 批次寫入使用者資訊
r.hmset(user_id, user_info)

# 獲取單一屬性
name = r.hget(user_id, 'name')
print(f"使用者名稱: {name}") # 輸出: David

# 增加聲望值
r.hincrby(user_id, 'reputation', 10)

# 獲取所有資訊
all_info = r.hgetall(user_id)
print(f"所有使用者資訊: {all_info}") # 輸出: {'name': 'David', 'email': '[email protected]', 'reputation': '60'}

列表 (List)

List 是一個雙向鏈結串列，可以從頭部（左側）或尾部（右側）進行元素的推入（push）和彈出（pop）操作。這使得 List 非常適合實現訊息佇列、任務佇列或儲存時間序列數據（如最新的 N 條使用者動態）。

常用操作：

Python 方法	Redis 指令	說明
r.lpush(name, *values)	LPUSH	從列表左側（頭部）插入一個或多個元素。
r.rpush(name, *values)	RPUSH	從列表右側（尾部）插入一個或多個元素。
r.lpop(name)	LPOP	從列表左側彈出一個元素並返回它。
r.rpop(name)	RPOP	從列表右側彈出一個元素並返回它。
r.blpop(keys, timeout)	BLPOP	阻塞式地從多個列表的左側彈出元素，timeout 0 表示永不超時。
r.brpop(keys, timeout)	BRPOP	阻塞式地從多個列表的右側彈出元素。
r.lrange(name, start, end)	LRANGE	獲取指定範圍內的元素 (-1 表示最後一個)。
r.linsert(name, where, refvalue, value)	LINSERT	在列表的某個標杆值前或後插入新值。
r.llen(name)	LLEN	獲取列表長度或元素個數。
r.ltrim(name, start, end)	LTRIM	修剪列表，只保留指定索引號範圍內的元素。

程式碼範例：

task_queue_key = 'task_queue'

# 生產者：向任務佇列中左邊添加元素
r.lpush(task_queue_key, 'task1:send_email', 'task2:generate_report')
r.lpush(task_queue_key, 'task3:process_image')
# 現在列表內容 (從左至右): ['task3:process_image', 'task2:generate_report', 'task1:send_email']

# 消費者：從佇列中獲取並處理任務
while r.llen(task_queue_key) > 0:
    task = r.rpop(task_queue_key) # 從右側取出，實現先進先出 (FIFO)
    print(f"正在處理任務: {task}")
    # ... 處理任務的邏輯 ...

# 實現一個簡單的日誌系統，只保留最新的 5 條日誌
log_key = 'system_logs'
for i in range(10):
    r.lpush(log_key, f"Log entry {i}")
    r.ltrim(log_key, 0, 4) # 只保留索引號是 0 到 4 的元素

print(f"最新的 5 條日誌: {r.lrange(log_key, 0, -1)}")
# 輸出: ['Log entry 9', 'Log entry 8', 'Log entry 7', 'Log entry 6', 'Log entry 5']

集合 (Set)

Set 是一個無序且元素唯一的集合。它支援高效的成員檢查、新增、刪除以及伺服器端的交集、聯集、差集運算，非常適合用於標籤系統、好友關係、共同興趣等場景。

常用操作：

Python 方法	Redis 指令	說明
r.sadd(name, *values)	SADD	向集合中添加一個或多個成員。
r.srem(name, *values)	SREM	從集合中移除一個或多個成員。
r.smembers(name)	SMEMBERS	獲取集合中的所有成員。
r.sismember(name, value)	SISMEMBER	判斷一個值是否為集合的成員。
r.scard(name)	SCARD	獲取集合的基數（成員個數）。
r.spop(name, count)	SPOP	隨機彈出並返回指定數量的成員。
r.sinter(keys)	SINTER	計算多個集合的交集。
r.sunion(keys)	SUNION	計算多個集合的聯集。
r.sdiff(keys)	SDIFF	計算多個集合的差集。

程式碼範例：

# 為文章添加標籤
article_1_tags = 'tags:article:1'
article_2_tags = 'tags:article:2'
r.sadd(article_1_tags, 'python', 'redis', 'web')
r.sadd(article_2_tags, 'python', 'database', 'performance')

# 獲取文章1的所有標籤
print(f"文章1的標籤: {r.smembers(article_1_tags)}")

# 檢查文章1是否有 'redis' 標籤
print(f"文章1是否有 'redis' 標籤: {r.sismember(article_1_tags, 'redis')}") # 輸出: True

# 找出同時擁有 'python' 標籤的文章 (此處用集合交集模擬)
# 假設我們還有 'tag:python:articles' 這樣的反向索引集合
# 找出兩篇文章的共同標籤
common_tags = r.sinter([article_1_tags, article_2_tags])
print(f"兩篇文章的共同標籤: {common_tags}") # 輸出: {'python'}

# 找出兩篇文章的所有標籤（去重）
all_tags = r.sunion([article_1_tags, article_2_tags])
print(f"兩篇文章的所有標籤: {all_tags}")

有序集合 (Sorted Set / ZSet)

Sorted Set 是 Set 的增強版，它在每個成員上關聯了一個浮點數分數（score）。這使得 Sorted Set 不僅擁有 Set 的唯一性，還能根據分數對成員進行排序。它非常適合用來實現排行榜、帶權重的任務佇列等。

常用操作：

Python 方法	Redis 指令	說明
r.zadd(name, mapping)	ZADD	向有序集合添加一個或多個成員及其分數。mapping 是一個字典。
r.zrange(name, start, end, withscores=True)	ZRANGE	按分數從低到高返回指定排名範圍的成員 (start, end 為索引，非分數)。
r.zrevrange(name, start, end, withscores=True)	ZREVRANGE	按分數從高到低返回指定排名範圍的成員。
r.zrangebyscore(name, min, max, withscores=True)	ZRANGEBYSCORE	按分數範圍返回成員。
r.zcard(name)	ZCARD	獲取有序集合的成員數量。
r.zscore(name, value)	ZSCORE	獲取指定成員的分數。
r.zincrby(name, value, amount=1)	ZINCRBY	為指定成員的分數增加指定值。
r.zrank(name, value)	ZRANK	返回成員的排名（分數從低到高）。
r.zrevrank(name, value)	ZREVRANK	返回成員的排名（分數從高到低）。

程式碼範例：

leaderboard_key = 'game:leaderboard'

# 添加玩家分數
player_scores = {'player:1': 1500, 'player:2': 2200, 'player:3': 1850}
r.zadd(leaderboard_key, player_scores)

# 為 player:1 增加分數
r.zincrby(leaderboard_key, 'player:1', 100) # player:1 分數變為 1600

# 獲取分數最高的 Top 3 玩家
top_3 = r.zrevrange(leaderboard_key, 0, 2, withscores=True)
print("遊戲排行榜 Top 3:")
for player, score in top_3:
    print(f"- {player}: {int(score)}")

# 獲取 player:3 的排名
rank = r.zrevrank(leaderboard_key, 'player:3')
if rank is not None:
    print(f"player:3 的排名是: {rank + 1}")

進階功能與最佳實踐

管道 (Pipeline)

當你需要連續執行多個 Redis 指令時，每次指令都會產生一次網路來回（RTT）。Pipeline 技術可以將多個指令打包在一起，一次性發送到 Redis 伺服器，伺服器執行完畢後再將所有結果一次性返回。這能極大地減少網路延遲，提升吞吐量。

redis-py 的 Pipeline 預設是原子性的（使用 MULTI/EXEC 包裹），可以透過設定 transaction=False 來關閉事務特性，只享受批次處理帶來的好處。

import redis

r = redis.Redis(decode_responses=True)
pipe = r.pipeline() # 預設 transaction=True

# 將指令放入管道，但尚未執行
pipe.set('pipe_key1', 'value1')
pipe.incr('pipe_counter')
pipe.hset('pipe_hash', 'field1', 'h_value1')
pipe.get('pipe_key1')

# 一次性執行所有指令
# results 將是一個包含所有指令返回值的列表
results = pipe.execute()

print(f"管道執行結果: {results}")
# 輸出: [True, 1, 1, 'value1']
# 分別對應 set, incr, hset, get 的返回值

實戰案例：快取模式（Cache-Aside Pattern）

這是最常見的 redis cache 應用模式。當應用程式需要讀取數據時：

首先嘗試從 Redis（快取）中讀取。
如果快取中存在（快取命中），則直接返回數據。
如果快取中不存在（快取未命中），則從主數據庫（如 MySQL, PostgreSQL）中讀取。
將從資料庫中讀取的數據寫入 Redis，並設定一個合理的過期時間。
將數據返回給應用程式。

以下的 code 範例展示了此模式：

import time
import json

# 模擬資料庫
DB = {
    'product:123': {'name': 'Awesome Widget', 'price': 99.99, 'stock': 100}
}

def get_product_from_db(product_id):
    """模擬從資料庫讀取數據"""
    print(f" 讀取數據庫: {product_id} ")
    time.sleep(0.5) # 模擬 I/O 延遲
    return DB.get(product_id)

def get_product(product_id):
    r = redis.Redis(decode_responses=True)
    cache_key = f"cache:product:{product_id}"

    # 1. 嘗試從快取讀取
    cached_data = r.get(cache_key)
    if cached_data:
        print(f"* 快取命中: {product_id} *")
        return json.loads(cached_data)

    # 2. 快取未命中，從資料庫讀取
    print(f" 快取未命中: {product_id} ")
    db_data = get_product_from_db(product_id)

    # 3. 如果資料庫中有，則寫入快取
    if db_data:
        r.set(cache_key, json.dumps(db_data), ex=300) # 快取 5 分鐘

    return db_data

# 第一次呼叫
print(get_product('product:123'))
print("-" * 20)
# 第二次呼叫
print(get_product('product:123'))

常見問題 (FAQ)

Q1: redis.Redis() 和 redis.StrictRedis() 有什麼區別？

A1: 在舊版的 redis-py 中，StrictRedis 嚴格遵循官方 Redis 指令語法，而 Redis 則為了一些指令（如 LREM）提供了更 Pythonic 的參數順序。然而，從 redis-py 3.0 開始，Redis 類別已經是 StrictRedis 的別名。因此，在現代專案中，直接使用 redis.Redis 即可，它們的行為是完全一致的。

Q2: 為什麼我從 Redis 取出的值是位元組 (bytes) 而不是字串 (string)？

A2: 這是因為 Redis 內部儲存的是二進位安全的位元組序列。redis-py 預設會返回 bytes 型別以保持數據的原始性。若要讓客戶端套件自動將回應解碼為 UTF-8 字串，需要在建立連線時傳入 decoderesponses=True 參數，如 redis.Redis(decoderesponses=True)。這在大多數處理文字數據的場景中都非常方便。

Q3: 我應該使用單一連線還是連線池？

A3: 這取決於你的應用場景。對於一次性運行的短腳本，直接建立單一連線 redis.Redis() 是可以接受的。但對於需要頻繁與 Redis 互動的應用程式，如 Web 伺服器（Flask, Django, FastAPI）或長駐的背景服務，強烈建議使用連線池（redis.ConnectionPool）。連接池可以複用已建立的連線，避免了頻繁建立和銷毀 TCP 連線所帶來的效能損耗和資源浪費。

Q4: 在生產環境中使用 KEYS * 有什麼風險？

A4: KEYS 指令會遍歷 Redis 伺服器中的所有 key，這是一個阻塞操作。當資料庫中的 key 數量巨大時（數百萬或更多），執行 KEYS 會導致 Redis 伺服器在數秒甚至更長時間內無法回應其他請求，造成嚴重的效能問題，甚至服務中斷。在生產環境中，應絕對避免*使用 KEYS。請改用基於遊標的非阻塞迭代指令 SCAN。

Q5: 如何高效地迭代處理 Redis 中的大量資料？

A5: 對於大型的 Hash, Set, 或 Sorted Set，直接使用 HGETALL, SMEMBERS 等指令會一次性將所有數據載入記憶體，可能導致應用程式記憶體溢位。正確的做法是使用 SCAN 系列指令進行增量迭代，它們是非阻塞的，並且可以分批次取回數據。redis-py 提供了方便的迭代器封裝：

r.scan_iter(): 迭代所有 key。
r.hscan_iter(name): 迭代 Hash 中的欄位。
r.sscan_iter(name): 迭代 Set 中的成員。
r.zscan_iter(name): 迭代 Sorted Set 中的成員。

總結

Redis 以其卓越的效能和靈活的資料結構，成為了現代軟體架構中不可或缺的一環。透過 redis-py 函式庫，Python 開發者可以無縫地利用 Redis 的所有功能。

本文從基礎的安裝連線，到對 String, Hash, List, Set, Sorted Set 五大資料結構的詳細操作，再到 Pipeline 和快取模式等進階應用，提供了一條清晰的學習路徑。掌握這些知識，你將能夠自信地在專案中引入 Redis，顯著提升應用程式的反應速度和擴展性。

然而，Redis 的世界遠不止於此。更進階的主題如 Lua 腳本、Transactions、Streams、發佈/訂閱（Pub/Sub）以及 RediSearch 等模組，都等待著你去探索。希望本指南能為你的 Redis 之旅奠定堅實的基礎。

環境準備與安裝

安裝並啟動 Redis 伺服器

安裝 Python Redis 客戶端

連線到 Redis：基礎與進階模式

基本連線

使用連線池 (Connection Pool)

非同步連線 (Asyncio)

特殊連線模式

核心資料結構與 Python 操作詳解

字串 (String)

雜湊 (Hash)

列表 (List)

集合 (Set)

有序集合 (Sorted Set / ZSet)

進階功能與最佳實踐

管道 (Pipeline)

實戰案例：快取模式（Cache-Aside Pattern）

常見問題 (FAQ)

Q1: redis.Redis() 和 redis.StrictRedis() 有什麼區別？

Q2: 為什麼我從 Redis 取出的值是位元組 (bytes) 而不是字串 (string)？

Q3: 我應該使用單一連線還是連線池？

Q4: 在生產環境中使用 KEYS * 有什麼風險？

Q5: 如何高效地迭代處理 Redis 中的大量資料？

總結

資料來源

相關文章