PostgreSQL 9.6中dblink持久连接触发器的错误处理与并发问题

你提到的并发场景下创建dblink命名连接导致的重复名称错误，确实是高插入量场景下很容易踩的坑。好在PostgreSQL的PL/pgSQL提供了类似Python try-except的异常处理机制——也就是BEGIN...EXCEPTION代码块，同时我们也可以结合咨询锁来更优雅地从根源避免这类并发冲突。

一、用PL/pgSQL异常处理捕获重复连接错误 #

首先直接针对你遇到的"duplicate connection name"错误，我们可以在创建连接的逻辑外包裹异常处理块，当捕获到这个特定错误时直接忽略（因为此时另一个进程已经创建好连接了）。另外还要考虑连接存在但已失效的情况（比如网络中断后连接断开），这时候执行dblink_exec会报错，我们也需要捕获这类错误并重新建立连接。

修改后的函数示例：

CREATE OR REPLACE FUNCTION db_link_trigger() RETURNS trigger AS $BODY$
DECLARE
    v_conn_exists BOOLEAN;
BEGIN
    -- 1. 检测连接是否存在
    v_conn_exists := COALESCE('dblinktest' = ANY (dblink_get_connections()), false);
    
    -- 2. 尝试创建连接（如果不存在）
    IF NOT v_conn_exists THEN
        BEGIN
            RAISE NOTICE 'dblink connection not established. Connecting now';
            PERFORM dblink_connect('dblinktest', 'hostaddr=192.168.1.30 port=5433 dbname=otherdb user=myuser password=mypassword');
        EXCEPTION
            WHEN duplicate_object THEN
                -- 捕获"duplicate connection name"错误，此时连接已被其他进程创建，直接跳过
                RAISE NOTICE 'Connection already created by another process';
            WHEN OTHERS THEN
                -- 处理其他连接错误（比如网络问题），可根据需要重试或抛出
                RAISE NOTICE 'Failed to establish connection: %', SQLERRM;
                RAISE; -- 重新抛出错误，也可根据业务逻辑调整为重试
        END;
    ELSE
        RAISE NOTICE 'dblink connection already established';
    END IF;

    -- 3. 尝试插入数据，若连接失效则重连后再尝试
    BEGIN
        -- 注意：这里改用参数化查询避免SQL注入风险
        PERFORM dblink_exec('dblinktest', 'insert into mytable(data) values($1);', ARRAY[NEW.data]);
    EXCEPTION
        WHEN others THEN
            -- 捕获连接失效类错误，重新连接后执行插入
            RAISE NOTICE 'Connection failed, reconnecting...';
            -- 先尝试关闭失效连接（如果存在）
            PERFORM COALESCE(dblink_disconnect('dblinktest'), true);
            -- 重新建立连接
            PERFORM dblink_connect('dblinktest', 'hostaddr=192.168.1.30 port=5433 dbname=otherdb user=myuser password=mypassword');
            -- 再次执行插入
            PERFORM dblink_exec('dblinktest', 'insert into mytable(data) values($1);', ARRAY[NEW.data]);
    END;

    RETURN NEW;
END;
$BODY$ LANGUAGE plpgsql VOLATILE COST 100;

关键说明： #

• duplicate_object是PostgreSQL内置的错误代码，正好对应"duplicate connection name"这类重复对象错误；
• 插入逻辑中加入异常处理，当连接失效时自动重连，保证数据能成功写入目标库；
• 替换了直接拼接NEW.data的写法，改用参数化查询避免SQL注入风险——这一点在生产环境一定要注意！

二、用咨询锁避免并发创建连接的冲突 #

另一种更高效的方式是使用咨询锁（Advisory Lock），确保同一时间只有一个进程执行创建连接的逻辑，从根源上避免重复创建的问题。这种方式不需要依赖异常捕获，性能更优，适合高并发场景。

修改后的函数示例：

CREATE OR REPLACE FUNCTION db_link_trigger() RETURNS trigger AS $BODY$
DECLARE
    v_conn_exists BOOLEAN;
    v_lock_acquired BOOLEAN;
BEGIN
    -- 1. 检测连接是否存在
    v_conn_exists := COALESCE('dblinktest' = ANY (dblink_get_connections()), false);
    
    -- 2. 如果连接不存在，获取咨询锁后再创建
    IF NOT v_conn_exists THEN
        -- 获取排他咨询锁（锁ID自定义，只要唯一即可），超时时间隐含在非阻塞逻辑里
        SELECT pg_try_advisory_lock(12345) INTO v_lock_acquired;
        
        IF v_lock_acquired THEN
            BEGIN
                -- 再次检查连接（避免等待锁期间其他进程已创建）
                v_conn_exists := COALESCE('dblinktest' = ANY (dblink_get_connections()), false);
                IF NOT v_conn_exists THEN
                    RAISE NOTICE 'dblink connection not established. Connecting now';
                    PERFORM dblink_connect('dblinktest', 'hostaddr=192.168.1.30 port=5433 dbname=otherdb user=myuser password=mypassword');
                END IF;
            EXCEPTION
                WHEN OTHERS THEN
                    RAISE NOTICE 'Failed to establish connection: %', SQLERRM;
                    RAISE;
            FINALLY
                -- 无论是否成功，都释放咨询锁，避免死锁
                PERFORM pg_advisory_unlock(12345);
            END;
        ELSE
            -- 获取锁失败，等待极短时间后重新检查连接
            PERFORM pg_sleep(0.1);
            v_conn_exists := COALESCE('dblinktest' = ANY (dblink_get_connections()), false);
            IF NOT v_conn_exists THEN
                RAISE EXCEPTION 'Cannot acquire lock to establish dblink connection';
            END IF;
        END IF;
    ELSE
        RAISE NOTICE 'dblink connection already established';
    END IF;

    -- 3. 插入数据，同样处理连接失效的情况
    BEGIN
        PERFORM dblink_exec('dblinktest', 'insert into mytable(data) values($1);', ARRAY[NEW.data]);
    EXCEPTION
        WHEN others THEN
            RAISE NOTICE 'Connection failed, reconnecting...';
            PERFORM COALESCE(dblink_disconnect('dblinktest'), true);
            PERFORM dblink_connect('dblinktest', 'hostaddr=192.168.1.30 port=5433 dbname=otherdb user=myuser password=mypassword');
            PERFORM dblink_exec('dblinktest', 'insert into mytable(data) values($1);', ARRAY[NEW.data]);
    END;

    RETURN NEW;
END;
$BODY$ LANGUAGE plpgsql VOLATILE COST 100;

关键说明： #

• pg_try_advisory_lock是无阻塞的锁获取方式，不会让进程等待，适合高并发场景；
• 获取锁后再次检查连接状态，避免在等待锁的间隙中其他进程已经创建了连接；
• 使用FINALLY块确保锁一定会被释放，杜绝死锁风险。

三、额外注意事项 #

• 连接资源限制：持久连接会占用目标数据库的连接资源，要注意目标库的max_connections配置，避免连接耗尽；
• 触发器性能：高插入量场景下，触发器本身会带来一定性能开销，可考虑结合pg_notify+后台进程实现异步同步，或者改用批量插入来降低触发频率；
• 密码安全：代码中直接明文写密码存在安全风险，建议改用pgpass文件或者连接服务名（pg_service.conf）来管理连接参数。

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