非规范化表并发触发器与事务隔离问题解决方案咨询

表结构 #

主表 #

CREATE TABLE profiles(id int PRIMARY KEY, active bool);
CREATE TABLE memberships(scope VARCHAR(10), profile int);

非规范化整合表 #

CREATE TABLE denormalised(scope VARCHAR(10), profile int, active bool);

核心需求 #

• 插入新profile时，自动关联对应的membership数据插入到denormalised表
• 插入新membership时，自动关联对应的profile数据插入到denormalised表

当前触发器实现 #

CREATE OR REPLACE FUNCTION on_membership_insert()
    RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO denormalised (scope, profile, active)
    SELECT NEW.scope, NEW.profile, p.active
    FROM profiles p
    WHERE p.id = NEW.profile
    FOR UPDATE
    ON CONFLICT DO NOTHING;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trigger_membership_insert
    AFTER INSERT ON memberships
    FOR EACH ROW
EXECUTE FUNCTION on_membership_insert();


CREATE OR REPLACE FUNCTION on_profile_insert()
    RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO denormalised (scope, profile, active)
    SELECT m.scope, NEW.id, NEW.active
    FROM memberships m
    WHERE m.profile = NEW.id
    FOR UPDATE
    ON CONFLICT DO NOTHING;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trigger_profile_insert
    AFTER INSERT ON profiles
    FOR EACH ROW
EXECUTE FUNCTION on_profile_insert();

并发插入问题 #

当profiles和memberships表同时执行并发插入时，denormalised表会出现无数据插入的情况，示例事务流程：
Tx1

BEGIN;
INSERT INTO profiles(id, active) VALUES (8, true);
-- 等待，暂不提交

Tx2

BEGIN;
INSERT INTO memberships(scope, profile) VALUES ('ub2', 8);
COMMIT;
-- 此时提交Tx1

问题根源：事务无法感知对方未提交的操作，且FOR UPDATE不会触发EvalPlanQual机制重新查询已提交的关联行。

疑问 #

如何正确处理profiles和memberships的并发插入？SERIALIZABLE隔离级别是可行方案，但有没有更优的替代选择？

方案1：使用SERIALIZABLE隔离级别 #

这是最直接的一致性保障方案，将事务隔离级别设置为SERIALIZABLE：

SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;

并发执行Tx1和Tx2时，其中一个事务会被检测到冲突并回滚，需要应用层添加重试逻辑。该方案能严格保证数据一致性，但会增加应用复杂度，高并发场景下可能降低吞吐量。

方案2：唯一约束+定期补全 #

步骤1：给denormalised添加唯一约束 #

ALTER TABLE denormalised ADD CONSTRAINT denormalised_unique UNIQUE (scope, profile);

步骤2：简化触发器（移除FOR UPDATE） #

触发器保留原有插入逻辑，但去掉不必要的行锁定，依赖ON CONFLICT处理重复数据。

步骤3：编写补全函数并定期执行 #

创建同步函数处理并发插入时遗漏的关联数据：

CREATE OR REPLACE FUNCTION sync_denormalised()
RETURNS void AS $$
BEGIN
    -- 补全memberships已存在、profile后插入的情况
    INSERT INTO denormalised (scope, profile, active)
    SELECT m.scope, m.profile, p.active
    FROM memberships m
    JOIN profiles p ON m.profile = p.id
    LEFT JOIN denormalised d ON d.scope = m.scope AND d.profile = m.profile
    WHERE d.scope IS NULL;

    -- 补全profile已存在、membership后插入的情况
    INSERT INTO denormalised (scope, profile, active)
    SELECT m.scope, p.id, p.active
    FROM profiles p
    JOIN memberships m ON p.id = m.profile
    LEFT JOIN denormalised d ON d.scope = m.scope AND d.profile = p.id
    WHERE d.scope IS NULL;
END;
$$ LANGUAGE plpgsql;

通过pg_cron扩展定期执行（需先安装）：

SELECT cron.schedule('sync-denormalised', '* * * * *', 'SELECT sync_denormalised();');

该方案不影响主插入性能，通过异步补全保证最终一致性，适合对实时性要求不高的场景。

方案3：NOTIFY/LISTEN异步同步 #

步骤1：修改触发器添加通知逻辑 #

CREATE OR REPLACE FUNCTION on_membership_insert()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO denormalised (scope, profile, active)
    SELECT NEW.scope, NEW.profile, p.active
    FROM profiles p
    WHERE p.id = NEW.profile
    ON CONFLICT DO NOTHING;

    -- 找不到对应profile时发送通知
    IF NOT EXISTS (SELECT 1 FROM profiles WHERE id = NEW.profile) THEN
        PERFORM pg_notify('profile_inserted', NEW.profile::text);
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION on_profile_insert()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO denormalised (scope, profile, active)
    SELECT m.scope, NEW.id, NEW.active
    FROM memberships m
    WHERE m.profile = NEW.id
    ON CONFLICT DO NOTHING;

    -- 找不到对应membership时发送通知
    IF NOT EXISTS (SELECT 1 FROM memberships WHERE profile = NEW.id) THEN
        PERFORM pg_notify('membership_inserted', NEW.id::text);
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

步骤2：编写后台监听进程 #

用Python/Go等语言编写进程监听通知，收到通知后执行同步逻辑（示例Python代码）：

import psycopg2
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT

conn = psycopg2.connect("dbname=your_db user=your_user")
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
cur = conn.cursor()

cur.execute("LISTEN profile_inserted;")
cur.execute("LISTEN membership_inserted;")

while True:
    conn.poll()
    while conn.notifies:
        notify = conn.notifies.pop(0)
        profile_id = int(notify.payload)
        if notify.channel == 'profile_inserted':
            cur.execute("""
                INSERT INTO denormalised (scope, profile, active)
                SELECT m.scope, m.profile, p.active
                FROM memberships m
                JOIN profiles p ON m.profile = p.id
                WHERE m.profile = %s
                ON CONFLICT DO NOTHING;
            """, (profile_id,))
        elif notify.channel == 'membership_inserted':
            cur.execute("""
                INSERT INTO denormalised (scope, profile, active)
                SELECT m.scope, m.profile, p.active
                FROM memberships m
                JOIN profiles p ON m.profile = p.id
                WHERE p.id = %s
                ON CONFLICT DO NOTHING;
            """, (profile_id,))
        conn.commit()

该方案能实现准实时同步，不影响主事务性能，但需要额外维护后台监听进程。

方案4：延迟外键约束 #

给memberships添加延迟外键约束，强制事务提交时才检查关联关系：

ALTER TABLE memberships ADD CONSTRAINT fk_membership_profile FOREIGN KEY (profile) REFERENCES profiles(id) DEFERRABLE INITIALLY DEFERRED;

该方案仅能解决“先插入membership再插入profile”的场景，且要求关联插入操作在同一事务中，否则Tx2会因外键约束失败回滚，适合能控制插入顺序的业务场景。

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