pip install redis

import redis

# 創(chuàng)建 Redis 客戶(hù)端連接
r = redis.Redis(
    host='localhost',  # Redis 服務(wù)器地址
    port=6379,         # Redis 端口
    db=0,              # 數(shù)據(jù)庫(kù)編號(hào)（0~15）
    password=None,     # 密碼（若無(wú)密碼則為 None）
    decode_responses=True  # 自動(dòng)解碼返回值為字符串
)

# 測(cè)試連接
try:
    r.ping()
    print("成功連接到 Redis 服務(wù)器")
except redis.ConnectionError:
    print("無(wú)法連接 Redis 服務(wù)器")

# 添加/覆蓋鍵值對(duì)
result = r.set("username", "john_doe")
print(f"添加結(jié)果: {result}")  # 輸出: True

# 添加帶條件的鍵值對(duì)（僅當(dāng)鍵不存在時(shí)）
result = r.set("email", "john@example.com", nx=True)
print(f"條件添加結(jié)果: {result}")  # 鍵不存在時(shí)返回 True，存在時(shí)返回 None

# 批量添加多個(gè)鍵值對(duì)
data = {
    "user:1001:name": "Alice",
    "user:1001:age": "28",
    "user:1001:city": "London"
}
result = r.mset(data)
print(f"批量添加結(jié)果: {result}")  # 輸出: True

# 設(shè)置鍵值對(duì)并在60秒后自動(dòng)過(guò)期
result = r.setex("session_token", 60, "a1b2c3d4e5")
print(f"帶過(guò)期時(shí)間的添加結(jié)果: {result}")  # 輸出: True

# 檢查剩余生存時(shí)間
ttl = r.ttl("session_token")
print(f"剩余生存時(shí)間: {ttl}秒")  # 輸出: 60

# 創(chuàng)建大型數(shù)據(jù)集
large_data = {f"item_{i}": f"value_{i}" for i in range(10000)}

# 添加大型哈希表
result = r.hset("large:hash", mapping=large_data)
print(f"添加大型哈希表結(jié)果: {result}")  # 輸出: 添加的字段數(shù)量

# 刪除單個(gè)鍵
result = r.delete("username")
print(f"刪除單個(gè)鍵結(jié)果: {result}")  # 輸出: 1 (成功刪除)

# 刪除多個(gè)鍵
keys_to_delete = ["user:1001:name", "user:1001:age", "non_existent_key"]
result = r.delete(*keys_to_delete)
print(f"刪除多個(gè)鍵結(jié)果: {result}")  # 輸出: 2 (實(shí)際刪除的鍵數(shù))

# 異步刪除單個(gè)大鍵
result = r.unlink("large:hash")
print(f"UNLINK 單個(gè)鍵結(jié)果: {result}")  # 輸出: 1

# 異步刪除多個(gè)鍵
keys_to_unlink = ["session_token", "temp:data", "cache:item"]
result = r.unlink(*keys_to_unlink)
print(f"UNLINK 多個(gè)鍵結(jié)果: {result}")  # 輸出: 實(shí)際刪除的鍵數(shù)

import time

# 創(chuàng)建大型測(cè)試數(shù)據(jù)
large_hash = {f"key_{i}": f"value_{i}" for i in range(50000)}
r.hset("test:large:hash", mapping=large_hash)

# 測(cè)試 DELETE 性能
start = time.time()
r.delete("test:large:hash")
delete_duration = time.time() - start

# 重新創(chuàng)建數(shù)據(jù)
r.hset("test:large:hash", mapping=large_hash)

# 測(cè)試 UNLINK 性能
start = time.time()
r.unlink("test:large:hash")
unlink_duration = time.time() - start

print(f"DELETE 耗時(shí): {delete_duration:.4f}秒")
print(f"UNLINK 耗時(shí): {unlink_duration:.4f}秒")
print(f"性能差異: DELETE 比 UNLINK 慢 {delete_duration/unlink_duration:.1f}倍")

# 使用管道批量添加和刪除
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")
    
    # 執(zhí)行所有命令
    results = pipe.execute()

print(f"管道操作結(jié)果: {results}")

# 添加哈希表
r.hset("user:1002", mapping={
    "name": "Bob",
    "email": "bob@example.com",
    "age": "32"
})

# 獲取哈希表字段
name = r.hget("user:1002", "name")
print(f"用戶(hù)名: {name}")

# 刪除哈希表字段
r.hdel("user:1002", "age")

# 獲取所有字段
all_fields = r.hgetall("user:1002")
print(f"用戶(hù)數(shù)據(jù): {all_fields}")

# 檢查單個(gè)鍵是否存在
exists = r.exists("username")
print(f"鍵存在: {bool(exists)}")  # 輸出: True 或 False

# 檢查多個(gè)鍵是否存在
count = r.exists("key1", "key2", "key3")
print(f"存在的鍵數(shù)量: {count}")

# 監(jiān)控內(nèi)存使用情況
info = r.info("memory")
print(f"已用內(nèi)存: {info['used_memory_human']}")
print(f"待刪除對(duì)象: {info['lazyfree_pending_objects']}")

from redis.exceptions import ConnectionError, TimeoutError
import time

def safe_operation():
    attempts = 0
    max_attempts = 3
    
    while attempts < max_attempts:
        try:
            # 嘗試執(zhí)行操作
            return r.set("important:data", "critical_value", ex=30)
        except (ConnectionError, TimeoutError) as e:
            attempts += 1
            print(f"操作失敗 ({attempts}/{max_attempts}): {str(e)}")
            time.sleep(2 ** attempts)  # 指數(shù)退避
    
    print("操作失敗，達(dá)到最大重試次數(shù)")
    return False

safe_operation()

# 最終推薦操作模式
def redis_best_practice():
    # 添加數(shù)據(jù)
    r.set("app:status", "running", ex=3600)  # 帶過(guò)期時(shí)間
    r.mset({"config:theme": "dark", "config:lang": "en"})
    
    # 刪除數(shù)據(jù)
    r.unlink("old:cache:data")  # 大型數(shù)據(jù)
    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()