多Python脚本更新PostgreSQL时如何保证数据库隔离避免冲突？

针对你的场景（定时cron脚本+手动脚本更新同一张PostgreSQL表），可以通过以下几种方式避免更新冲突：

1. 行级锁+事务控制（推荐） #

PostgreSQL支持行级锁，在更新前锁定要修改的行，确保同一时间只有一个脚本能修改特定行。结合事务的原子性，能有效避免冲突。

示例代码（psycopg2）： #

import psycopg2
from psycopg2 import OperationalError

def update_data():
    conn = None
    try:
        conn = psycopg2.connect(
            dbname="your_db",
            user="your_user",
            password="your_pass",
            host="your_host"
        )
        cur = conn.cursor()
        # 设置事务隔离级别为REPEATABLE READ（增强一致性）
        conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_REPEATABLE_READ)
        
        # 锁定需要更新的行，其他脚本会等待锁释放
        cur.execute("SELECT id FROM your_table WHERE your_condition FOR UPDATE;")
        
        # 执行更新操作
        cur.execute("UPDATE your_table SET column1 = %s WHERE your_condition;", (new_value,))
        
        conn.commit()
        print("更新完成")
    except OperationalError as e:
        if conn:
            conn.rollback()
        print(f"更新失败：{e}")
    finally:
        if conn:
            conn.close()

if __name__ == "__main__":
    update_data()

注意：FOR UPDATE会锁定匹配的行，直到事务提交/回滚。如果是批量操作，锁定必要的行而非全表，能尽量减少对并发的影响。

2. 表级排他锁（适合全表更新场景） #

如果脚本需要更新全表或大部分数据，可以直接给表加排他锁，确保同一时间只有一个脚本能操作该表。

示例代码： #

def full_table_update():
    conn = None
    try:
        conn = psycopg2.connect(...)
        cur = conn.cursor()
        # 加表级排他锁，其他更新操作会等待
        cur.execute("LOCK TABLE your_table IN EXCLUSIVE MODE;")
        
        # 执行全表更新逻辑
        cur.execute("UPDATE your_table SET column1 = %s;", (new_value,))
        
        conn.commit()
    except OperationalError as e:
        if conn:
            conn.rollback()
        print(f"更新失败：{e}")
    finally:
        if conn:
            conn.close()

注意：表锁会阻塞所有其他写操作，只适合执行时间短的全表更新，避免影响cron脚本的定时执行。

3. 文件系统锁（跨脚本的外部锁） #

利用文件系统的排他锁，让两个脚本互相感知对方是否在运行。这种方式不需要修改数据库，适合简单场景。

示例代码： #

import fcntl
import os
import psycopg2

LOCK_FILE = "/tmp/db_update.lock"

def update_with_file_lock():
    lock_fd = None
    conn = None
    try:
        # 打开锁文件，不存在则创建
        lock_fd = open(LOCK_FILE, "w")
        # 加排他锁，非阻塞模式（如果锁被占用直接退出）
        fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        
        # 执行数据库更新逻辑
        conn = psycopg2.connect(...)
        cur = conn.cursor()
        cur.execute("UPDATE your_table SET ...;")
        conn.commit()
        print("更新完成")
    except BlockingIOError:
        print("已有脚本在执行更新，请稍后再试")
    except OperationalError as e:
        if conn:
            conn.rollback()
        print(f"更新失败：{e}")
    finally:
        if lock_fd:
            fcntl.flock(lock_fd, fcntl.LOCK_UN)
            lock_fd.close()
        if conn:
            conn.close()

if __name__ == "__main__":
    update_with_file_lock()

注意：锁文件要放在所有脚本都能访问的路径下，且必须确保脚本异常退出时锁能被释放（finally块的解锁逻辑很重要）。

4. 数据库状态表锁 #

在数据库中创建一个专门的锁表，用来标记当前是否有脚本在执行更新。这种方式更可靠，避免文件锁在分布式环境下的问题。

步骤： #

1. 先创建锁表：

CREATE TABLE update_locks (
    script_name VARCHAR(50) PRIMARY KEY,
    is_running BOOLEAN DEFAULT false,
    last_run TIMESTAMP
);

2. 脚本中检查并设置锁：

def update_with_db_lock():
    conn = None
    try:
        conn = psycopg2.connect(...)
        cur = conn.cursor()
        # 尝试获取锁：如果脚本未在运行，则标记为运行中
        cur.execute("""
            INSERT INTO update_locks (script_name, is_running, last_run)
            VALUES ('manual_script', true, NOW())
            ON CONFLICT (script_name) DO UPDATE
            SET is_running = true, last_run = NOW()
            WHERE NOT update_locks.is_running;
        """)
        # 检查是否成功获取锁（影响行数为1则成功）
        if cur.rowcount == 0:
            print("已有脚本在执行更新，请稍后再试")
            conn.rollback()
            return
        
        # 执行更新逻辑
        cur.execute("UPDATE your_table SET ...;")
        conn.commit()
        
        # 更新锁状态为未运行
        cur.execute("UPDATE update_locks SET is_running = false WHERE script_name = 'manual_script';")
        conn.commit()
    except OperationalError as e:
        if conn:
            conn.rollback()
        # 异常时重置锁状态
        cur.execute("UPDATE update_locks SET is_running = false WHERE script_name = 'manual_script';")
        conn.commit()
        print(f"更新失败：{e}")
    finally:
        if conn:
            conn.close()

注意：cron脚本也要加入相同的锁检查逻辑，确保双向互斥。

内容的提问来源于stack exchange，提问作者bendo97