python操作redis基礎

更新時間：2025年05月27日 11:07:16 作者：hiquant

Redis(Remote?Dictionary?Server)是一個開源的、基于內(nèi)存的鍵值對（Key-Value）存儲系統(tǒng),它通常用作數(shù)據(jù)庫、緩存和消息代理,這篇文章主要介紹了python操作redis基礎,需要的朋友可以參考下

1. Redis 簡介

Redis (Remote Dictionary Server) 是一個開源的、基于內(nèi)存的鍵值對（Key-Value）存儲系統(tǒng)。它通常用作數(shù)據(jù)庫、緩存和消息代理。由于數(shù)據(jù)存儲在內(nèi)存中，Redis 的讀寫速度非常快，適用于需要高性能數(shù)據(jù)訪問的場景。

支持的數(shù)據(jù)類型：

String (字符串): 最基本的數(shù)據(jù)類型，可以是任何數(shù)據(jù)，如文本、JSON、序列化對象等（最大 512MB）。
List (列表): 字符串列表，按插入順序排序。可以從兩端進行 push/pop 操作。
Set (集合): 無序的字符串集合，不允許重復成員。
Hash (哈希/散列): 包含鍵值對的集合，非常適合存儲對象。
Sorted Set (有序集合): 類似 Set，但每個成員都關(guān)聯(lián)一個 double 類型的分數(shù)（score），并根據(jù)分數(shù)進行排序。成員必須唯一，但分數(shù)可以重復。
以及其他更高級的類型如 Streams, Bitmaps, HyperLogLogs 等。

2. 前提條件

安裝 Python: 確保你的系統(tǒng)已安裝 Python (建議 3.6+)。
安裝 Redis 服務器: 你需要在你的機器上或可訪問的網(wǎng)絡上運行一個 Redis 服務器實例。
- Linux/macOS: 可以通過包管理器安裝（如 apt install redis-server, brew install redis）。
- Windows: 可以通過 WSL (Windows Subsystem for Linux) 安裝，或者下載官方不推薦但可用的 Windows 版本，或者使用 Docker。
- Docker: 推薦方式，跨平臺且易于管理：docker run --name my-redis -p 6379:6379 -d redis
啟動 Redis 服務器: 確保 Redis 服務正在運行。默認監(jiān)聽端口是 6379。

你可以使用 redis-cli ping 命令測試服務器是否響應（應返回 PONG）。

3. 安裝 Python Redis 客戶端庫

與 Redis 服務器交互最常用的 Python 庫是 redis-py。

pip install redis

4. 連接到 Redis

方法一：直接連接

這是最簡單的方式，適用于快速測試或簡單腳本。

import redis
try:
    # 連接到本地默認端口的 Redis
    # decode_responses=True: 讓 get() 等方法返回字符串而不是 bytes
    r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
    # 如果 Redis 設置了密碼
    # r = redis.Redis(host='your_redis_host', port=6379, db=0, password='your_password', decode_responses=True)
    # 測試連接
    response = r.ping()
    print(f"Redis PING response: {response}") # 應該打印 True
    print("成功連接到 Redis!")
except redis.exceptions.ConnectionError as e:
    print(f"無法連接到 Redis: {e}")
# 注意：使用完后，理論上應該關(guān)閉連接 r.close()，但在簡單腳本中不總是必需。
# 對于長時間運行的應用，強烈推薦使用連接池。

方法二：使用連接池 (Connection Pool) - 推薦

對于需要頻繁與 Redis 交互的應用程序（如 Web 服務），每次操作都建立和斷開連接會非常低效。連接池可以管理一組預先建立的連接，供應用程序復用。

import redis
try:
    # 創(chuàng)建連接池
    pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=True)
    # 如果有密碼:
    # pool = redis.ConnectionPool(host='your_redis_host', port=6379, db=0, password='your_password', decode_responses=True)
    # 從連接池獲取一個連接
    r = redis.Redis(connection_pool=pool)
    # 測試連接
    response = r.ping()
    print(f"Redis PING response (from pool): {response}")
    print("成功通過連接池連接到 Redis!")
    # 使用連接池時，不需要手動關(guān)閉單個連接 r.close()
    # 連接會在使用完畢后自動返回池中
except redis.exceptions.ConnectionError as e:
    print(f"無法創(chuàng)建 Redis 連接池或連接失敗: {e}")
# 當你的應用程序退出時，可以考慮關(guān)閉整個連接池（雖然不總是必需）
# pool.disconnect()

decode_responses=True 的重要性:

