在 PostgreSQL 中如何实现数据的分布式锁机制？

在 PostgreSQL 中如何实现数据的分布式锁机制？

引用

CSDN

1.

https://blog.csdn.net/2401_86074221/article/details/140284079

在分布式系统中，确保对共享资源的并发访问控制是至关重要的，分布式锁机制则是实现这种控制的常见手段。本文将介绍在PostgreSQL中实现分布式锁的三种方法：使用advisory locks、使用表模拟分布式锁以及结合事务和条件更新实现分布式锁。

一、使用 advisory locks（建议锁）

PostgreSQL 提供了 advisory locks 来处理并发控制。advisory locks 是一种应用级别的锁，可以由应用程序自己定义锁的键值来进行加锁和解锁操作。

-- 获取共享建议锁
SELECT pg_advisory_lock_shared(42);
-- 获取独占建议锁
SELECT pg_advisory_lock(42);
-- 释放共享建议锁
SELECT pg_advisory_unlock_shared(42);
-- 释放独占建议锁
SELECT pg_advisory_unlock(42);

在上述代码中，42 是自定义的锁键值，可以是任意整数。pg_advisory_lock_shared 用于获取共享建议锁，多个事务可以同时持有同一个共享建议锁。pg_advisory_lock 用于获取独占建议锁，在同一时间只有一个事务可以持有独占建议锁。

优点：

• 实现相对简单，直接使用内置函数。
• 性能开销相对较小。

缺点：

• 锁的范围完全由应用程序定义的键值控制，可能存在误操作或不一致。

二、使用表模拟分布式锁

创建一张专门用于存储锁信息的表：

CREATE TABLE distributed_locks (
    lock_name VARCHAR(255) PRIMARY KEY,
    locked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    locked_by VARCHAR(255)
);

获取锁的操作：

INSERT INTO distributed_locks (lock_name, locked_by)
VALUES ('lock1', 'current_process')
ON CONFLICT (lock_name) DO NOTHING;
-- 检查是否获取成功
SELECT COUNT(*) FROM distributed_locks WHERE lock_name = 'lock1' AND locked_by = 'current_process';

释放锁的操作：

DELETE FROM distributed_locks WHERE lock_name = 'lock1' AND locked_by = 'current_process';

优点：

• 灵活性高，可以记录更多关于锁的信息，如获取时间、所有者等。

缺点：

• 涉及到表的插入、删除操作，性能开销相对较大。

三、结合事务和条件更新实现分布式锁

BEGIN;
-- 尝试获取锁
UPDATE distributed_locks
SET locked_by = 'current_process'
WHERE lock_name = 'lock1' AND locked_by IS NULL;
-- 如果更新影响的行数为 0，表示获取锁失败
IF NOT FOUND THEN
    ROLLBACK;
    -- 处理获取锁失败的情况
ELSE
    -- 执行锁定期间的操作
    -- 释放锁
    UPDATE distributed_locks
    SET locked_by = NULL
    WHERE lock_name = 'lock1' AND locked_by = 'current_process';
    COMMIT;
END IF;

优点：

• 能够精确控制锁的获取和释放逻辑。

缺点：

• 代码相对复杂，需要对事务和条件更新有深入理解。

下面是一个更详细的示例，展示如何在一个实际的场景中使用上述方法来实现分布式锁。

假设我们有一个分布式的任务处理系统，多个节点需要竞争处理一个任务。当一个节点获取到任务的处理权（即获取到锁）后，它会执行一系列的操作，完成后释放锁，让其他节点有机会处理。

使用 advisory locks 的示例

-- 节点 1 尝试获取独占建议锁
BEGIN;
SELECT pg_advisory_lock(12345); 
-- 执行任务处理逻辑
--...
SELECT pg_advisory_unlock(12345);
COMMIT;

-- 节点 2 同时尝试获取独占建议锁
BEGIN;
SELECT pg_advisory_lock(12345); 
-- 由于节点 1 持有锁，将阻塞等待
COMMIT;

在这个示例中，节点 1 成功获取到锁并执行任务，节点 2 在获取锁时会被阻塞，直到节点 1 释放锁。

使用表模拟分布式锁的示例

-- 节点 1 获取锁的操作
BEGIN;
INSERT INTO distributed_locks (lock_name, locked_by)
VALUES ('task_lock', 'node1')
ON CONFLICT (lock_name) DO NOTHING;
-- 检查是否获取成功
SELECT COUNT(*) FROM distributed_locks WHERE lock_name = 'task_lock' AND locked_by = 'node1';
-- 若获取成功，执行任务处理逻辑
--...
-- 释放锁
DELETE FROM distributed_locks WHERE lock_name = 'task_lock' AND locked_by = 'node1';
COMMIT;

-- 节点 2 获取锁的操作
BEGIN;
INSERT INTO distributed_locks (lock_name, locked_by)
VALUES ('task_lock', 'node2')
ON CONFLICT (lock_name) DO NOTHING;
-- 检查是否获取成功
SELECT COUNT(*) FROM distributed_locks WHERE lock_name = 'task_lock' AND locked_by = 'node2';
-- 若未获取成功，则等待或执行其他操作
--...
COMMIT;

在这个示例中，通过对锁表的操作来实现锁的获取和释放。

结合事务和条件更新实现分布式锁的示例

-- 节点 1
BEGIN;
UPDATE distributed_locks
SET locked_by = 'node1'
WHERE lock_name = 'task_lock' AND locked_by IS NULL;
-- 如果更新影响的行数为 0，表示获取锁失败
IF NOT FOUND THEN
    ROLLBACK;
    -- 处理获取锁失败的情况
ELSE
    -- 执行锁定期间的任务处理操作
    --...
    -- 释放锁
    UPDATE distributed_locks
    SET locked_by = NULL
    WHERE lock_name = 'task_lock' AND locked_by = 'node1';
    COMMIT;
END IF;

-- 节点 2
BEGIN;
UPDATE distributed_locks
SET locked_by = 'node2'
WHERE lock_name = 'task_lock' AND locked_by IS NULL;
-- 如果更新影响的行数为 0，表示获取锁失败
IF NOT FOUND THEN
    ROLLBACK;
    -- 处理获取锁失败的情况
ELSE
    -- 执行锁定期间的任务处理操作
    --...
    -- 释放锁
    UPDATE distributed_locks
    SET locked_by = NULL
    WHERE lock_name = 'task_lock' AND locked_by = 'node2';
    COMMIT;
END IF;

通过这种方式，能够准确地在事务中进行锁的获取和释放，确保数据的一致性。

在实际应用中，选择哪种方式实现分布式锁取决于具体的需求和系统架构。如果只是简单的并发控制，并且对性能要求较高，可以选择使用 advisory locks。如果需要更复杂的锁管理和更多的锁相关信息记录，可以使用表模拟分布式锁或者结合事务和条件更新的方式。

此外，无论使用哪种方式，都需要注意处理锁获取失败、死锁等异常情况，以保证系统的稳定性和可靠性。同时，还要考虑分布式环境中的故障恢复、节点间的通信延迟等因素对锁机制的影响。