Python中的分布式锁机制：保证资源访问的互斥性与一致性 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Python 分布式锁机制：保证资源访问的互斥性与一致性

大家好，今天我们来聊聊Python中的分布式锁机制。在单体应用中，我们通常使用线程锁、进程锁等机制来保证多线程或多进程环境下对共享资源的互斥访问。然而，在分布式系统中，多个独立的节点都需要访问同一份共享资源，这时传统的锁机制就无法满足需求了。分布式锁应运而生，它提供了一种跨多个节点的互斥访问机制，保证数据的一致性和正确性。

什么是分布式锁

简单来说，分布式锁就是一个在多个节点之间共享的锁。它的主要目标是确保在任何时刻，只有一个客户端可以持有锁，从而访问或修改共享资源。这就像现实世界中的物理锁一样，只有拿到钥匙的人才能打开门。

分布式锁需要满足的特性：

互斥性 (Mutual Exclusion): 在任何时刻，只有一个客户端可以持有锁。
容错性 (Fault Tolerance): 即使持有锁的节点发生故障，锁也应该能够自动释放，避免死锁。
可靠性 (Reliability): 锁机制本身需要具有高可用性，避免单点故障。
可重入性 (Reentrancy): 同一个客户端可以多次获取同一个锁。 (可选，但通常需要支持)
高性能 (Performance): 锁的获取和释放操作应该尽可能高效。

分布式锁的实现方式

实现分布式锁的方式有很多种，常见的包括：

基于数据库的锁
基于Redis的锁
基于ZooKeeper的锁
基于etcd的锁

我们接下来分别介绍这些实现方式，并提供相应的Python代码示例。

1. 基于数据库的锁

利用数据库的唯一索引或乐观锁机制可以实现简单的分布式锁。

a. 唯一索引实现：

创建一个锁表，包含锁的名称和持有者信息。
尝试插入一条记录，如果插入成功，则获得锁；如果插入失败（违反唯一索引约束），则表示锁已被其他客户端持有。
释放锁时，删除对应的记录。

Python代码示例：

import pymysql
import uuid
import time

class DatabaseLock:
    def __init__(self, host, user, password, database, lock_name):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.lock_name = lock_name
        self.connection = None

    def connect(self):
        self.connection = pymysql.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database,
            cursorclass=pymysql.cursors.DictCursor
        )

    def acquire_lock(self, timeout=10):
        self.connect()
        cursor = self.connection.cursor()
        client_id = str(uuid.uuid4())
        start_time = time.time()

        while True:
            try:
                cursor.execute(
                    "INSERT INTO locks (lock_name, client_id) VALUES (%s, %s)",
                    (self.lock_name, client_id)
                )
                self.connection.commit()
                return True  # 成功获取锁
            except pymysql.err.IntegrityError:
                # 锁已被占用
                self.connection.rollback()
                if time.time() - start_time > timeout:
                    return False  # 获取锁超时
                time.sleep(0.1)
            except Exception as e:
                print(f"Error acquiring lock: {e}")
                self.connection.rollback()
                return False
            finally:
                cursor.close()

    def release_lock(self):
        if self.connection is None:
            return  # 锁未被持有

        cursor = self.connection.cursor()
        try:
            cursor.execute(
                "DELETE FROM locks WHERE lock_name = %s",
                (self.lock_name,)
            )
            self.connection.commit()
        except Exception as e:
            print(f"Error releasing lock: {e}")
            self.connection.rollback()
        finally:
            cursor.close()
            self.connection.close()
            self.connection = None

# 使用示例
if __name__ == '__main__':
    lock = DatabaseLock(
        host="localhost",
        user="your_user",
        password="your_password",
        database="your_database",
        lock_name="my_resource_lock"
    )

    if lock.acquire_lock(timeout=5):
        print("Successfully acquired lock!")
        try:
            # 访问共享资源
            print("Processing shared resource...")
            time.sleep(5)  # 模拟资源处理
        finally:
            lock.release_lock()
            print("Released lock.")
    else:
        print("Failed to acquire lock.")

数据库表结构：

CREATE TABLE locks (
    lock_name VARCHAR(255) NOT NULL,
    client_id VARCHAR(255) NOT NULL,
    PRIMARY KEY (lock_name),
    UNIQUE INDEX lock_name (lock_name)
);

b. 乐观锁实现：