如果不設置 decode_responses=True (默認為 False)，get()、lrange()、smembers() 等方法返回的是字節(jié)串 (bytes)，例如 b'value'。
設置 decode_responses=True 后，這些方法會嘗試使用指定的編碼（默認為 utf-8）將字節(jié)串解碼為字符串 (str)，例如 'value'，這通常更方便處理。

5. 操作 Redis 數(shù)據(jù)類型

假設我們已經(jīng)通過連接池獲取了 r 對象 (r = redis.Redis(connection_pool=pool))。

5.1 String (字符串)

# --- String 操作 ---
print("\n--- String 操作 ---")
# 設置值 (SET key value)
# 如果鍵已存在，會覆蓋舊值
r.set('user:name', 'Alice')
print(f"設置 user:name: {r.get('user:name')}")
# 獲取值 (GET key)
name = r.get('user:name')
print(f"獲取 user:name: {name}")
# 設置帶過期時間的值 (SET key value EX seconds)
# 'counter' 將在 60 秒后自動刪除
r.set('counter', 100, ex=60)
print(f"設置帶過期時間的 counter: {r.get('counter')}")
# 設置值，僅當鍵不存在時 (SET key value NX)
# 如果 'user:name' 已存在，setnx 返回 False，不設置
result_nx = r.setnx('user:name', 'Bob')
print(f"嘗試設置已存在的 user:name (setnx): {result_nx} (當前值: {r.get('user:name')})")
result_nx_new = r.setnx('user:city', 'New York')
print(f"嘗試設置不存在的 user:city (setnx): {result_nx_new} (當前值: {r.get('user:city')})")
# 遞增 (INCR key) - 對字符串表示的數(shù)字進行加 1
r.set('visit_count', '0')
r.incr('visit_count')
r.incr('visit_count')
print(f"遞增 visit_count: {r.get('visit_count')}") # 輸出 '2'
# 按指定步長遞增 (INCRBY key increment)
r.incrby('visit_count', 10)
print(f"遞增 visit_count by 10: {r.get('visit_count')}") # 輸出 '12'
# 遞減 (DECR key / DECRBY key decrement)
r.decr('visit_count')
print(f"遞減 visit_count: {r.get('visit_count')}") # 輸出 '11'
# 檢查鍵是否存在 (EXISTS key)
exists = r.exists('user:name')
print(f"user:name 是否存在: {exists}") # 輸出 1 (表示存在)
exists_non = r.exists('nonexistent_key')
print(f"nonexistent_key 是否存在: {exists_non}") # 輸出 0 (表示不存在)
# 設置鍵的過期時間 (EXPIRE key seconds)
r.set('temp_data', 'will disappear')
r.expire('temp_data', 5) # 5 秒后過期
print(f"設置 temp_data 過期時間為 5 秒")
# time.sleep(6) # 可以取消注釋這行來驗證過期
# print(f"5 秒后 temp_data 的值: {r.get('temp_data')}") # 如果 sleep 了，這里會輸出 None
# 刪除鍵 (DEL key [key ...])
deleted_count = r.delete('user:city', 'temp_data') # 可以同時刪除多個
print(f"刪除了 {deleted_count} 個鍵 ('user:city', 'temp_data')")
print(f"刪除后 user:city 的值: {r.get('user:city')}") # 輸出 None

5.2 List (列表)

Redis 列表是有序的字符串序列，類似 Python 列表，但操作主要在兩端進行。

# --- List 操作 ---
print("\n--- List 操作 ---")
list_key = 'tasks'
# 清理舊數(shù)據(jù)（如果存在）
r.delete(list_key)
# 從左側(cè)推入元素 (LPUSH key element [element ...])
r.lpush(list_key, 'task3', 'task2', 'task1') # 插入順序: task1, task2, task3
print(f"從左側(cè)推入 'task1', 'task2', 'task3'")
# 從右側(cè)推入元素 (RPUSH key element [element ...])
r.rpush(list_key, 'task4', 'task5') # 插入順序: task4, task5
print(f"從右側(cè)推入 'task4', 'task5'")
# 獲取列表范圍 (LRANGE key start stop) - 獲取所有元素
# 0 是第一個元素，-1 是最后一個元素
all_tasks = r.lrange(list_key, 0, -1)
print(f"當前列表 ({list_key}): {all_tasks}") # 輸出: ['task1', 'task2', 'task3', 'task4', 'task5'] (注意 LPUSH/RPUSH 的順序)
# 獲取列表長度 (LLEN key)
length = r.llen(list_key)
print(f"列表長度: {length}") # 輸出: 5
# 從左側(cè)彈出元素 (LPOP key) - 獲取并移除第一個元素
first_task = r.lpop(list_key)
print(f"從左側(cè)彈出: {first_task}") # 輸出: 'task1'
print(f"彈出后列表: {r.lrange(list_key, 0, -1)}") # 輸出: ['task2', 'task3', 'task4', 'task5']
# 從右側(cè)彈出元素 (RPOP key) - 獲取并移除最后一個元素
last_task = r.rpop(list_key)
print(f"從右側(cè)彈出: {last_task}") # 輸出: 'task5'
print(f"彈出后列表: {r.lrange(list_key, 0, -1)}") # 輸出: ['task2', 'task3', 'task4']
# 獲取指定索引的元素 (LINDEX key index)
task_at_index_1 = r.lindex(list_key, 1)
print(f"索引 1 處的元素: {task_at_index_1}") # 輸出: 'task3'

5.3 Set (集合)

無序、唯一的字符串集合。

# --- Set 操作 ---
print("\n--- Set 操作 ---")
set_key = 'unique_users'
# 清理舊數(shù)據(jù)
r.delete(set_key)
# 添加成員 (SADD key member [member ...]) - 返回成功添加的新成員數(shù)量
added_count = r.sadd(set_key, 'user1', 'user2', 'user3')
print(f"向集合添加了 {added_count} 個成員")
added_again = r.sadd(set_key, 'user2', 'user4') # 'user2' 已存在，不會重復添加
print(f"再次嘗試添加 'user2', 'user4'，新增 {added_again} 個") # 輸出 1
# 獲取所有成員 (SMEMBERS key)
members = r.smembers(set_key)
print(f"集合所有成員: {members}") # 輸出: {'user1', 'user2', 'user3', 'user4'} (注意是無序的)
# 檢查成員是否存在 (SISMEMBER key member)
is_member = r.sismember(set_key, 'user3')
print(f"'user3' 是否是集合成員: {is_member}") # 輸出: True
is_member_no = r.sismember(set_key, 'user5')
print(f"'user5' 是否是集合成員: {is_member_no}") # 輸出: False
# 獲取集合成員數(shù)量 (SCARD key)
count = r.scard(set_key)
print(f"集合成員數(shù)量: {count}") # 輸出: 4
# 移除成員 (SREM key member [member ...]) - 返回成功移除的數(shù)量
removed_count = r.srem(set_key, 'user1', 'user5') # 'user5' 不存在，只移除 'user1'
print(f"嘗試移除 'user1', 'user5'，成功移除 {removed_count} 個") # 輸出: 1
print(f"移除后集合成員: {r.smembers(set_key)}") # 輸出: {'user2', 'user3', 'user4'}

5.4 Hash (哈希/散列)

存儲鍵值對的集合，適合表示對象。

# --- Hash 操作 ---
print("\n--- Hash 操作 ---")
hash_key = 'user:1000' # 通常用 : 分隔表示層級
# 清理舊數(shù)據(jù)
r.delete(hash_key)
# 設置單個字段值 (HSET key field value)
r.hset(hash_key, 'name', 'Bob')
print(f"設置 Hash 字段 'name'")
# 同時設置多個字段值 (HSET key mapping)
user_data = {'email': 'bob@example.com', 'city': 'London'}
r.hset(hash_key, mapping=user_data)
print(f"同時設置 Hash 字段 'email', 'city'")
# 獲取單個字段值 (HGET key field)
name = r.hget(hash_key, 'name')
email = r.hget(hash_key, 'email')
print(f"獲取 Hash 字段 'name': {name}") # 輸出: 'Bob'
print(f"獲取 Hash 字段 'email': {email}") # 輸出: 'bob@example.com'
# 獲取所有字段和值 (HGETALL key) - 返回字典
all_fields = r.hgetall(hash_key)
print(f"獲取 Hash 所有字段和值: {all_fields}") # 輸出: {'name': 'Bob', 'email': 'bob@example.com', 'city': 'London'}
# 獲取所有字段名 (HKEYS key)
keys = r.hkeys(hash_key)
print(f"獲取 Hash 所有字段名: {keys}") # 輸出: ['name', 'email', 'city']
# 獲取所有值 (HVALS key)
values = r.hvals(hash_key)
print(f"獲取 Hash 所有值: {values}") # 輸出: ['Bob', 'bob@example.com', 'London']
# 檢查字段是否存在 (HEXISTS key field)
exists = r.hexists(hash_key, 'city')
print(f"Hash 字段 'city' 是否存在: {exists}") # 輸出: True
# 刪除字段 (HDEL key field [field ...]) - 返回成功刪除的字段數(shù)量
deleted_fields = r.hdel(hash_key, 'city', 'nonexistent_field')
print(f"嘗試刪除 Hash 字段 'city', 'nonexistent_field'，成功刪除 {deleted_fields} 個") # 輸出: 1
print(f"刪除字段后 Hash: {r.hgetall(hash_key)}") # 輸出: {'name': 'Bob', 'email': 'bob@example.com'}

