一、環境準備
1. 安裝依賴庫
pip install redis
2. 連接 Redis 數據庫
import redis# 創建 Redis 客戶端連接
r = redis.Redis(host='localhost', # Redis 服務器地址port=6379, # Redis 端口db=0, # 數據庫編號（0~15）password=None, # 密碼（若無密碼則為 None）decode_responses=True # 自動解碼返回值為字符串
)# 測試連接
try:r.ping()print("成功連接到 Redis 服務器")
except redis.ConnectionError:print("無法連接 Redis 服務器")
二、添加鍵值對操作
1. 添加單個鍵值對
# 添加/覆蓋鍵值對
result = r.set("username", "john_doe")
print(f"添加結果: {result}") # 輸出: True# 添加帶條件的鍵值對（僅當鍵不存在時）
result = r.set("email", "john@example.com", nx=True)
print(f"條件添加結果: {result}") # 鍵不存在時返回 True，存在時返回 None
2. 批量添加鍵值對
# 批量添加多個鍵值對
data = {"user:1001:name": "Alice","user:1001:age": "28","user:1001:city": "London"
}
result = r.mset(data)
print(f"批量添加結果: {result}") # 輸出: True
3. 添加帶過期時間的鍵值對
# 設置鍵值對并在60秒后自動過期
result = r.setex("session_token", 60, "a1b2c3d4e5")
print(f"帶過期時間的添加結果: {result}") # 輸出: True# 檢查剩余生存時間
ttl = r.ttl("session_token")
print(f"剩余生存時間: {ttl}秒") # 輸出: 60
4. 添加大型鍵值對
# 創建大型數據集
large_data = {f"item_{i}": f"value_{i}" for i in range(10000)}# 添加大型哈希表
result = r.hset("large:hash", mapping=large_data)
print(f"添加大型哈希表結果: {result}") # 輸出: 添加的字段數量
三、刪除鍵值對操作
1. DELETE 命令（同步刪除）
# 刪除單個鍵
result = r.delete("username")
print(f"刪除單個鍵結果: {result}") # 輸出: 1 (成功刪除)# 刪除多個鍵
keys_to_delete = ["user:1001:name", "user:1001:age", "non_existent_key"]
result = r.delete(*keys_to_delete)
print(f"刪除多個鍵結果: {result}") # 輸出: 2 (實際刪除的鍵數)
2. UNLINK 命令（異步刪除）
# 異步刪除單個大鍵
result = r.unlink("large:hash")
print(f"UNLINK 單個鍵結果: {result}") # 輸出: 1# 異步刪除多個鍵
keys_to_unlink = ["session_token", "temp:data", "cache:item"]
result = r.unlink(*keys_to_unlink)
print(f"UNLINK 多個鍵結果: {result}") # 輸出: 實際刪除的鍵數
3. DELETE vs UNLINK 對比
| 特性 | DELETE 命令 | UNLINK 命令 |
|---|---|---|
| 執行方式 | 同步刪除（阻塞操作） | 異步刪除（非阻塞操作） |
| 適用場景 | 小型鍵值對 | 大型鍵值對（哈希、列表等） |
| 性能影響 | 可能阻塞服務器 | 后臺執行，不影響主線程 |
| 返回值 | 刪除的鍵數量 | 刪除的鍵數量 |
| 內存回收 | 立即回收 | 延遲回收 |
| Python方法 | .delete() | .unlink() |
4. 刪除性能對比測試
import time# 創建大型測試數據
large_hash = {f"key_{i}": f"value_{i}" for i in range(50000)}
r.hset("test:large:hash", mapping=large_hash)# 測試 DELETE 性能
start = time.time()
r.delete("test:large:hash")
delete_duration = time.time() - start# 重新創建數據
r.hset("test:large:hash", mapping=large_hash)# 測試 UNLINK 性能
start = time.time()
r.unlink("test:large:hash")
unlink_duration = time.time() - startprint(f"DELETE 耗時: {delete_duration:.4f}秒")
print(f"UNLINK 耗時: {unlink_duration:.4f}秒")
print(f"性能差異: DELETE 比 UNLINK 慢 {delete_duration/unlink_duration:.1f}倍")
四、高級操作技巧
1. 管道操作（批量執行）
# 使用管道批量添加和刪除
with r.pipeline() as pipe:# 批量添加pipe.set("counter", 0)pipe.incrby("counter", 100)pipe.set("status", "active")# 批量刪除pipe.delete("temp:data1", "temp:data2")pipe.unlink("large:cache")# 執行所有命令results = pipe.execute()print(f"管道操作結果: {results}")
2. 哈希表操作
# 添加哈希表
r.hset("user:1002", mapping={"name": "Bob","email": "bob@example.com","age": "32"
})# 獲取哈希表字段
name = r.hget("user:1002", "name")
print(f"用戶名: {name}")# 刪除哈希表字段
r.hdel("user:1002", "age")# 獲取所有字段
all_fields = r.hgetall("user:1002")
print(f"用戶數據: {all_fields}")
3. 鍵存在性檢查
# 檢查單個鍵是否存在
exists = r.exists("username")
print(f"鍵存在: {bool(exists)}") # 輸出: True 或 False# 檢查多個鍵是否存在
count = r.exists("key1", "key2", "key3")
print(f"存在的鍵數量: {count}")
五、最佳實踐與注意事項
1. 鍵操作選擇指南
- 小型字符串鍵：DELETE 或 UNLINK 均可
- 大型數據結構：始終使用 UNLINK
- 批量刪除操作：優先使用 UNLINK + 管道
- 需要立即釋放內存：使用 DELETE
- 高并發環境：優先使用 UNLINK
2. 內存管理建議
# 監控內存使用情況
info = r.info("memory")
print(f"已用內存: {info['used_memory_human']}")
print(f"待刪除對象: {info['lazyfree_pending_objects']}")
3. 錯誤處理與重試
from redis.exceptions import ConnectionError, TimeoutError
import timedef safe_operation():attempts = 0max_attempts = 3while attempts < max_attempts:try:# 嘗試執行操作return r.set("important:data", "critical_value", ex=30)except (ConnectionError, TimeoutError) as e:attempts += 1print(f"操作失敗 ({attempts}/{max_attempts}): {str(e)}")time.sleep(2 ** attempts) # 指數退避print("操作失敗，達到最大重試次數")return Falsesafe_operation()
4. 性能優化技巧
- 批量操作：使用 MSET 替代多個 SET，使用管道處理批量命令
- 鍵名設計：使用可讀的命名空間（如
user:1000:profile） - 過期時間：為臨時數據設置 TTL，避免手動刪除
- 異步刪除：大型數據刪除始終使用 UNLINK
- 連接復用：避免頻繁創建/關閉連接
六、總結與選擇建議
操作選擇矩陣
| 場景 | 推薦操作 | 替代方案 |
|---|---|---|
| 添加小數據 | SET | HSET（對象） |
| 添加大數據 | HSET/MSET | 分批次添加 |
| 添加臨時數據 | SETEX | SET + EXPIRE |
| 刪除小數據 | DELETE | UNLINK |
| 刪除大數據 | UNLINK | 無 |
| 批量操作 | 管道 + MSET/UNLINK | 單獨命令 |
核心要點總結
-
添加操作：
- 使用
set()添加單個鍵值對 - 使用
mset()批量添加多個鍵值對 - 使用
setex()添加帶過期時間的鍵值對
- 使用
-
刪除操作：
- 優先使用
unlink()進行刪除（尤其大型數據） - 僅在需要立即釋放內存時使用
delete() - 批量刪除時結合管道提高效率
- 優先使用
-
性能關鍵：
- UNLINK 比 DELETE 快數百倍（大型數據）
- 管道操作可減少網絡往返時間
- 合理設置過期時間減少手動刪除
-
最佳實踐：
- 生產環境默認使用 UNLINK
- 監控 lazyfree_pending_objects 指標
- 使用指數退避策略處理連接錯誤
# 最終推薦操作模式
def redis_best_practice():# 添加數據r.set("app:status", "running", ex=3600) # 帶過期時間r.mset({"config:theme": "dark", "config:lang": "en"})# 刪除數據r.unlink("old:cache:data") # 大型數據r.unlink("temp:session:1", "temp:session:2") # 批量刪除# 使用管道with r.pipeline() as pipe:pipe.incr("counter:requests")pipe.expire("counter:requests", 86400)pipe.unlink("obsolete:key")pipe.execute()
通過本指南，您應該能夠熟練地在 Python 中實現 Redis 的鍵值對添加和刪除操作，理解 DELETE 和 UNLINK 的核心區別，并在不同場景下選擇最優的操作策略。
)
 - 加法器)