在表中增加一个版本号字段。
获取锁时，先查询当前版本号。
更新记录时，比较当前版本号是否与之前查询到的版本号一致。如果一致，则更新成功，获得锁；否则，更新失败，表示锁已被其他客户端持有。

Python代码示例：

import pymysql
import time

class OptimisticLock:
    def __init__(self, host, user, password, database, resource_id):
        self.host = host
        self.user = user
        self.password = password
        self.database = database
        self.resource_id = resource_id
        self.connection = None

    def connect(self):
        self.connection = pymysql.connect(
            host=self.host,
            user=self.user,
            password=self.password,
            database=self.database,
            cursorclass=pymysql.cursors.DictCursor
        )

    def acquire_lock(self, timeout=10):
        self.connect()
        cursor = self.connection.cursor()
        start_time = time.time()

        while True:
            try:
                # 1. 获取当前版本号
                cursor.execute("SELECT version FROM resources WHERE id = %s", (self.resource_id,))
                result = cursor.fetchone()
                if result is None:
                    print("Resource not found.")
                    return False
                current_version = result['version']

                # 2. 尝试更新版本号，同时执行资源操作
                cursor.execute(
                    "UPDATE resources SET data = 'locked', version = version + 1 WHERE id = %s AND version = %s",
                    (self.resource_id, current_version)
                )
                rows_affected = cursor.rowcount
                self.connection.commit()

                if rows_affected > 0:
                    # 成功获取锁
                    return True
                else:
                    # 锁已被占用，或资源不存在
                    self.connection.rollback()
                    if time.time() - start_time > timeout:
                        return False  # 获取锁超时
                    time.sleep(0.1)

            except Exception as e:
                print(f"Error acquiring lock: {e}")
                self.connection.rollback()
                return False
            finally:
                cursor.close()

    def release_lock(self):
         if self.connection is None:
            return

         cursor = self.connection.cursor()
         try:
             # 释放锁，将数据恢复到初始状态
             cursor.execute(
                 "UPDATE resources SET data = 'unlocked' WHERE id = %s",
                 (self.resource_id,)
             )
             self.connection.commit()
         except Exception as e:
             print(f"Error releasing lock: {e}")
             self.connection.rollback()
         finally:
             cursor.close()
             self.connection.close()
             self.connection = None

# 使用示例
if __name__ == '__main__':
    lock = OptimisticLock(
        host="localhost",
        user="your_user",
        password="your_password",
        database="your_database",
        resource_id=1  # 假设资源ID为1
    )

    if lock.acquire_lock(timeout=5):
        print("Successfully acquired lock!")
        try:
            # 访问共享资源
            print("Processing shared resource...")
            time.sleep(5)  # 模拟资源处理
        finally:
            lock.release_lock()
            print("Released lock.")
    else:
        print("Failed to acquire lock.")

数据库表结构：

CREATE TABLE resources (
    id INT PRIMARY KEY,
    data VARCHAR(255),
    version INT
);

INSERT INTO resources (id, data, version) VALUES (1, 'unlocked', 0);

优点：

实现简单，易于理解。
不需要引入额外的组件。

缺点：

性能较低，频繁的数据库操作会增加数据库的压力。
存在单点故障的风险。
轮询获取锁会占用数据库连接资源。
唯一索引方式，如果持有锁的客户端崩溃，可能导致死锁，需要额外的机制来解决（例如设置超时时间）。
乐观锁方式，在高并发场景下，冲突概率较高，重试次数会增加。

2. 基于 Redis 的锁

Redis 是一种高性能的内存数据库，常被用作缓存和消息队列。利用 Redis 的原子操作可以实现高效的分布式锁。

实现方式：

使用 SETNX 命令 (SET if Not eXists) 尝试设置一个键值对。如果键不存在，则设置成功，表示获得锁；如果键已存在，则设置失败，表示锁已被其他客户端持有。
为了避免死锁，可以设置一个过期时间，即使持有锁的客户端崩溃，锁也会自动释放。
释放锁时，需要使用原子操作 DEL 删除键。为了防止误删锁，可以验证客户端 ID。

Python代码示例：

import redis
import uuid
import time

