使用Python操作Redis的完整指南

更新時間：2026年03月20日 08:28:07 作者：傻啦嘿喲

Redis作為業(yè)界主流的內存數(shù)據(jù)庫,配合Python的redis-py庫,能在不改變業(yè)務代碼結構的前提下,把數(shù)據(jù)庫的查詢壓力降低90%以上,今天我們就從零開始,聊聊怎么用Python操作Redis,搭一套真正能打的緩存系統(tǒng),需要的朋友可以參考下

上個月幫一個創(chuàng)業(yè)團隊排查線上事故，他們的電商活動頁在大促高峰期整整卡了十分鐘。監(jiān)控顯示數(shù)據(jù)庫連接數(shù)直接打滿，慢查詢堆積了上千條。創(chuàng)始人盯著屏幕問我：“不就是展示一下商品詳情嗎，怎么就把數(shù)據(jù)庫干崩了？”

我看了一眼代碼，每次請求都直連MySQL查商品信息，熱門商品被上千人同時刷，數(shù)據(jù)庫扛得住才怪。其實這個問題有個標準解法——緩存。

Redis作為業(yè)界主流的內存數(shù)據(jù)庫，配合Python的redis-py庫，能在不改變業(yè)務代碼結構的前提下，把數(shù)據(jù)庫的查詢壓力降低90%以上。今天我們就從零開始，聊聊怎么用Python操作Redis，搭一套真正能打的緩存系統(tǒng)。

基礎篇：先讓Redis跑起來

安裝redis-py只需要一行命令：

pip install redis

連接Redis的代碼也極其簡單：

import?redis
#?連接本地Redis
r?=?redis.Redis(host='localhost',?port=6379,?db=0,?decode_responses=True)
#?測試連接
r.set('foo',?'bar')
print(r.get('foo'))??#?輸出:?bar

這里有個小細節(jié)：decode_responses=True會讓返回的結果自動從字節(jié)串轉成字符串，省去手動decode的麻煩。如果是生產(chǎn)環(huán)境，建議用連接池管理連接，避免頻繁創(chuàng)建銷毀消耗資源：

pool?=?redis.ConnectionPool(host='localhost',?port=6379,?db=0,?max_connections=10)
r?=?redis.Redis(connection_pool=pool)

實戰(zhàn)篇：緩存最簡單的寫法

最常見的緩存場景是數(shù)據(jù)庫查詢。一個用戶信息服務，如果不加緩存，代碼長這樣：

def?get_user(user_id):
????#?直接查數(shù)據(jù)庫
????return?db.query(f"SELECT?*?FROM?users?WHERE?id={user_id}")

加一層Redis緩存，代碼變成這樣：

def?get_user(user_id):
????#?先查緩存
????cache_key?=?f"user:{user_id}"
????user?=?r.get(cache_key)
????
????if?user:
????????return?json.loads(user)??#?緩存命中，直接返回
????
????#?緩存未命中，查數(shù)據(jù)庫
????user?=?db.query(f"SELECT?*?FROM?users?WHERE?id={user_id}")
????
????#?寫入緩存，設置過期時間
????r.setex(cache_key,?3600,?json.dumps(user))
????return?user

這個模式叫Cache-Aside，是業(yè)界最通用的緩存策略。流程很簡單：讀的時候先讀緩存，沒有就查數(shù)據(jù)庫然后回寫；寫的時候先更新數(shù)據(jù)庫，然后刪除緩存（或者更新緩存）。

這里有兩個關鍵點。一是緩存要有過期時間。上面的setex設置了3600秒，避免緩存項永遠駐留導致數(shù)據(jù)不一致。二是key的命名規(guī)范。用user:1001這樣的格式，冒號分隔不同部分，在Redis里會自動按層級展示，調試時一目了然。

進階篇：用裝飾器把緩存寫成一行

上面的寫法已經(jīng)能解決問題，但還是不夠優(yōu)雅。每次都要手寫緩存key、手動序列化、手動處理異常，重復代碼太多。Python的裝飾器可以把這些臟活累活封裝起來。

一個最簡版緩存裝飾器可以這么寫：

from?functools?import?wraps
import?json

def?redis_cache(ttl=300):
????def?decorator(func):
????????@wraps(func)
????????def?wrapper(*args,?**kwargs):
????????????#?生成緩存key：函數(shù)名?+?參數(shù)
????????????key?=?f"{func.__name__}:{str(args)}:{str(kwargs)}"
????????????
????????????#?嘗試從緩存獲取
????????????cached?=?r.get(key)
????????????if?cached:
????????????????return?json.loads(cached)
????????????
????????????#?執(zhí)行原函數(shù)
????????????result?=?func(*args,?**kwargs)
????????????
????????????#?寫入緩存
????????????r.setex(key,?ttl,?json.dumps(result))
????????????return?result
????????return?wrapper
????return?decorator

