Serializable事务批量插入序列化错误原因及高并发解决方案咨询

操作步骤 #

• 创建两张表participants和participant_dates；
• 在一个SERIALIZABLE隔离级别的事务中，插入一条participant记录及其对应的多条participant_dates记录。

完整Python代码 #

import asyncio
from datetime import date
from uuid import uuid4

import asyncpg

async def get_connection():
    return await asyncpg.connect(
        user='postgres',
        password='1234',
        database='postgres',
        host='127.0.0.1',
        port='5432'
    )


async def bulk() -> None:
    """
    CREATE EXTENSION IF NOT EXISTS "uuid-ossp";

    CREATE TABLE IF NOT EXISTS participants (
        id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
        name VARCHAR(255) NOT NULL
    );

    CREATE TABLE IF NOT EXISTS participant_dates (
        id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
        participant_id UUID REFERENCES participants(id),
        date DATE
    );
    """
    conn = await get_connection()
    try:
        async with conn.transaction(isolation="serializable"):
            id_ = await conn.fetchval(
                f""" 
            INSERT INTO participants (name) VALUES ('test_{str(uuid4())}') RETURNING id;
            """
            )
            await conn.executemany(
                """
                INSERT INTO participant_dates (participant_id, date) VALUES (
                    $1, 
                    $2
                ); 
                """,
                [
                    (id_, date(2025, 1, 1)),
                    (id_, date(2025, 1, 2)),
                    (id_, date(2025, 1, 3)),
                    (id_, date(2025, 1, 4)),
                    (id_, date(2025, 1, 5)),
                    (id_, date(2025, 1, 6)),
                    (id_, date(2025, 1, 7)),
                    (id_, date(2025, 1, 8)),
                    (id_, date(2025, 1, 9)),
                    (id_, date(2025, 1, 10)),
                    (id_, date(2025, 1, 11)),
                    (id_, date(2025, 1, 12)),
                    (id_, date(2025, 1, 13)),
                    (id_, date(2025, 1, 14)),
                    (id_, date(2025, 1, 15)),
                    (id_, date(2025, 1, 16)),
                    (id_, date(2025, 1, 17)),
                    (id_, date(2025, 1, 18)),
                    (id_, date(2025, 1, 19)),
                    (id_, date(2025, 1, 20)),
                ]
            )

    except Exception as e:
        print(e)


async def main() -> None:
    # 50 concurrent call
    await asyncio.gather(
        bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(),
        bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(),
        bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(),
        bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(),
        bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(), bulk(),
    )

asyncio.run(main())

问题现象 #

执行50次并发调用后出现大量如下错误：

无法序列化访问，因为事务之间存在读写依赖关系
详细信息: 原因代码：读取期间因冲突取消，涉及支点882。
提示: 重试事务可能会成功。

需求说明 #

• 无法降低事务隔离级别（使用EdgeDB，强制要求SERIALIZABLE）；
• 需要避免上述序列化错误；
• 需维持高并发能力（支持50及以上并发调用）。

疑问 #

为什么会出现冲突？我用UUID作为主键（非自增整数，不需要锁），所有事务插入的都是不同行，冲突（锁）来自哪里？

补充信息 #

即使添加SET enable_seqscan = off;语句禁用序列扫描，错误仍然存在。修改后的代码片段及新错误信息如下：

修改后的代码片段 #

await conn.execute(
    "SET enable_seqscan = off;"
)
id_ = await conn.fetchval(
    f"""
INSERT INTO participants (name) VALUES ('test_{str(uuid4())}') RETURNING id;
"""
...
)

新错误信息 #

无法序列化访问，因为事务之间存在读写依赖关系
详细信息: 原因代码：提交尝试时因预准备支点中的冲突取消。
提示: 重试事务可能会成功。

问题原因分析 #

即使使用UUID主键，SERIALIZABLE隔离级别下仍会触发冲突，核心原因有两点：

1. 快照依赖追踪
   SERIALIZABLE模式下，PostgreSQL会严格追踪所有事务的执行顺序是否符合可串行化规则。即使插入的是完全独立的行，事务在执行过程中会隐式读取表的元数据（如统计信息、索引状态），这些共享资源的访问会被判定为潜在的依赖冲突，触发序列化校验失败。

2. 外键约束的隐式检查
   participant_dates表的外键participant_id关联participants主键，插入从表记录时，数据库会验证主表记录存在。在SERIALIZABLE模式下，这个验证操作会被纳入事务依赖链，当大量并发事务同时执行主从表插入时，数据库会认为事务执行顺序存在歧义，进而触发冲突。

解决方案 #

针对你的需求，推荐以下可行方案：

1. 实现事务自动重试机制 #

序列化冲突属于临时性错误，按照提示重试即可解决。修改bulk函数添加重试逻辑：

async def bulk() -> None:
    max_retries = 5
    for attempt in range(max_retries):
        conn = await get_connection()
        try:
            async with conn.transaction(isolation="serializable"):
                id_ = await conn.fetchval(
                    f""" 
                INSERT INTO participants (name) VALUES ('test_{str(uuid4())}') RETURNING id;
                """
                )
                await conn.executemany(
                    """
                    INSERT INTO participant_dates (participant_id, date) VALUES ($1, $2); 
                    """,
                    [(id_, date(2025, 1, d)) for d in range(1,21)]
                )
            break
        except asyncpg.exceptions.SerializationError as e:
            if attempt == max_retries -1:
                print(f"重试{max_retries}次后仍失败: {e}")
            else:
                await asyncio.sleep(0.1)  # 短暂等待后重试
        finally:
            await conn.close()

2. 延迟外键检查 #

将外键约束设置为延迟检查，减少事务执行过程中的依赖追踪。修改participant_dates表结构：

CREATE TABLE IF NOT EXISTS participant_dates (
    id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
    participant_id UUID REFERENCES participants(id) DEFERRABLE INITIALLY DEFERRED,
    date DATE
);

3. 优化批量插入逻辑 #

通过合并SQL语句减少事务内操作步骤，比如用INSERT ... RETURNING结合UNNEST一次性完成主从表插入，降低依赖冲突概率：

WITH inserted_participant AS (
    INSERT INTO participants (name) VALUES ($1) RETURNING id
)
INSERT INTO participant_dates (participant_id, date)
SELECT id, unnest($2::date[]) FROM inserted_participant;

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