class RedisLock:
    def __init__(self, host, port, db, lock_name, lock_timeout=10):
        self.redis = redis.Redis(host=host, port=port, db=db)
        self.lock_name = lock_name
        self.lock_timeout = lock_timeout
        self.client_id = str(uuid.uuid4())

    def acquire_lock(self, timeout=10):
        start_time = time.time()
        while True:
            # 使用 SETNX 设置锁，并设置过期时间
            lock_acquired = self.redis.set(self.lock_name, self.client_id, nx=True, ex=self.lock_timeout)
            if lock_acquired:
                return True  # 成功获取锁

            if time.time() - start_time > timeout:
                return False  # 获取锁超时

            time.sleep(0.01)  # 避免忙等待

    def release_lock(self):
        # 使用 Lua 脚本原子性地删除锁，避免误删
        script = """
        if redis.call("GET", KEYS[1]) == ARGV[1] then
            return redis.call("DEL", KEYS[1])
        else
            return 0
        end
        """
        release_script = self.redis.register_script(script)
        result = release_script(keys=[self.lock_name], args=[self.client_id])
        return result == 1

# 使用示例
if __name__ == '__main__':
    lock = RedisLock(host="localhost", port=6379, db=0, lock_name="my_resource_lock", lock_timeout=10)

    if lock.acquire_lock(timeout=5):
        print("Successfully acquired lock!")
        try:
            # 访问共享资源
            print("Processing shared resource...")
            time.sleep(5)  # 模拟资源处理
        finally:
            if lock.release_lock():
                print("Released lock.")
            else:
                print("Failed to release lock (lock might have expired or been released by another client).")
    else:
        print("Failed to acquire lock.")

Lua脚本解释：

使用Lua脚本的目的是保证删除锁操作的原子性。如果不使用Lua脚本，而是先GET再DEL，则可能出现以下情况：

客户端A GET锁，发现是自己的ClientId
客户端A 准备DEL锁
锁过期，被客户端B 获取
客户端A 执行DEL操作，将客户端B的锁删除，导致其他客户端可以同时访问共享资源。

Lua脚本保证了只有当锁的ClientId与当前客户端的ClientId一致时，才删除锁，避免了误删锁的问题。

Redlock算法:

上述简单的Redis锁实现方式存在一些问题，例如：

单点故障： 如果Redis主节点发生故障，可能会导致锁丢失。

为了解决这些问题，Redis官方提出了Redlock算法。 Redlock算法的核心思想是使用多个独立的Redis节点，客户端需要同时从多个节点获取锁，只有当超过半数的节点成功获取锁时，才认为获取锁成功。

虽然Redlock算法可以提高锁的可靠性，但也增加了实现的复杂性，并且存在一些争议。具体可以参考Redis官方文档和相关论文。这里不再提供Redlock算法的代码示例，因为它的实现比较复杂，并且在实际应用中并不常见。

优点：

性能高，Redis的读写速度非常快。
实现相对简单。
可以设置过期时间，避免死锁。

缺点：

需要引入额外的组件 Redis。
存在一定的复杂性，需要保证操作的原子性。
Redlock算法虽然提高了可靠性，但也增加了复杂性。
即使使用Redlock，也依然存在争议，其安全性并没有得到完全证明。

3. 基于 ZooKeeper 的锁

ZooKeeper 是一个分布式协调服务，提供数据一致性保证。利用 ZooKeeper 的临时节点和 Watcher 机制可以实现可靠的分布式锁。

实现方式：

创建一个持久节点作为锁的根节点。
客户端尝试在锁节点下创建一个临时顺序节点。
客户端获取锁节点下的所有子节点，并判断自己创建的节点是否是序号最小的节点。如果是，则获得锁；否则，监听序号比自己小的节点的变化。
如果序号比自己小的节点被删除，则客户端再次尝试获取锁。
释放锁时，删除自己创建的临时节点。

Python代码示例：

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

class ZookeeperLock:
    def __init__(self, hosts, lock_name):
        self.hosts = hosts
        self.lock_name = lock_name
        self.client = KazooClient(hosts=self.hosts)
        self.lock_path = "/locks/" + self.lock_name
        self.client_id = str(uuid.uuid4())
        self.lock_node = None # 保存创建的临时顺序节点

    def start(self):
        self.client.start()
        # 创建锁的根节点，如果不存在
        if not self.client.exists("/locks"):
            self.client.create("/locks", makepath=True)

    def acquire_lock(self, timeout=10):
        self.start()
        start_time = time.time()

        while True:
            try:
                # 创建临时顺序节点
                self.lock_node = self.client.create(self.lock_path + "/", ephemeral=True, sequence=True)

                # 获取所有子节点并排序
                children = sorted(self.client.get_children("/locks/" + self.lock_name[:self.lock_name.rfind('/')])) # 处理lock_name包含/的情况

                # 判断自己是否是序号最小的节点
                if self.lock_node.split("/")[-1] == children[0]:
                    # 成功获取锁
                    return True
                else:
                    # 监听序号比自己小的节点
                    index = children.index(self.lock_node.split("/")[-1])
                    predecessor = children[index - 1]
                    predecessor_path = "/locks/" + self.lock_name[:self.lock_name.rfind('/')] + "/" + predecessor

                    # 监听前一个节点
                    def watch_function(event):
                        pass  #  不需要做任何操作，触发回调即可

                    if self.client.exists(predecessor_path): # 确保前一个节点存在
                        self.client.get(predecessor_path, watch=watch_function) # 监听前一个节点的变化

            except NodeExistsError:
                # 锁节点已经存在，通常是并发创建导致
                pass # 继续尝试获取锁
            except Exception as e:
                print(f"Error acquiring lock: {e}")
                return False

            if time.time() - start_time > timeout:
                # 获取锁超时
                self.release_lock() # 超时后释放锁，防止死锁
                return False

            time.sleep(0.01)

    def release_lock(self):
        try:
            if self.lock_node and self.client.exists(self.lock_node):
                self.client.delete(self.lock_node)
        except Exception as e:
            print(f"Error releasing lock: {e}")
        finally:
            self.client.stop()
            self.lock_node = None

# 使用示例
if __name__ == '__main__':
    lock = ZookeeperLock(hosts="127.0.0.1:2181", lock_name="my_resource_lock")

    if lock.acquire_lock(timeout=5):
        print("Successfully acquired lock!")
        try:
            # 访问共享资源
            print("Processing shared resource...")
            time.sleep(5)  # 模拟资源处理
        finally:
            lock.release_lock()
            print("Released lock.")
    else:
        print("Failed to acquire lock.")

优点：

可靠性高，ZooKeeper具有高可用性。
可以保证锁的公平性（按照请求顺序获取锁）。
可以避免死锁（临时节点在客户端崩溃时会自动删除）。

缺点：

性能相对较低，ZooKeeper的读写速度不如Redis。
实现相对复杂。
需要引入额外的组件 ZooKeeper。

4. 基于 etcd 的锁

etcd 是一个分布式键值存储系统，常用于服务发现和配置管理。 etcd 提供了 Lease 机制和 Watch 机制，可以用于实现分布式锁。

实现方式：

创建一个 Lease，并设置过期时间。
使用 Lease 创建一个键值对，如果创建成功，则获得锁；如果创建失败，则表示锁已被其他客户端持有。
监听锁键的变化，如果锁键被删除，则表示锁已被释放。
客户端可以通过续租 Lease 来延长锁的持有时间。
释放锁时，删除锁键或撤销 Lease。

Python代码示例：

import etcd3
import time
import uuid

class EtcdLock:
    def __init__(self, host, port, lock_name, ttl=10):
        self.etcd = etcd3.Etcd3Client(host=host, port=port)
        self.lock_name = lock_name
        self.ttl = ttl
        self.lease = None
        self.client_id = str(uuid.uuid4())
        self.lock_key = '/locks/' + self.lock_name  # 锁的键

    def acquire_lock(self, timeout=10):
        start_time = time.time()

        while True:
            try:
                # 创建 Lease
                self.lease = self.etcd.lease(self.ttl)

                # 尝试创建锁键
                acquired = self.etcd.put(self.lock_key, self.client_id, lease=self.lease) # value 可以存 client_id

                return True

            except etcd3.exceptions.PreconditionFailedError: # 键已存在
                # 锁已被占用

                if time.time() - start_time > timeout:
                    self.release_lock()
                    return False # 获取锁超时

                time.sleep(0.01) # 避免忙等待

            except Exception as e:
                print(f"Error acquiring lock: {e}")
                self.release_lock()
                return False

    def release_lock(self):
        try:
           if self.lease:
               self.etcd.delete(self.lock_key)
               self.lease.revoke() # 撤销租约
        except Exception as e:
            print(f"Error releasing lock: {e}")

    def keep_alive(self, interval=2):
        # 续租 Lease，保持锁的有效性
        while True:
            try:
                if self.lease:
                    self.lease.refresh()
                    time.sleep(interval)
                else:
                    break  # 锁已被释放
            except Exception as e:
                print(f"Error keeping lock alive: {e}")
                break

