Python分布式锁实现：构建高并发环境下的资源保护机制

Python分布式锁实现：构建高并发环境下的资源保护机制

引言

在分布式系统中，多个进程或服务同时访问共享资源时，需要一种机制来保证数据的一致性和操作的原子性。分布式锁就是解决这类问题的核心技术。作为一名从Python转向Rust的后端开发者，我在实践中总结了多种分布式锁的实现方式。本文将深入探讨Python中分布式锁的设计与实现，帮助你掌握这一关键技术。

一、分布式锁核心概念

1.1 什么是分布式锁

分布式锁是一种在分布式系统中协调多个节点对共享资源访问的机制，确保同一时刻只有一个节点可以访问特定资源。

1.2 分布式锁的特性

一个可靠的分布式锁需要满足以下特性：

• 互斥性：同一时刻只有一个客户端持有锁
• 可重入性：同一个客户端可以多次获取同一把锁
• 超时释放：锁持有超过一定时间自动释放，防止死锁
• 高可用性：锁服务本身需要高可用
• 公平性：请求锁的顺序应该得到尊重

1.3 常见分布式锁实现方案对比

二、基于Redis的分布式锁实现

2.1 基础实现

import redis
import uuid
import time

class RedisLock:
def __init__(self, redis_client: redis.Redis, lock_key: str, timeout: int = 30):
self.redis_client = redis_client
self.lock_key = lock_key
self.timeout = timeout
self.lock_value = None

def acquire(self, blocking: bool = True, wait_timeout: float = None) -> bool:
"""
获取分布式锁
:param blocking: 是否阻塞等待
:param wait_timeout: 等待超时时间
:return: 是否成功获取锁
"""
self.lock_value = str(uuid.uuid4())
end_time = time.time() + (wait_timeout or float('inf'))

while time.time() < end_time:
result = self.redis_client.set(
self.lock_key,
self.lock_value,
ex=self.timeout,
nx=True
)

if result:
return True

if not blocking:
return False

time.sleep(0.1)

return False

def release(self) -> bool:
"""
释放分布式锁
:return: 是否成功释放
"""
script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
"""
result = self.redis_client.eval(script, 1, self.lock_key, self.lock_value)
return result == 1

def __enter__(self):
self.acquire()
return self

def __exit__(self, exc_type, exc_val, exc_tb):
self.release()

2.2 可重入锁实现

class RedisReentrantLock:
def __init__(self, redis_client: redis.Redis, lock_key: str, timeout: int = 30):
self.redis_client = redis_client
self.lock_key = lock_key
self.timeout = timeout
self.lock_value = None
self.reentry_count = 0

def acquire(self, blocking: bool = True, wait_timeout: float = None) -> bool:
if self.reentry_count > 0:
self.reentry_count += 1
return True

self.lock_value = str(uuid.uuid4())
end_time = time.time() + (wait_timeout or float('inf'))

while time.time() < end_time:
result = self.redis_client.set(
self.lock_key,
self.lock_value,
ex=self.timeout,
nx=True
)

if result:
self.reentry_count = 1
return True

if not blocking:
return False

time.sleep(0.1)

return False

def release(self) -> bool:
if self.reentry_count == 0:
return False

self.reentry_count -= 1

if self.reentry_count == 0:
script = """
if redis.call('get', KEYS[1]) == ARGV[1] then
return redis.call('del', KEYS[1])
else
return 0
end
"""
result = self.redis_client.eval(script, 1, self.lock_key, self.lock_value)
return result == 1

return True

三、基于ZooKeeper的分布式锁实现

3.1 基础实现

from kazoo.client import KazooClient
from kazoo.exceptions import NodeExistsError
import time

class ZookeeperLock:
def __init__(self, hosts: str, lock_path: str):
self.hosts = hosts
self.lock_path = lock_path
self.client = KazooClient(hosts=hosts)
self.client.start()
self.lock_node = None

def acquire(self, blocking: bool = True, wait_timeout: float = None) -> bool:
"""
获取分布式锁
"""
try:
self.lock_node = self.client.create(
f"{self.lock_path}/lock-",
ephemeral=True,
sequence=True
)
except NodeExistsError:
return False

nodes = sorted(self.client.get_children(self.lock_path))
lock_name = self.lock_node.split('/')[-1]

if nodes[0] == lock_name:
return True

if not blocking:
self.client.delete(self.lock_node)
return False

predecessor = nodes[nodes.index(lock_name) – 1]
predecessor_path = f"{self.lock_path}/{predecessor}"

event = self.client.WatchedEvent(type='DELETED', state='CONNECTED', path=predecessor_path)

def watch_callback(event):
nonlocal event
event = event

self.client.get(predecessor_path, watch=watch_callback)

if wait_timeout:
start_time = time.time()
while time.time() – start_time < wait_timeout:
if event.type == 'DELETED':
return True
time.sleep(0.1)
self.client.delete(self.lock_node)
return False

while event.type != 'DELETED':
time.sleep(0.1)

return True

def release(self) -> bool:
if self.lock_node:
self.client.delete(self.lock_node)
self.lock_node = None
return True
return False

def close(self):
self.client.stop()
self.client.close()

四、基于数据库的分布式锁实现

4.1 使用MySQL实现

import time
import uuid
from sqlalchemy import create_engine, Column, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetime

Base = declarative_base()

class DistributedLock(Base):
__tablename__ = 'distributed_locks'

lock_key = Column(String(255), primary_key=True)
lock_value = Column(String(255))
expire_time = Column(DateTime)

class DatabaseLock:
def __init__(self, db_url: str):
self.engine = create_engine(db_url)
Session = sessionmaker(bind=self.engine)
self.session = Session()
self.lock_value = None

def acquire(self, lock_key: str, timeout: int = 30) -> bool:
self.lock_value = str(uuid.uuid4())
expire_time = datetime.now() + timedelta(seconds=timeout)

try:
lock = DistributedLock(
lock_key=lock_key,
lock_value=self.lock_value,
expire_time=expire_time
)
self.session.add(lock)
self.session.commit()
return True
except Exception:
self.session.rollback()

lock = self.session.query(DistributedLock).filter_by(lock_key=lock_key).first()
if lock and lock.expire_time < datetime.now():
lock.lock_value = self.lock_value
lock.expire_time = expire_time
self.session.commit()
return True

return False

def release(self, lock_key: str) -> bool:
if not self.lock_value:
return False

lock = self.session.query(DistributedLock).filter_by(
lock_key=lock_key,
lock_value=self.lock_value
).first()

if lock:
self.session.delete(lock)
self.session.commit()
return True

return False

五、分布式锁设计模式

5.1 红锁算法实现

红锁（Redlock）是Redis官方提出的分布式锁算法，通过多个Redis实例来提高锁的可靠性。

class Redlock:
def __init__(self, redis_clients: list, quorum: int = None):
self.redis_clients = redis_clients
self.quorum = quorum or (len(redis_clients) // 2 + 1)
self.lock_value = str(uuid.uuid4())

def acquire(self, lock_key: str, timeout: int = 30) -> bool:
acquired_count = 0
start_time = time.time()

for client in self.redis_clients:
try:
result = client.set(
lock_key,
self.lock_value,
ex=timeout,
nx=True
)
if result:
acquired_count += 1
except Exception:
pass

elapsed = time.time() – start_time

if acquired_count >= self.quorum and elapsed < timeout:
return True

self.release(lock_key)
return False

def release(self, lock_key: str):
script = """
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(script, 1, lock_key, self.lock_value)
except Exception:
pass

5.2 乐观锁模式

乐观锁假设冲突很少发生，通过版本号或时间戳来检测冲突。

class OptimisticLock:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def get_with_version(self, key: str):
value = self.redis_client.get(key)
if value:
data = json.loads(value)
return data.get('value'), data.get('version', 0)
return None, 0

def update_with_version(self, key: str, value, expected_version: int) -> bool:
current_value = self.redis_client.get(key)

if current_value:
data = json.loads(current_value)
if data.get('version', 0) != expected_version:
return False

