快捷導(dǎo)航

使用Python操作Redis的完整指南

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

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

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

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

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

基礎(chǔ)篇：先讓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

這里有個(gè)小細(xì)節(jié)：decode_responses=True會讓返回的結(jié)果自動從字節(jié)串轉(zhuǎn)成字符串，省去手動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)

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

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

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}")
????
????#?寫入緩存，設(shè)置過期時(shí)間
????r.setex(cache_key,?3600,?json.dumps(user))
????return?user

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

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

進(jìn)階篇：用裝飾器把緩存寫成一行

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

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

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è)務(wù)代碼完全不需要關(guān)心緩存邏輯，一個(gè)裝飾器搞定。實(shí)際項(xiàng)目中可以用更成熟的庫，比如redis_func_cache，它支持LRU、LFU等多種淘汰策略，還封裝好了Lua腳本保證原子性。

坑點(diǎn)篇：緩存穿透、擊穿、雪崩怎么破

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

緩存穿透指查詢一個(gè)根本不存在的數(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

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

def?get_hot_data(key):
????data?=?r.get(key)
????if?data:
????????return?data
????
????#?加鎖，只允許一個(gè)線程去查數(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í)過期，導(dǎo)致數(shù)據(jù)庫瞬時(shí)壓力暴增。解決方案是給過期時(shí)間加隨機(jī)偏移量：

import?random

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

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

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

這就是多級緩存架構(gòu)：本地緩存（毫秒級）→ 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):
????#?構(gòu)造命令元組作為緩存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

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

收尾

回到開頭那個(gè)創(chuàng)業(yè)團(tuán)隊(duì)的故事。后來幫他們把用戶信息和商品詳情都加了Redis緩存，數(shù)據(jù)庫連接數(shù)從打滿降到個(gè)位數(shù)，接口響應(yīng)時(shí)間從秒級降到幾十毫秒。技術(shù)負(fù)責(zé)人發(fā)了條朋友圈：“原來我們之前一直在用石器時(shí)代的方式寫代碼。”

Redis緩存的本質(zhì)很簡單——用內(nèi)存換速度，用空間換時(shí)間。但用好它需要理解背后的數(shù)據(jù)一致性、過期策略、并發(fā)控制這些細(xì)節(jié)。希望這篇文章能幫你把這些細(xì)節(jié)串起來，寫出真正高效的緩存代碼。

以上就是使用Python操作Redis的完整指南的詳細(xì)內(nèi)容，更多關(guān)于Python操作Redis的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: