在現代互聯網系統中，高并發已經成為常態。無論是電商的秒殺場景、社交平臺的熱點推薦，還是支付接口的風控，系統需要同時應對成千上萬的請求。這時候，Redis 作為一個高性能的內存數據庫，憑借其極快的讀寫速度和豐富的數據結構，成為解決高并發問題的利器。
1. 緩存熱點數據 —— 緩解數據庫壓力
問題背景：數據庫在高并發場景下容易成為瓶頸，特別是熱點數據的訪問會導致 QPS 瞬間飆升。
Redis 解決方案：
使用
String或Hash緩存熱點數據。采用 Cache Aside 模式：先查緩存，緩存未命中再查數據庫并回填。
示例代碼：
import redisr = redis.Redis(host='localhost', port=6379, db=0)def get_data_from_db(key):print("查詢數據庫")return {"id": key, "value": f"data_{key}"}def get_data(key):cache_key = f"cache:{key}"val = r.get(cache_key)if val:print("命中緩存")return eval(val)data = get_data_from_db(key)r.setex(cache_key, 60, str(data))return dataprint(get_data(1))
print(get_data(1))
在高并發場景下使用 Cache-Aside（旁路緩存） 模式，確實需要特別注意一些坑點和邊界情況。主要有以下幾個方面：
1. 緩存擊穿（Cache Miss Under High Concurrency）
問題：某個熱點 key 過期瞬間，大量請求并發打到數據庫，造成瞬時壓力飆升。
解決：
加 互斥鎖（分布式鎖） 控制只允許一個請求回源數據庫并更新緩存。
或者使用 邏輯過期：緩存里放置過期時間戳，過期時先返回舊值，再異步刷新。
2. 緩存穿透（Cache Penetration）
問題：請求的數據在數據庫中不存在，每次都會查詢 DB，導致 DB 壓力大。
解決：
緩存空對象（短 TTL），避免重復訪問 DB。
增加 布隆過濾器 或 前置攔截機制，快速判斷 key 是否可能存在。
3. 緩存雪崩（Cache Avalanche）
問題：大量緩存 key 在同一時間點同時過期，引發請求全部打到 DB。
解決：
給 key 設置 隨機過期時間，避免同時失效。
使用 多級緩存（L1/L2） 減輕瞬時壓力。
4. 數據一致性（Consistency）
問題：Cache-Aside 是 最終一致性，更新邏輯里 “先寫 DB 再刪緩存” 會有時間窗口問題。
解決：
常見策略是 先更新 DB，再刪除緩存（推薦），避免臟讀。
或者使用 延時雙刪（寫 DB → 刪緩存 → 延遲一小段時間再刪一次緩存）。
若業務對一致性要求極高，可能需要 訂閱 binlog（如 Canal）異步更新緩存。
5. 熱點 Key 與大 Value
熱點 Key：可能被頻繁訪問，甚至成為單點瓶頸。可通過 分片/多副本 緩解。
大 Value：更新頻率低但體積大時，更新緩存成本高，可以考慮 局部字段緩存 或 拆分存儲。
6. 并發控制 & 原子性
多線程并發下需要注意 讀寫競爭：避免多個線程同時回源刷新。
分布式場景下要用 原子操作（如 Redis 的 setnx + expire） 做鎖和限流。
7. 監控與降級
監控緩存命中率、請求 QPS、DB 壓力。
當緩存大面積失效時，可以考慮：
啟動 限流/熔斷。
提供 降級數據（兜底邏輯）。
應用場景：用戶信息、商品詳情頁、首頁推薦數據等。
2. 限流 —— 防止接口過載
問題背景：突發流量可能會壓垮下游服務，例如秒殺、登錄、支付接口。
Redis 解決方案：
使用
String作為計數器：INCR+EXPIRE實現固定時間窗口限流。使用
Sorted Set存儲時間戳：實現滑動窗口限流，更加精細化。可以配合 Lua 腳本保證原子性操作。
應用場景：API 調用限流、防止刷單、短信/驗證碼接口保護。
示例代碼（滑動窗口限流）：
import time
import redisr = redis.Redis(host='localhost', port=6379, db=0)def is_allowed_sliding_window(user_id, action, period=60, max_count=5):key = f"rate_limit:{user_id}:{action}"now = time.time()pipeline = r.pipeline()pipeline.zadd(key, {now: now})pipeline.zremrangebyscore(key, 0, now - period)pipeline.zcard(key)pipeline.expire(key, period)_, _, count, _ = pipeline.execute()return count <= max_countfor i in range(10):if is_allowed_sliding_window("u1001", "login"):print(i, "允許")else:print(i, "拒絕")time.sleep(0.5)
或者 使用 String 作為計數器
import redis
import timer = redis.Redis(host='localhost', port=6379, db=0)def is_allowed(user_id, action, period=60, max_count=5):"""在 period 時間內最多允許 max_count 次操作"""key = f"rate_limit:{user_id}:{action}"count = r.incr(key)if count == 1:r.expire(key, period) # 設置過期時間if count > max_count:return Falsereturn True# 測試
for i in range(10):if is_allowed("u1001", "login"):print(i, "允許")else:print(i, "拒絕")time.sleep(0.5)
3. 分布式隊列 —— 異步削峰
問題背景：大量請求直接打到數據庫或下游服務，可能會瞬間撐爆系統。
Redis 解決方案：
使用
List（LPUSH+BRPOP）作為簡單任務隊列。使用
Stream實現類似 Kafka 的消息隊列：支持消費組、消息確認和持久化。常見用途：削峰填谷，將請求轉為異步任務，讓系統有更多緩沖時間。
應用場景：訂單處理、日志收集、消息通知、秒殺排隊。
示例代碼（Stream 隊列）：
import redisr = redis.Redis(host='localhost', port=6379, db=0)
stream_key = "order_stream"# 生產者
def produce():for i in range(5):r.xadd(stream_key, {"order": f"order_{i}"})print("生產訂單:", f"order_{i}")# 消費者
def consume():last_id = "0-0"while True:messages = r.xread({stream_key: last_id}, block=2000, count=2)if not messages:breakfor _, msgs in messages:for msg_id, data in msgs:print("消費訂單:", msg_id, data)last_id = msg_idproduce()
consume()
4. 分布式鎖 —— 保證數據一致性
問題背景：在分布式環境中，多實例同時修改同一資源，可能造成數據沖突。
Redis 解決方案：
使用
String類型：SET key value NX PX timeout實現加鎖，避免重復設置。校驗
value防止誤刪鎖。
高級方案：使用 RedLock 算法 提升可靠性，避免單點故障。
應用場景：庫存扣減、任務調度、并發下的冪等操作。
示例代碼：
import redis
import uuid
import timer = redis.Redis(host='localhost', port=6379, db=0)def acquire_lock(lock_key, expire=5):lock_id = str(uuid.uuid4())result = r.set(lock_key, lock_id, nx=True, ex=expire)return lock_id if result else Nonedef release_lock(lock_key, lock_id):script = """if redis.call('get', KEYS[1]) == ARGV[1] thenreturn redis.call('del', KEYS[1])elsereturn 0end"""return r.eval(script, 1, lock_key, lock_id)lock_id = acquire_lock("my_lock", expire=3)
if lock_id:print("獲得鎖，執行業務邏輯...")time.sleep(2)print("釋放鎖:", release_lock("my_lock", lock_id))
else:print("未獲得鎖")
5. 實時統計與排行榜
問題背景：高并發下，需要對訪問量、點贊數、排名進行實時計算。
Redis 解決方案：
使用
INCR/HINCRBY做計數器。使用
Sorted Set實現排行榜，支持按分數排序。
應用場景：網站 PV/UV 統計、熱搜榜、積分排名、點贊排行榜。
示例代碼：
import redisr = redis.Redis(host='localhost', port=6379, db=0)# 統計 PV
r.incr("page_view")
print("頁面訪問量:", r.get("page_view").decode())# 排行榜
leaderboard = "game_rank"
r.zincrby(leaderboard, 10, "user1")
r.zincrby(leaderboard, 20, "user2")print("排行榜:", r.zrevrange(leaderboard, 0, -1, withscores=True))
總結
在互聯網高并發場景下，Redis 可以解決以下核心問題：
緩存熱點數據 → 緩解數據庫壓力，提升響應速度。
限流 → 防止突發流量沖擊系統。
分布式隊列 → 實現削峰填谷，支撐異步處理。
分布式鎖 → 保證數據一致性，防止并發沖突。
實時統計與排行榜 → 提供快速的聚合計算能力。
Redis 不僅僅是一個緩存，更是一個 高并發場景下的基礎設施。合理利用其數據結構與特性，可以極大提升系統的可擴展性與穩定性。
 窗口函數介紹)
