Python多进程操作PostgreSQL读写并行问题排查

核心问题及修复点 #

1. 数据库连接未做进程隔离 #

PostgreSQL的数据库连接是进程绑定的，不能在主进程创建连接后直接给子进程复用。如果子进程用了主进程的连接，会导致连接状态异常，根本读不到数据。

• 修复：在write_database和read_database函数内部各自独立创建数据库连接和游标，不要在主进程初始化后传递给子进程。

2. 并行时序与事务可见性问题 #

读进程可能在写进程还没插入emp_id=60，或者插入后未提交事务时就执行查询，这时候自然读不到数据。PostgreSQL默认的READ COMMITTED隔离级别下，只有已提交的事务数据才会被其他进程看到。

• 修复：
  • 写进程每次插入（或插入到60时）及时提交事务，不要攒到最后才提交；
  • 读进程增加循环查询逻辑，直到找到目标数据再执行更新，不要只查一次就处理。

3. 未处理空查询结果的情况 #

你直接读取data的索引但没判断是否为空，一旦查询不到数据就会触发索引越界报错。

• 修复：查询后先判断if data，确认有数据后再执行更新操作。

示例修正代码 #

import multiprocessing
import psycopg2
from time import sleep

def write_database():
    # 子进程内独立创建连接
    conn = psycopg2.connect(dbname="你的数据库名", user="用户名", password="密码", host="localhost")
    cur = conn.cursor()
    try:
        for emp_id in range(100):
            cur.execute("INSERT INTO employees (emp_id) VALUES (%s)", (emp_id,))
            conn.commit()  # 每次插入后提交，确保读进程能看到数据
            sleep(0.01)  # 模拟业务耗时，给读进程触发机会
    except Exception as e:
        conn.rollback()
        print(f"写入失败: {e}")
    finally:
        cur.close()
        conn.close()

def read_database():
    # 子进程内独立创建连接
    conn = psycopg2.connect(dbname="你的数据库名", user="用户名", password="密码", host="localhost")
    cur = conn.cursor()
    try:
        while True:
            cur.execute("SELECT emp_id FROM employees WHERE emp_id = %s", (60,))
            data = cur.fetchall()
            if data:
                # 找到数据后执行更新
                cur.execute("UPDATE employees SET emp_id = %s WHERE emp_id = %s", (1000, 60))
                conn.commit()
                print("emp_id=60已更新为1000")
                break
            sleep(0.05)  # 轮询间隔，避免频繁查询数据库
    except Exception as e:
        conn.rollback()
        print(f"读/更新失败: {e}")
    finally:
        cur.close()
        conn.close()

if __name__ == "__main__":
    write_proc = multiprocessing.Process(target=write_database)
    read_proc = multiprocessing.Process(target=read_database)
    write_proc.start()
    read_proc.start()
    write_proc.join()
    read_proc.join()

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