new_data = {
'value': value,
'version': expected_version + 1
}
self.redis_client.set(key, json.dumps(new_data))
return True

return False

六、实际业务场景应用

6.1 秒杀系统中的分布式锁

class SeckillService:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client
self.lock = RedisLock(redis_client, 'seckill_lock')

def seckill(self, product_id: str, user_id: str) -> bool:
with self.lock:
stock_key = f"product:{product_id}:stock"
user_key = f"user:{user_id}:purchased"

if self.redis_client.sismember(user_key, product_id):
print("User already purchased this product")
return False

stock = self.redis_client.get(stock_key)
if not stock or int(stock) <= 0:
print("Product sold out")
return False

self.redis_client.decr(stock_key)
self.redis_client.sadd(user_key, product_id)
print("Purchase successful")
return True

6.2 分布式任务调度

class TaskScheduler:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def schedule_task(self, task_id: str, task_data: dict):
lock = RedisLock(self.redis_client, f"task:{task_id}:lock")

if lock.acquire(blocking=False):
try:
print(f"Executing task: {task_id}")
self._execute_task(task_data)
finally:
lock.release()
else:
print(f"Task {task_id} is already running")

def _execute_task(self, task_data: dict):
time.sleep(5)
print(f"Task completed with data: {task_data}")

七、性能优化策略

7.1 锁粒度优化

class FineGrainedLock:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def acquire_resource_lock(self, resource_type: str, resource_id: str) -> RedisLock:
lock_key = f"{resource_type}:{resource_id}:lock"
return RedisLock(self.redis_client, lock_key)

def acquire_batch_lock(self, resource_type: str, resource_ids: list) -> bool:
locks = []

try:
for resource_id in sorted(resource_ids):
lock = self.acquire_resource_lock(resource_type, resource_id)
if not lock.acquire(blocking=False):
for l in locks:
l.release()
return False
locks.append(lock)

self._locks = locks
return True
except Exception:
for l in locks:
l.release()
return False

def release_batch_lock(self):
if hasattr(self, '_locks'):
for lock in self._locks:
lock.release()
delattr(self, '_locks')

7.2 锁超时自动续期

import threading

class AutoRenewLock:
def __init__(self, redis_client: redis.Redis, lock_key: str, timeout: int = 30):
self.lock = RedisLock(redis_client, lock_key, timeout)
self.timeout = timeout
self.renew_interval = timeout // 3
self.renew_thread = None
self.running = False

def _renew_loop(self):
while self.running:
self.lock.redis_client.expire(self.lock.lock_key, self.timeout)
time.sleep(self.renew_interval)

def acquire(self) -> bool:
if self.lock.acquire():
self.running = True
self.renew_thread = threading.Thread(target=self._renew_loop, daemon=True)
self.renew_thread.start()
return True
return False

def release(self):
self.running = False
if self.renew_thread:
self.renew_thread.join()
self.lock.release()

八、监控与运维

8.1 锁使用统计

class LockMonitor:
def __init__(self, redis_client: redis.Redis):
self.redis_client = redis_client

def get_lock_stats(self, lock_key_pattern: str = '*') -> dict:
stats = {}

for key in self.redis_client.keys(f"{lock_key_pattern}"):
key_str = key.decode('utf-8')
ttl = self.redis_client.ttl(key)
stats[key_str] = {
'ttl': ttl,
'exists': True
}

return stats

def detect_dead_locks(self, max_age_seconds: int = 300) -> list:
dead_locks = []
stats = self.get_lock_stats()

for lock_key, info in stats.items():
if info['ttl'] == -1:
dead_locks.append(lock_key)

return dead_locks

def cleanup_expired_locks(self):
stats = self.get_lock_stats()

for lock_key, info in stats.items():
if info['ttl'] == -2:
self.redis_client.delete(lock_key)

总结

分布式锁是构建高并发分布式系统的核心组件。通过本文的学习，你应该掌握了以下核心要点：

作为从Python转向Rust的后端开发者，掌握分布式锁的设计与实现对于构建高可用系统至关重要。后续文章将探讨如何在Rust中实现高性能分布式锁。