#?使用
@redis_cache(ttl=600)
def?get_user(user_id):
????return?db.query(f"SELECT?*?FROM?users?WHERE?id={user_id}")

這樣一來，業(yè)務代碼完全不需要關心緩存邏輯，一個裝飾器搞定。實際項目中可以用更成熟的庫，比如redis_func_cache，它支持LRU、LFU等多種淘汰策略，還封裝好了Lua腳本保證原子性。

坑點篇：緩存穿透、擊穿、雪崩怎么破

緩存用不好，有時候比不用還糟糕。三個經(jīng)典問題值得留意。

緩存穿透指查詢一個根本不存在的數(shù)據(jù)。每次請求都繞過緩存直擊數(shù)據(jù)庫，如果被惡意利用，數(shù)據(jù)庫分分鐘被打掛。解決方案是緩存空值：

def?get_user(user_id):
????user?=?r.get(f"user:{user_id}")
????if?user?is?not?None:??#?注意：None表示緩存未命中，空字符串表示緩存了空值
????????return?user?if?user?!=?"NULL"?else?None
????
????user?=?db.query(...)
????#?無論查沒查到，都寫緩存
????r.setex(f"user:{user_id}",?600,?user?or?"NULL")
????return?user

緩存擊穿指某個熱點key過期瞬間，大量并發(fā)請求同時穿透到數(shù)據(jù)庫。用分布式鎖可以解決：

def?get_hot_data(key):
????data?=?r.get(key)
????if?data:
????????return?data
????
????#?加鎖，只允許一個線程去查數(shù)據(jù)庫
????with?r.lock(f"lock:{key}",?timeout=10):
????????#?雙重檢查：拿到鎖后可能已經(jīng)被其他線程更新了
????????data?=?r.get(key)
????????if?data:
????????????return?data
????????????
????????data?=?expensive_query()
????????r.setex(key,?3600,?data)
????????return?data

緩存雪崩指大量key同時過期，導致數(shù)據(jù)庫瞬時壓力暴增。解決方案是給過期時間加隨機偏移量：

import?random

#?基礎過期時間3600秒，加上0-300秒的隨機偏移
expire?=?3600?+?random.randint(0,?300)
r.setex(key,?expire,?value)

高階篇：多級緩存讓速度再翻倍

單靠Redis做緩存，每次請求還是有一次網(wǎng)絡開銷。如果能把最熱的數(shù)據(jù)放在應用本地內存里，速度能再快一個數(shù)量級。

這就是多級緩存架構：本地緩存（毫秒級）→ Redis集群（亞毫秒級）→ 數(shù)據(jù)庫（毫秒級）。80%的請求被本地緩存攔截，剩下的20%由Redis承載，數(shù)據(jù)庫幾乎只處理寫請求和緩存未命中的場景。

redis-py自帶了本地緩存模塊_LocalCache，可以搭配使用：

from?redis._cache?import?_LocalCache,?EvictionPolicy

#?初始化本地緩存：最多存10000條，30秒過期，LRU淘汰策略
local_cache?=?_LocalCache(max_size=10000,?ttl=30,?eviction_policy=EvictionPolicy.LRU)

def?get_user_with_multilevel_cache(user_id):
????#?構造命令元組作為緩存key
????command?=?("GET",?f"user:{user_id}")
????
????#?查本地緩存
????cached?=?local_cache.get(command)
????if?cached:
????????return?cached
????
????#?查Redis
????user?=?r.get(f"user:{user_id}")
????if?user:
????????#?寫入本地緩存
????????local_cache.set(command,?user,?keys_in_command=[f"user:{user_id}"])
????????return?user
????
????#?查數(shù)據(jù)庫
????user?=?db.query(...)
????r.setex(f"user:{user_id}",?3600,?user)
????return?user

這套架構在實踐中有幾個優(yōu)化點：熱點數(shù)據(jù)可以提前預熱，比如活動開始前把商品信息加載到緩存；監(jiān)控指標要跟上，重點關注本地緩存命中率（目標90%以上）和Redis查詢延遲（目標1ms以下）；數(shù)據(jù)更新時要同時淘汰兩級緩存，保證一致性。