5.5 Sorted Set (有序集合)

與 Set 類似，但每個成員都有一個分數(shù)（score），Redis 會根據(jù)分數(shù)排序。

# --- Sorted Set 操作 ---
print("\n--- Sorted Set 操作 ---")
zset_key = 'leaderboard'
# 清理舊數(shù)據(jù)
r.delete(zset_key)
# 添加成員和分數(shù) (ZADD key {member1: score1, member2: score2 ...})
# 如果成員已存在，會更新其分數(shù)
r.zadd(zset_key, {'player1': 1500, 'player2': 2100, 'player3': 1800})
print(f"添加成員到有序集合")
r.zadd(zset_key, {'player1': 1650}) # 更新 player1 的分數(shù)
print(f"更新 player1 的分數(shù)")
# 按分數(shù)范圍獲取成員 (ZRANGEBYSCORE key min max [WITHSCORES=True])
# (1000, 2000] 表示分數(shù) > 1000 且 <= 2000
players_in_range = r.zrangebyscore(zset_key, '(1000', 2000) # 默認不包含分數(shù)
print(f"分數(shù)在 (1000, 2000] 之間的玩家: {players_in_range}") # 輸出: ['player1', 'player3']
players_with_scores = r.zrangebyscore(zset_key, 1000, 2000, withscores=True)
print(f"分數(shù)在 [1000, 2000] 之間的玩家 (帶分數(shù)): {players_with_scores}") # 輸出: [('player1', 1650.0), ('player3', 1800.0)]
# 按排名范圍獲取成員 (ZRANGE key start stop [WITHSCORES=True])
# 0 是排名第一（分數(shù)最低），-1 是排名最后（分數(shù)最高）
top_players = r.zrange(zset_key, 0, -1, desc=True) # desc=True 按分數(shù)從高到低排
print(f"所有玩家按分數(shù)降序排列: {top_players}") # 輸出: ['player2', 'player3', 'player1']
top_2_with_scores = r.zrange(zset_key, 0, 1, desc=True, withscores=True)
print(f"排名前 2 的玩家 (帶分數(shù)): {top_2_with_scores}") # 輸出: [('player2', 2100.0), ('player3', 1800.0)]
# 獲取成員的分數(shù) (ZSCORE key member)
score_player2 = r.zscore(zset_key, 'player2')
print(f"'player2' 的分數(shù): {score_player2}") # 輸出: 2100.0
# 獲取成員數(shù)量 (ZCARD key)
num_players = r.zcard(zset_key)
print(f"有序集合成員數(shù)量: {num_players}") # 輸出: 3
# 移除成員 (ZREM key member [member ...]) - 返回成功移除的數(shù)量
removed_count = r.zrem(zset_key, 'player1', 'nonexistent')
print(f"嘗試移除 'player1', 'nonexistent'，成功移除 {removed_count} 個") # 輸出: 1
print(f"移除后有序集合 (降序): {r.zrange(zset_key, 0, -1, desc=True, withscores=True)}")

6. 進階用法

6.1 Pipeline (管道)

當你需要連續(xù)執(zhí)行多個 Redis 命令時，每次命令都需要一次網(wǎng)絡往返 (Client <-> Server)。使用 Pipeline 可以將多個命令打包一次性發(fā)送給服務器，服務器執(zhí)行完所有命令后再將結(jié)果一次性返回，從而顯著減少網(wǎng)絡延遲，提高性能。

# --- Pipeline 操作 ---
print("\n--- Pipeline 操作 ---")
# 使用連接池獲取連接
r_pipe = redis.Redis(connection_pool=pool)
# 創(chuàng)建 Pipeline 對象
pipe = r_pipe.pipeline()
# 在 Pipeline 中緩存命令 (不會立即執(zhí)行)
pipe.set('pipe_test:name', 'Pipeline User')
pipe.incr('pipe_test:counter', 1)
pipe.expire('pipe_test:name', 30)
pipe.get('pipe_test:name')
pipe.get('pipe_test:counter')
# 一次性執(zhí)行所有緩存的命令
# results 是一個列表，包含每個命令的執(zhí)行結(jié)果，順序與添加順序一致
results = pipe.execute()
print(f"Pipeline 執(zhí)行結(jié)果: {results}")
# 可能輸出: [True, 1, True, 'Pipeline User', '1'] (具體值取決于 counter 之前的值)
# 清理
r_pipe.delete('pipe_test:name', 'pipe_test:counter')

6.2 發(fā)布/訂閱 (Pub/Sub)

Redis 提供簡單的發(fā)布/訂閱消息模式。

發(fā)布者 (Publisher): 向指定頻道 (Channel) 發(fā)送消息。
訂閱者 (Subscriber): 監(jiān)聽一個或多個頻道，接收發(fā)送到這些頻道的消息。

訂閱者代碼 (subscriber.py):

import redis
import time
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=True)
r = redis.Redis(connection_pool=pool)
# 創(chuàng)建 PubSub 對象
p = r.pubsub()
# 訂閱頻道 'news_channel' 和 'alerts_channel'
p.subscribe('news_channel', 'alerts_channel')
print("開始監(jiān)聽頻道 'news_channel' 和 'alerts_channel'...")
# 持續(xù)監(jiān)聽消息 (這是一個阻塞操作)
# p.listen() 返回一個生成器
try:
    for message in p.listen():
        print(f"收到消息: {message}")
        # 消息格式通常是:
        # {'type': 'subscribe', 'pattern': None, 'channel': 'news_channel', 'data': 1}
        # {'type': 'message', 'pattern': None, 'channel': 'news_channel', 'data': 'Hello World!'}
        # 這里可以根據(jù) message['type'] 和 message['channel'] 處理不同消息
        if message['type'] == 'message':
            print(f"  頻道 [{message['channel']}] 收到數(shù)據(jù): {message['data']}")
            if message['data'] == 'stop':
                print("收到停止信號，退出監(jiān)聽。")
                break # 可以根據(jù)特定消息退出循環(huán)
except KeyboardInterrupt:
    print("手動中斷監(jiān)聽。")
finally:
    # 取消訂閱并關(guān)閉連接
    p.unsubscribe()
    p.close()
    pool.disconnect() # 關(guān)閉整個池
    print("監(jiān)聽已停止，連接已關(guān)閉。")

發(fā)布者代碼 (publisher.py):

import redis
import time
pool = redis.ConnectionPool(host='localhost', port=6379, db=0, decode_responses=True)
r = redis.Redis(connection_pool=pool)
# 向 'news_channel' 發(fā)布消息
count1 = r.publish('news_channel', 'Breaking News: Redis is awesome!')
print(f"向 'news_channel' 發(fā)布消息，接收者數(shù)量: {count1}")
time.sleep(1)
# 向 'alerts_channel' 發(fā)布消息
count2 = r.publish('alerts_channel', 'System Alert: High CPU usage detected!')
print(f"向 'alerts_channel' 發(fā)布消息，接收者數(shù)量: {count2}")
time.sleep(1)
# 發(fā)送停止信號
count3 = r.publish('news_channel', 'stop')
print(f"向 'news_channel' 發(fā)布停止信號，接收者數(shù)量: {count3}")
pool.disconnect()

運行 Pub/Sub: 先運行 subscriber.py，它會阻塞并等待消息。然后運行 publisher.py，你將在 subscriber.py 的控制臺中看到接收到的消息。

7. 最佳實踐與提示

使用連接池: 對于生產(chǎn)環(huán)境或需要頻繁交互的應用，務必使用連接池。decode_responses=True: 大多數(shù)情況下設置此選項能簡化代碼，避免手動解碼 bytes。
鍵名設計 (Key Naming): 使用有意義且結(jié)構(gòu)化的鍵名，常用 object-type:id:field 格式（如 user:1000:profile）。保持一致性。
錯誤處理: 使用 try...except 捕獲可能的 redis.exceptions.ConnectionError、redis.exceptions.TimeoutError 等異常。
大 Value 處理: 避免在 Redis 中存儲過大的單個 Value（幾 MB 或更大），這可能影響性能和內(nèi)存使用。考慮拆分或使用其他存儲。
原子性: Redis 的單個命令是原子的。對于需要多個命令組合的原子操作，考慮使用 Lua 腳本 (通過 r.eval() 或 r.register_script()) 或 Redis Transactions (通過 Pipeline 的 multi() 和 exec())。
資源管理: 確保在適當?shù)臅r候（如應用退出時）斷開連接或關(guān)閉連接池 (pool.disconnect())，尤其是在非長時間運行的腳本中。