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-CN/encode/hash
- JSON格式化：/zh-CN/json/format
- Base64编解码：/zh-CN/encode/base64
- JWT解码：/zh-CN/encode/jwt-decode
本站提供浏览器本地工具，免注册即可试用 →
#Redis#分布式锁#Redlock#并发控制#分布式系统#2026#Redisson