# 使用示例
if __name__ == '__main__':
    lock = EtcdLock(host="localhost", port=2379, lock_name="my_resource_lock", ttl=10)

    if lock.acquire_lock(timeout=5):
        print("Successfully acquired lock!")
        try:
            # 启动线程续租 Lease
            import threading
            keep_alive_thread = threading.Thread(target=lock.keep_alive)
            keep_alive_thread.daemon = True # 设置为守护线程
            keep_alive_thread.start()

            # 访问共享资源
            print("Processing shared resource...")
            time.sleep(5)  # 模拟资源处理
        finally:
            lock.release_lock()
            print("Released lock.")
    else:
        print("Failed to acquire lock.")

优点：

可靠性高，etcd具有高可用性。
支持 Lease 机制，可以自动续租，避免锁过期。
使用 gRPC 协议，性能较高。

缺点：

实现相对复杂。
需要引入额外的组件 etcd。
需要手动续租 Lease，增加了实现的复杂性。

锁方案对比

特性	数据库锁	Redis 锁	ZooKeeper 锁	etcd 锁
性能	较低	较高	较低	较高
可靠性	较低	中等	较高	较高
实现难度	简单	中等	复杂	复杂
额外依赖	无	Redis	ZooKeeper	etcd
死锁避免	需要额外机制	设置过期时间	临时节点	Lease机制
公平性	不保证	不保证	保证	不保证

如何选择合适的分布式锁方案

选择合适的分布式锁方案需要根据具体的应用场景进行权衡。

如果对性能要求很高，且可以容忍一定的锁丢失风险，可以选择Redis锁。
如果对可靠性要求很高，可以选择ZooKeeper锁或etcd锁。
如果应用本身已经使用了ZooKeeper或etcd，则可以优先选择对应的锁方案。
如果只是简单的互斥需求，且对性能要求不高，可以选择数据库锁。

此外，还需要考虑实现的复杂度和维护成本。

分布式锁的常见问题

死锁： 如果持有锁的客户端崩溃，可能导致死锁。可以通过设置过期时间或使用临时节点来避免死锁。
锁过期： 如果锁的过期时间设置得太短，可能导致锁在客户端完成操作之前过期，从而导致多个客户端同时访问共享资源。可以通过续租 Lease 或使用 Watch 机制来延长锁的持有时间。
脑裂： 在分布式系统中，由于网络问题，可能导致出现多个独立的集群。这时，如果每个集群都认为自己拥有锁，就会导致数据不一致。 Redlock 算法可以缓解脑裂问题，但并不能完全解决。
锁的粒度： 锁的粒度越细，并发性能越高，但实现复杂度也越高。需要根据具体的应用场景选择合适的锁粒度。
重试机制： 获取锁失败时，需要进行重试。需要设置合理的重试次数和重试间隔，避免过度消耗资源。

最佳实践

选择合适的分布式锁方案： 根据具体的应用场景进行权衡，选择最合适的方案。
设置合理的过期时间： 避免锁在客户端完成操作之前过期。
使用原子操作： 保证锁的获取和释放操作的原子性。
使用 Lua 脚本： 避免误删锁。
添加监控和告警： 监控锁的获取和释放情况，及时发现和解决问题。
进行压力测试： 模拟高并发场景，验证锁的性能和可靠性。

保证资源访问的互斥性与一致性

分布式锁是保证分布式系统中资源访问的互斥性与一致性的重要手段。通过选择合适的锁方案，并遵循最佳实践，可以有效地避免并发问题，提高系统的可靠性和性能。数据库锁易于实现，但性能较低；Redis锁性能高，但可靠性稍逊；ZooKeeper锁和etcd锁可靠性高，但实现较为复杂。选择适合应用场景的锁方案，并注意死锁、过期等常见问题，才能确保分布式锁的有效性。

更多IT精英技术系列讲座，到智猿学院

Python 分布式锁机制：保证资源访问的互斥性与一致性

什么是分布式锁

分布式锁的实现方式

1. 基于数据库的锁

2. 基于 Redis 的锁

3. 基于 ZooKeeper 的锁

4. 基于 etcd 的锁

锁方案对比

如何选择合适的分布式锁方案

分布式锁的常见问题

最佳实践

保证资源访问的互斥性与一致性

发表回复 取消回复

发表回复取消回复