以下是使用 Python 操作 Redis 的完整整理，涵蓋基礎操作、高級功能及最佳實踐：
1. 安裝與連接
(1) 安裝庫
pip install redis
(2) 基礎連接
import redis# 創建連接池（推薦復用連接）
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, password='your_password', decode_responses=True # 自動解碼為字符串
)
client = redis.Redis(connection_pool=pool)# 測試連接
print(client.ping()) # 輸出 True 表示成功
(3) SSL 連接
client = redis.Redis(host='your-host',port=6380,ssl=True,ssl_certfile='/path/to/cert.crt',ssl_keyfile='/path/to/key.key',ssl_ca_certs='/path/to/ca.crt'
)
2. 基本數據類型操作
(1) 字符串（String）
# 設置值（帶過期時間）
client.set('user:1001', 'Alice', ex=3600) # ex:秒級過期，px:毫秒級# 獲取值
name = client.get('user:1001') # 返回字符串（若 decode_responses=True）# 自增/自減
client.incr('counter') # +1
client.incrby('counter', 5) # +5
client.decr('counter') # -1
(2) 哈希（Hash）
# 設置字段
client.hset('user:profile:1001', 'name', 'Alice')
client.hmset('user:profile:1001', {'age': 30, 'email': 'alice@example.com'})# 獲取字段
name = client.hget('user:profile:1001', 'name') # 'Alice'
all_data = client.hgetall('user:profile:1001') # {'name': 'Alice', 'age': '30'}# 刪除字段
client.hdel('user:profile:1001', 'email')
(3) 列表（List）
# 插入元素
client.lpush('tasks', 'task1') # 左側插入
client.rpush('tasks', 'task2') # 右側插入# 獲取元素
task = client.rpop('tasks') # 右側彈出 'task2'
all_tasks = client.lrange('tasks', 0, -1) # 獲取全部元素# 截斷列表
client.ltrim('tasks', 0, 2) # 保留前3個元素
(4) 集合（Set）
# 添加元素
client.sadd('tags', 'python', 'redis')# 集合運算
common_tags = client.sinter('tags1', 'tags2') # 交集
all_tags = client.sunion('tags1', 'tags2') # 并集# 隨機彈出元素
random_tag = client.spop('tags')
(5) 有序集合（ZSet）
# 添加元素（帶分數）
client.zadd('rank', {'Alice': 90, 'Bob': 85})# 獲取排名
rank = client.zrevrank('rank', 'Alice') # 降序排名（0為最高）# 范圍查詢
top3 = client.zrevrange('rank', 0, 2, withscores=True) # 前3名及分數
3. 高級功能
(1) 管道（Pipeline）
# 批量操作（減少網絡往返）
pipe = client.pipeline()
pipe.set('key1', 'value1')
pipe.incr('counter')
pipe.execute() # 返回 [True, 6]
(2) 事務（Transaction）
try:pipe = client.pipeline(transaction=True)pipe.multi() # 開啟事務pipe.set('balance:1001', 100)pipe.decrby('balance:1001', 50)pipe.incrby('balance:1002', 50)pipe.execute() # 原子性提交
except redis.exceptions.WatchError:print("事務沖突，需重試")
(3) 分布式鎖
def acquire_lock(lock_name, timeout=10):identifier = str(uuid.uuid4())end = time.time() + timeoutwhile time.time() < end:if client.set(lock_name, identifier, nx=True, ex=timeout):return identifiertime.sleep(0.001)return Falsedef release_lock(lock_name, identifier):script = """if redis.call('get', KEYS[1]) == ARGV[1] thenreturn redis.call('del', KEYS[1])elsereturn 0end"""return client.eval(script, 1, lock_name, identifier) == 1
(4) 發布訂閱
# 發布消息
client.publish('news', 'Breaking News!')# 訂閱頻道
pubsub = client.pubsub()
pubsub.subscribe('news')for message in pubsub.listen():if message['type'] == 'message':print(f"收到消息: {message['data']}")
4. 數據序列化與復雜對象存儲
(1) JSON 序列化
import jsonuser = {'id': 1001, 'name': 'Alice'}
client.set('user:1001', json.dumps(user))# 反序列化
data = client.get('user:1001')
if data:user_data = json.loads(data)
(2) 使用 MessagePack（更高效）
pip install msgpack
import msgpackuser_bytes = msgpack.packb(user)
client.set('user:1001', user_bytes)# 反序列化
data = client.get('user:1001')
if data:user_data = msgpack.unpackb(data)
5. 鍵管理與維護
(1) 掃描鍵（避免阻塞）
# 查找所有以 'cache:' 開頭的鍵
for key in client.scan_iter(match='cache:*', count=1000):print(key)
(2) 設置過期時間
client.expire('user:1001', 3600) # 設置過期時間（秒）
client.pexpire('user:1001', 3600000) # 毫秒級過期
(3) 批量刪除鍵
# 刪除所有以 'temp:' 開頭的鍵
keys = list(client.scan_iter(match='temp:*'))
if keys:client.delete(*keys)
6. 最佳實踐
(1) 連接池管理
? 使用 ConnectionPool 復用連接，避免頻繁創建/銷毀。
? 配置最大連接數：
pool = redis.ConnectionPool(max_connections=10)
(2) 性能優化
? 使用 pipeline 批量操作減少網絡延遲。
? 避免大 Key（如超過 1MB 的 String 或 Hash），拆分存儲。
(3) 安全建議
? 禁用高危命令（在 Redis 配置中）：
rename-command FLUSHDB ""
rename-command KEYS ""
? 使用低權限賬號運行 Redis。
(4) 監控與調試
? 查看慢查詢：
slow_logs = client.slowlog_get()
? 監控內存使用：
info = client.info('memory')
print(info['used_memory_human'])
7. 常用場景代碼示例
(1) 緩存穿透解決方案
def get_user(user_id):key = f'user:{user_id}'data = client.get(key)if data is None:# 查詢數據庫user = db.query_user(user_id)if user:client.set(key, json.dumps(user), ex=300)else:# 緩存空值防止穿透client.set(key, 'NULL', ex=60)elif data == 'NULL':return Noneelse:return json.loads(data)
(2) 排行榜實現
# 更新分數
client.zadd('leaderboard', {'Alice': 100, 'Bob': 90})# 獲取前10名
top10 = client.zrevrange('leaderboard', 0, 9, withscores=True)
總結
| 操作類型 | 核心方法 | 適用場景 |
|---|---|---|
| 基礎數據操作 | set/hset/lpush/sadd/zadd | 緩存、隊列、標簽、排行榜 |
| 事務與管道 | pipeline + multi | 批量操作、原子性執行 |
| 分布式鎖 | SET + NX + Lua 腳本 | 資源競爭控制 |
| 發布訂閱 | publish + subscribe | 實時消息通知 |
| 性能優化 | 連接池、Pipeline、異步刪除 | 高并發、大數據量場景 |
通過合理選擇數據結構和操作方式，Redis 能高效支撐緩存、隊列、計數器、實時分析等多樣化需求。建議結合業務場景靈活應用，并通過監控工具持續優化性能。
