python操作redis基礎(chǔ)

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

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

1. Redis 簡介

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

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

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

2. 前提條件

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

你可以使用 redis-cli ping 命令測試服務(wù)器是否響應(yīng)（應(yīng)返回 PONG）。

3. 安裝 Python Redis 客戶端庫

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

pip install redis

4. 連接到 Redis

方法一：直接連接

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

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

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

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

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)
    # 從連接池獲取一個(gè)連接
    r = redis.Redis(connection_pool=pool)
    # 測試連接
    response = r.ping()
    print(f"Redis PING response (from pool): {response}")
    print("成功通過連接池連接到 Redis!")
    # 使用連接池時(shí)，不需要手動(dòng)關(guān)閉單個(gè)連接 r.close()
    # 連接會(huì)在使用完畢后自動(dòng)返回池中
except redis.exceptions.ConnectionError as e:
    print(f"無法創(chuàng)建 Redis 連接池或連接失敗: {e}")
# 當(dāng)你的應(yīng)用程序退出時(shí)，可以考慮關(guān)閉整個(gè)連接池（雖然不總是必需）
# pool.disconnect()

decode_responses=True 的重要性:

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

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

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

5.1 String (字符串)

# --- String 操作 ---
print("\n--- String 操作 ---")
# 設(shè)置值 (SET key value)
# 如果鍵已存在，會(huì)覆蓋舊值
r.set('user:name', 'Alice')
print(f"設(shè)置 user:name: {r.get('user:name')}")
# 獲取值 (GET key)
name = r.get('user:name')
print(f"獲取 user:name: {name}")
# 設(shè)置帶過期時(shí)間的值 (SET key value EX seconds)
# 'counter' 將在 60 秒后自動(dòng)刪除
r.set('counter', 100, ex=60)
print(f"設(shè)置帶過期時(shí)間的 counter: {r.get('counter')}")
# 設(shè)置值，僅當(dāng)鍵不存在時(shí) (SET key value NX)
# 如果 'user:name' 已存在，setnx 返回 False，不設(shè)置
result_nx = r.setnx('user:name', 'Bob')
print(f"嘗試設(shè)置已存在的 user:name (setnx): {result_nx} (當(dāng)前值: {r.get('user:name')})")
result_nx_new = r.setnx('user:city', 'New York')
print(f"嘗試設(shè)置不存在的 user:city (setnx): {result_nx_new} (當(dāng)前值: {r.get('user:city')})")
# 遞增 (INCR key) - 對字符串表示的數(shù)字進(jìn)行加 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 (表示不存在)
# 設(shè)置鍵的過期時(shí)間 (EXPIRE key seconds)
r.set('temp_data', 'will disappear')
r.expire('temp_data', 5) # 5 秒后過期
print(f"設(shè)置 temp_data 過期時(shí)間為 5 秒")
# time.sleep(6) # 可以取消注釋這行來驗(yàn)證過期
# print(f"5 秒后 temp_data 的值: {r.get('temp_data')}") # 如果 sleep 了，這里會(huì)輸出 None
# 刪除鍵 (DEL key [key ...])
deleted_count = r.delete('user:city', 'temp_data') # 可以同時(shí)刪除多個(gè)
print(f"刪除了 {deleted_count} 個(gè)鍵 ('user:city', 'temp_data')")
print(f"刪除后 user:city 的值: {r.get('user:city')}") # 輸出 None

5.2 List (列表)

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

# --- 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 是第一個(gè)元素，-1 是最后一個(gè)元素
all_tasks = r.lrange(list_key, 0, -1)
print(f"當(dāng)前列表 ({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) - 獲取并移除第一個(gè)元素
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) - 獲取并移除最后一個(gè)元素
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} 個(gè)成員")
added_again = r.sadd(set_key, 'user2', 'user4') # 'user2' 已存在，不會(huì)重復(fù)添加
print(f"再次嘗試添加 'user2', 'user4'，新增 {added_again} 個(gè)") # 輸出 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} 個(gè)") # 輸出: 1
print(f"移除后集合成員: {r.smembers(set_key)}") # 輸出: {'user2', 'user3', 'user4'}

5.4 Hash (哈希/散列)

存儲(chǔ)鍵值對的集合，適合表示對象。

# --- Hash 操作 ---
print("\n--- Hash 操作 ---")
hash_key = 'user:1000' # 通常用 : 分隔表示層級(jí)
# 清理舊數(shù)據(jù)
r.delete(hash_key)
# 設(shè)置單個(gè)字段值 (HSET key field value)
r.hset(hash_key, 'name', 'Bob')
print(f"設(shè)置 Hash 字段 'name'")
# 同時(shí)設(shè)置多個(gè)字段值 (HSET key mapping)
user_data = {'email': 'bob@example.com', 'city': 'London'}
r.hset(hash_key, mapping=user_data)
print(f"同時(shí)設(shè)置 Hash 字段 'email', 'city'")
# 獲取單個(gè)字段值 (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} 個(gè)") # 輸出: 1
print(f"刪除字段后 Hash: {r.hgetall(hash_key)}") # 輸出: {'name': 'Bob', 'email': 'bob@example.com'}

5.5 Sorted Set (有序集合)

與 Set 類似，但每個(gè)成員都有一個(gè)分?jǐn)?shù)（score），Redis 會(huì)根據(jù)分?jǐn)?shù)排序。

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

6. 進(jìn)階用法

6.1 Pipeline (管道)

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

# --- Pipeline 操作 ---
print("\n--- Pipeline 操作 ---")
# 使用連接池獲取連接
r_pipe = redis.Redis(connection_pool=pool)
# 創(chuàng)建 Pipeline 對象
pipe = r_pipe.pipeline()
# 在 Pipeline 中緩存命令 (不會(huì)立即執(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 是一個(gè)列表，包含每個(gè)命令的執(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)聽一個(gè)或多個(gè)頻道，接收發(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)聽消息 (這是一個(gè)阻塞操作)
# p.listen() 返回一個(gè)生成器
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("收到停止信號(hào)，退出監(jiān)聽。")
                break # 可以根據(jù)特定消息退出循環(huán)
except KeyboardInterrupt:
    print("手動(dòng)中斷監(jiān)聽。")
finally:
    # 取消訂閱并關(guān)閉連接
    p.unsubscribe()
    p.close()
    pool.disconnect() # 關(guān)閉整個(gè)池
    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ā)送停止信號(hào)
count3 = r.publish('news_channel', 'stop')
print(f"向 'news_channel' 發(fā)布停止信號(hào)，接收者數(shù)量: {count3}")
pool.disconnect()

運(yùn)行 Pub/Sub: 先運(yùn)行 subscriber.py，它會(huì)阻塞并等待消息。然后運(yùn)行 publisher.py，你將在 subscriber.py 的控制臺(tái)中看到接收到的消息。

7. 最佳實(shí)踐與提示

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

這個(gè)教程涵蓋了 Python 操作 Redis 的大部分常用功能。根據(jù)你的具體需求，可以進(jìn)一步探索 Redis 的事務(wù)、Lua 腳本、Streams 等更高級(jí)的特性。記得查閱 redis-py 的官方文檔和 Redis 官方文檔以獲取最全面和最新的信息。

到此這篇關(guān)于python操作redis基礎(chǔ)的文章就介紹到這了,更多相關(guān)python操作redis內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: