Redis分散式鎖實戰：從Redlock到生產級鎖服務的5種實現模式
数据库
分散式鎖：你踩過的坑比寫的程式碼還多
庫存扣減超賣、定時任務重複執行、冪等介面被併發擊穿——這些生產事故的根因都是分散式併發控制失效。你用SET NX EX加鎖，結果鎖超時了業務還沒完成；你換成Redisson，發現看門狗續期在GC停頓時失效；你上Redlock，又被Martin Kleppmann那篇論文嚇退了。2026年，Redis分散式鎖依然是分散式系統中最容易踩坑的元件之一。
本文將從5種實現模式出發，帶你完成基礎鎖→可重入鎖→Redlock→鎖續期→生產級鎖服務的全鏈路實戰，每一步都有完整程式碼和避坑指南。
Redis分散式鎖核心概念
| 概念 | 說明 |
|---|---|
| SET NX EX | Redis原生命令，NX表示不存在才設定，EX設定過期秒數 |
| 可重入鎖 | 同一執行緒/協程可多次獲取同一把鎖，需計數器配合 |
| 看門狗（Watchdog） | 背景定時續期執行緒，防止業務未完成鎖就過期 |
| Redlock演算法 | 多節點分散式鎖，在N/2+1個節點加鎖成功才算獲取 |
| Lua指令碼 | 原子操作保證，加鎖/解鎖的check-and-set必須原子 |
| 公平鎖 | 按請求順序獲取鎖，避免飢餓問題 |
| 讀寫鎖 | 讀鎖共享、寫鎖排他，提升讀多寫少場景併發度 |
| 號誌 | 允許N個持有者同時獲取，用於限流/資源池 |
問題分析：分散式鎖的5大挑戰
- 鎖超時與業務時長不匹配：鎖10秒過期，但業務執行了15秒，導致鎖提前釋放、併發擊穿
- GC停頓導致看門狗失效：JVM/Go runtime的STW暫停會讓續期執行緒無法按時執行
- Redlock的時鐘漂移問題：多節點時鐘不同步，可能導致鎖安全性被破壞
- 解鎖誤刪他人鎖：A的鎖超時後B獲取鎖，A執行完解鎖把B的鎖刪了
- 網路分割下的腦裂：客戶端與Redis節點網路斷開，鎖狀態不一致
分步實操：5種Redis分散式鎖實現
模式1：基礎SET NX EX鎖
import redis
import uuid
import time
class RedisBasicLock:
def __init__(self, redis_client: redis.Redis, lock_name: str, timeout: int = 10):
self.redis = redis_client
self.lock_name = f"lock:{lock_name}"
self.timeout = timeout
self.identifier = str(uuid.uuid4())
def acquire(self) -> bool:
result = self.redis.set(
self.lock_name,
self.identifier,
nx=True,
ex=self.timeout
)
return result is not None
def release(self) -> bool:
lua_script = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = self.redis.eval(lua_script, 1, self.lock_name, self.identifier)
return result == 1
def __enter__(self):
if not self.acquire():
raise RuntimeError(f"Failed to acquire lock: {self.lock_name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
模式2：可重入鎖（Reentrant Lock）
import redis
import uuid
import threading
class RedisReentrantLock:
def __init__(self, redis_client: redis.Redis, lock_name: str, timeout: int = 30):
self.redis = redis_client
self.lock_name = f"reentrant_lock:{lock_name}"
self.timeout = timeout
self.identifier = str(uuid.uuid4())
self._local = threading.local()
def acquire(self) -> bool:
count = getattr(self._local, 'count', 0)
if count > 0:
self._local.count = count + 1
return True
lua_acquire = """
if redis.call("exists", KEYS[1]) == 0 then
redis.call("hset", KEYS[1], ARGV[1], 1)
redis.call("expire", KEYS[1], ARGV[2])
return 1
elseif redis.call("hexists", KEYS[1], ARGV[1]) == 1 then
redis.call("hincrby", KEYS[1], ARGV[1], 1)
redis.call("expire", KEYS[1], ARGV[2])
return 1
else
return 0
end
"""
result = self.redis.eval(lua_acquire, 1, self.lock_name, self.identifier, str(self.timeout))
if result == 1:
self._local.count = 1
return True
return False
def release(self) -> bool:
count = getattr(self._local, 'count', 0)
if count == 0:
return False
if count > 1:
self._local.count = count - 1
lua_decr = """
if redis.call("hexists", KEYS[1], ARGV[1]) == 1 then
redis.call("hincrby", KEYS[1], ARGV[1], -1)
return 1
else
return 0
end
"""
self.redis.eval(lua_decr, 1, self.lock_name, self.identifier)
return True
lua_release = """
if redis.call("hexists", KEYS[1], ARGV[1]) == 0 then
return 0
elseif redis.call("hincrby", KEYS[1], ARGV[1], -1) > 0 then
redis.call("expire", KEYS[1], ARGV[2])
return 1
else
return redis.call("del", KEYS[1])
end
"""
result = self.redis.eval(lua_release, 1, self.lock_name, self.identifier, str(self.timeout))
self._local.count = 0
return result in (1,)
def __enter__(self):
if not self.acquire():
raise RuntimeError(f"Failed to acquire reentrant lock: {self.lock_name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
模式3：看門狗自動續期鎖
import redis
import uuid
import threading
import time
import logging
logger = logging.getLogger(__name__)
class RedisWatchdogLock:
def __init__(self, redis_client: redis.Redis, lock_name: str, timeout: int = 30, renewal_interval: int = 10):
self.redis = redis_client
self.lock_name = f"watchdog_lock:{lock_name}"
self.timeout = timeout
self.renewal_interval = renewal_interval
self.identifier = str(uuid.uuid4())
self._watchdog_thread = None
self._stop_event = threading.Event()
def acquire(self, blocking: bool = True, wait_timeout: float = 30.0) -> bool:
deadline = time.time() + wait_timeout
while True:
result = self.redis.set(self.lock_name, self.identifier, nx=True, ex=self.timeout)
if result is not None:
self._start_watchdog()
return True
if not blocking:
return False
if time.time() >= deadline:
return False
time.sleep(0.1)
def _start_watchdog(self):
self._stop_event.clear()
self._watchdog_thread = threading.Thread(target=self._watchdog_loop, daemon=True)
self._watchdog_thread.start()
def _watchdog_loop(self):
while not self._stop_event.is_set():
self._stop_event.wait(self.renewal_interval)
if self._stop_event.is_set():
break
try:
lua_renew = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("expire", KEYS[1], ARGV[2])
else
return 0
end
"""
result = self.redis.eval(lua_renew, 1, self.lock_name, self.identifier, str(self.timeout))
if result != 1:
logger.warning("Watchdog renewal failed for lock %s", self.lock_name)
break
except Exception as e:
logger.error("Watchdog error: %s", e)
break
def release(self) -> bool:
self._stop_event.set()
if self._watchdog_thread and self._watchdog_thread.is_alive():
self._watchdog_thread.join(timeout=2.0)
lua_release = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = self.redis.eval(lua_release, 1, self.lock_name, self.identifier)
return result == 1
def __enter__(self):
if not self.acquire():
raise RuntimeError(f"Failed to acquire watchdog lock: {self.lock_name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
模式4：Redlock多節點鎖
import redis
import uuid
import time
import logging
from typing import List
logger = logging.getLogger(__name__)
class Redlock:
def __init__(self, redis_clients: List[redis.Redis], lock_name: str, timeout: int = 10, retry_count: int = 3, retry_delay: float = 0.2):
self.redis_clients = redis_clients
self.quorum = len(redis_clients) // 2 + 1
self.lock_name = f"redlock:{lock_name}"
self.timeout = timeout
self.retry_count = retry_count
self.retry_delay = retry_delay
self.identifier = str(uuid.uuid4())
def acquire(self) -> bool:
for attempt in range(self.retry_count):
acquired_count = 0
start_time = time.monotonic()
for client in self.redis_clients:
try:
result = client.set(self.lock_name, self.identifier, nx=True, ex=self.timeout)
if result is not None:
acquired_count += 1
except Exception as e:
logger.warning("Redlock acquire error on node: %s", e)
elapsed = time.monotonic() - start_time
validity_time = self.timeout - elapsed
if acquired_count >= self.quorum and validity_time > 0:
return True
self._release_all_nodes()
if attempt < self.retry_count - 1:
jitter = (attempt * 0.01)
time.sleep(self.retry_delay + jitter)
return False
def _release_all_nodes(self):
lua_release = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
for client in self.redis_clients:
try:
client.eval(lua_release, 1, self.lock_name, self.identifier)
except Exception as e:
logger.warning("Redlock release error on node: %s", e)
def release(self) -> bool:
self._release_all_nodes()
return True
def __enter__(self):
if not self.acquire():
raise RuntimeError(f"Failed to acquire Redlock: {self.lock_name}")
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.release()
模式5：生產級鎖服務（Go實現）
package distlock
import (
"context"
"crypto/rand"
"encoding/hex"
"fmt"
"log"
"sync"
"time"
"github.com/redis/go-redis/v9"
)
type LockService struct {
client *redis.Client
watchdogCancel map[string]context.CancelFunc
mu sync.Mutex
}
func NewLockService(client *redis.Client) *LockService {
return &LockService{
client: client,
watchdogCancel: make(map[string]context.CancelFunc),
}
}
type LockOptions struct {
Timeout time.Duration
RenewalInterval time.Duration
RetryCount int
RetryDelay time.Duration
}
func DefaultLockOptions() LockOptions {
return LockOptions{
Timeout: 30 * time.Second,
RenewalInterval: 10 * time.Second,
RetryCount: 3,
RetryDelay: 200 * time.Millisecond,
}
}
func generateIdentifier() string {
b := make([]byte, 16)
rand.Read(b)
return hex.EncodeToString(b)
}
var acquireScript = redis.NewScript(`
if redis.call("exists", KEYS[1]) == 0 then
redis.call("hset", KEYS[1], "identifier", ARGV[1], "count", 1)
redis.call("expire", KEYS[1], ARGV[2])
return 1
elseif redis.call("hget", KEYS[1], "identifier") == ARGV[1] then
redis.call("hincrby", KEYS[1], "count", 1)
redis.call("expire", KEYS[1], ARGV[2])
return 1
else
return 0
end
`)
var releaseScript = redis.NewScript(`
if redis.call("hget", KEYS[1], "identifier") ~= ARGV[1] then
return 0
end
local count = redis.call("hincrby", KEYS[1], "count", -1)
if count > 0 then
redis.call("expire", KEYS[1], ARGV[2])
return 1
end
return redis.call("del", KEYS[1])
`)
var renewScript = redis.NewScript(`
if redis.call("hget", KEYS[1], "identifier") == ARGV[1] then
return redis.call("expire", KEYS[1], ARGV[2])
else
return 0
end
`)
func (ls *LockService) Acquire(ctx context.Context, lockName string, opts LockOptions) (string, error) {
identifier := generateIdentifier()
key := fmt.Sprintf("lock_service:%s", lockName)
for i := 0; i < opts.RetryCount; i++ {
result, err := acquireScript.Run(ctx, ls.client, []string{key}, identifier, int(opts.Timeout.Seconds())).Int()
if err != nil {
return "", fmt.Errorf("acquire script error: %w", err)
}
if result == 1 {
ls.startWatchdog(ctx, key, identifier, opts)
return identifier, nil
}
select {
case <-ctx.Done():
return "", ctx.Err()
case <-time.After(opts.RetryDelay):
}
}
return "", fmt.Errorf("failed to acquire lock after %d retries", opts.RetryCount)
}
func (ls *LockService) startWatchdog(ctx context.Context, key, identifier string, opts LockOptions) {
wdCtx, cancel := context.WithCancel(context.Background())
ls.mu.Lock()
ls.watchdogCancel[key+":"+identifier] = cancel
ls.mu.Unlock()
go func() {
defer cancel()
ticker := time.NewTicker(opts.RenewalInterval)
defer ticker.Stop()
for {
select {
case <-wdCtx.Done():
return
case <-ticker.C:
result, err := renewScript.Run(wdCtx, ls.client, []string{key}, identifier, int(opts.Timeout.Seconds())).Int()
if err != nil || result != 1 {
log.Printf("Watchdog renewal failed for key=%s identifier=%s: result=%d err=%v", key, identifier, result, err)
return
}
}
}
}()
}
func (ls *LockService) Release(ctx context.Context, lockName, identifier string, opts LockOptions) error {
key := fmt.Sprintf("lock_service:%s", lockName)
ls.mu.Lock()
if cancel, ok := ls.watchdogCancel[key+":"+identifier]; ok {
cancel()
delete(ls.watchdogCancel, key+":"+identifier)
}
ls.mu.Unlock()
result, err := releaseScript.Run(ctx, ls.client, []string{key}, identifier, int(opts.Timeout.Seconds())).Int()
if err != nil {
return fmt.Errorf("release script error: %w", err)
}
if result == 0 {
return fmt.Errorf("lock not owned by identifier %s", identifier)
}
return nil
}
避坑指南
坑1：解鎖不驗證owner直接DEL
# ❌ 錯誤：直接刪除，可能刪了別人的鎖
redis_client.delete("lock:order:123")
# ✅ 正確：Lua指令碼原子check-and-delete
lua = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
redis_client.eval(lua, 1, "lock:order:123", my_identifier)
坑2：鎖超時時間設太短
# ❌ 錯誤：3秒超時，資料庫慢查詢就超了
redis_client.set("lock:order", identifier, nx=True, ex=3)
# ✅ 正確：看門狗續期 + 合理初始超時
lock = RedisWatchdogLock(redis_client, "order", timeout=30, renewal_interval=10)
with lock:
process_order()
坑3：可重入鎖不計數
# ❌ 錯誤：每次SET NX，巢狀呼叫獲取不到鎖
def outer():
with basic_lock:
inner()
def inner():
with basic_lock: # 死鎖！自己持有的鎖自己獲取不到
pass
# ✅ 正確：使用可重入鎖
def outer():
with reentrant_lock:
inner()
def inner():
with reentrant_lock: # 計數+1，可正常獲取
pass
坑4：Redlock忽略時鐘漂移
# ❌ 錯誤：不校驗鎖的有效剩餘時間
acquired_count = 0
for client in redis_clients:
result = client.set(lock_name, identifier, nx=True, ex=timeout)
if result:
acquired_count += 1
if acquired_count >= quorum:
return True # 可能鎖已經快過期了！
# ✅ 正確：校驗有效時間
start = time.monotonic()
# ... 加鎖邏輯 ...
elapsed = time.monotonic() - start
validity = timeout - elapsed
if acquired_count >= quorum and validity > 0:
return True
坑5：看門狗續期間隔等於鎖超時
# ❌ 錯誤：續期間隔30秒 = 鎖超時30秒，GC停頓就續不上
lock = RedisWatchdogLock(client, "order", timeout=30, renewal_interval=30)
# ✅ 正確：續期間隔 = 鎖超時 / 3，留足餘量
lock = RedisWatchdogLock(client, "order", timeout=30, renewal_interval=10)
報錯排查
| 序號 | 報錯訊息 | 原因 | 解決方法 |
|---|---|---|---|
| 1 | UNLOCK_FAILED: lock not owned |
解鎖時identifier不匹配 | 確保加鎖和解鎖使用同一個identifier |
| 2 | LOCK_TIMEOUT: acquire failed after retries |
鎖被長時間持有或競爭激烈 | 增加重試次數，檢查是否有死鎖 |
| 3 | WATCHDOG_RENEWAL_FAILED |
看門狗續期失敗，鎖已被刪除或過期 | 檢查網路連線，確認鎖超時設定合理 |
| 4 | RedisConnectionError |
Redis連線斷開 | 設定連線池重試，使用Sentinel/Cluster |
| 5 | LuaScriptError: wrong number of arguments |
Lua指令碼引數不匹配 | 檢查KEYS和ARGV的數量和順序 |
| 6 | Redlock quorum not reached |
多數節點加鎖失敗 | 檢查各節點狀態，增加retry_count |
| 7 | CONCURRENT_MODIFICATION: data inconsistency |
鎖提前釋放導致併發修改 | 使用看門狗續期，增大超時時間 |
| 8 | OOM: Redis out of memory |
鎖key未設定過期，大量堆積 | 確保SET NX EX中EX引數生效 |
| 9 | DEADLOCK: circular wait detected |
多鎖迴圈等待 | 統一加鎖順序，設定全域性超時 |
| 10 | CLOCK_DRIFT: lock validity expired |
Redlock節點時鐘偏差過大 | 設定NTP同步，校驗validity time |
進階最佳化
1. 公平鎖實現
class RedisFairLock:
def __init__(self, redis_client: redis.Redis, lock_name: str, timeout: int = 30):
self.redis = redis_client
self.lock_name = f"fair_lock:{lock_name}"
self.queue_name = f"fair_lock_queue:{lock_name}"
self.timeout = timeout
self.identifier = str(uuid.uuid4())
def acquire(self, wait_timeout: float = 30.0) -> bool:
timestamp = time.time()
self.redis.zadd(self.queue_name, {self.identifier: timestamp})
self.redis.expire(self.queue_name, wait_timeout * 2)
deadline = time.time() + wait_timeout
while time.time() < deadline:
lua = """
local first = redis.call("zrange", KEYS[2], 0, 0)
if first[1] == ARGV[1] then
local result = redis.call("set", KEYS[1], ARGV[1], "nx", "ex", ARGV[2])
if result then
redis.call("zrem", KEYS[2], ARGV[1])
return 1
end
end
return 0
"""
result = self.redis.eval(lua, 2, self.lock_name, self.queue_name, self.identifier, str(self.timeout))
if result == 1:
return True
time.sleep(0.05)
self.redis.zrem(self.queue_name, self.identifier)
return False
def release(self) -> bool:
lua = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = self.redis.eval(lua, 1, self.lock_name, self.identifier)
return result == 1
2. 讀寫鎖實現
class RedisReadWriteLock:
def __init__(self, redis_client: redis.Redis, lock_name: str, timeout: int = 30):
self.redis = redis_client
self.read_lock_name = f"rw_lock:{lock_name}:read"
self.write_lock_name = f"rw_lock:{lock_name}:write"
self.timeout = timeout
self.identifier = str(uuid.uuid4())
def acquire_read(self) -> bool:
lua = """
if redis.call("exists", KEYS[2]) == 1 then
return 0
end
redis.call("hincrby", KEYS[1], "readers", 1)
redis.call("expire", KEYS[1], ARGV[2])
return 1
"""
result = self.redis.eval(lua, 2, self.read_lock_name, self.write_lock_name, self.identifier, str(self.timeout))
return result == 1
def release_read(self) -> bool:
lua = """
local count = redis.call("hincrby", KEYS[1], "readers", -1)
if count <= 0 then
redis.call("del", KEYS[1])
end
return 1
"""
self.redis.eval(lua, 1, self.read_lock_name, self.identifier)
return True
def acquire_write(self) -> bool:
result = self.redis.set(self.write_lock_name, self.identifier, nx=True, ex=self.timeout)
if result is None:
return False
lua = """
if redis.call("exists", KEYS[1]) == 1 and redis.call("hget", KEYS[1], "readers") ~= "0" then
redis.call("del", KEYS[2])
return 0
end
return 1
"""
check = self.redis.eval(lua, 2, self.read_lock_name, self.write_lock_name, self.identifier)
return check == 1
def release_write(self) -> bool:
lua = """
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
"""
result = self.redis.eval(lua, 1, self.write_lock_name, self.identifier)
return result == 1
3. 鎖監控指標採集
package distlock
import (
"context"
"fmt"
"time"
"github.com/redis/go-redis/v9"
)
type LockMetrics struct {
LockName string
CurrentHolder string
RemainTTL time.Duration
AcquireCount int64
WaitQueueLen int64
}
func CollectLockMetrics(ctx context.Context, client *redis.Client, lockName string) (*LockMetrics, error) {
key := fmt.Sprintf("lock_service:%s", lockName)
ttl, err := client.TTL(ctx, key).Result()
if err != nil {
return nil, err
}
identifier, _ := client.HGet(ctx, key, "identifier").Result()
count, _ := client.HGet(ctx, key, "count").Int64()
queueKey := fmt.Sprintf("fair_lock_queue:%s", lockName)
queueLen, _ := client.ZCard(ctx, queueKey).Result()
return &LockMetrics{
LockName: lockName,
CurrentHolder: identifier,
RemainTTL: ttl,
AcquireCount: count,
WaitQueueLen: queueLen,
}, nil
}
對比分析
| 維度 | SET NX EX | 可重入鎖 | 看門狗鎖 | Redlock | 生產級鎖服務 |
|---|---|---|---|---|---|
| 實現複雜度 | ⭐低 | ⭐⭐中 | ⭐⭐⭐高 | ⭐⭐⭐高 | ⭐⭐⭐⭐極高 |
| 原子性 | ⚠️需Lua | ✅Lua保證 | ✅Lua保證 | ✅多節點 | ✅Lua保證 |
| 可重入 | ❌ | ✅ | ✅ | ❌ | ✅ |
| 自動續期 | ❌ | ❌ | ✅ | ❌ | ✅ |
| 多節點容錯 | ❌ | ❌ | ❌ | ✅ | 可選 |
| 防誤刪 | ⚠️需Lua | ✅ | ✅ | ✅ | ✅ |
| GC容忍度 | ⭐高 | ⭐高 | ⚠️中 | ⭐高 | ⚠️中 |
| 生產推薦 | 原型驗證 | 常規業務 | 長事務 | 高可用 | 核心鏈路 |
總結：Redis分散式鎖不是「一個命令」的事，而是「一套體系」的工程。從SET NX EX到生產級鎖服務，核心原則只有三條：原子操作用Lua、鎖超時要續期、解鎖先驗owner。Redlock在多數場景下是過度設計——單節點+Sentinel高可用足夠應對99%的業務場景。只有在「鎖失效會導致嚴重資料不一致」的核心鏈路上，才值得投入Redlock多節點方案。
線上工具推薦
- Hash計算：/zh-TW/encode/hash
- JSON格式化：/zh-TW/json/format
- Base64編解碼：/zh-TW/encode/base64
- JWT解碼：/zh-TW/encode/jwt-decode
本站提供瀏覽器本地工具，免註冊即可試用 →
#Redis#分布式锁#Redlock#并发控制#分布式系统#2026#Redisson