如何在PostgreSQL主存储过程中异步并行执行子存储过程

我当前正在基于AWS RDS托管的PostgreSQL数据库构建数据管道，现有一个由cron触发的主存储过程main_procedure，代码如下：

CREATE OR REPLACE PROCEDURE <postgres_schema>.main_procedure()
LANGUAGE plpgsql
AS $procedure$
begin
    
call <postgres_schema>.procedure_one();
call <postgres_schema>.procedure_two();
call <postgres_schema>.procedure_three();

END;
$procedure$
;

想咨询是否可以异步或并行执行后两个存储过程procedure_two和procedure_three，实现主存储过程先执行完procedure_one，再同时启动后两个子过程在后台运行的效果。

可行实现方案 #

PostgreSQL本身不支持在存储过程内直接异步调用其他存储过程，但可以通过以下几种方式实现需求：

1. 利用pg_cron动态调度异步任务 #

由于主过程已由pg_cron触发，可在procedure_one执行完成后，通过pg_cron的schedule函数立即调度后两个存储过程并行执行：

CREATE OR REPLACE PROCEDURE <postgres_schema>.main_procedure()
LANGUAGE plpgsql
AS $procedure$
declare
    current_time timestamp := now();
begin
    -- 同步执行procedure_one
    call <postgres_schema>.procedure_one();
    
    -- 立即调度procedure_two执行
    perform cron.schedule(
        'run-procedure-two-' || current_time,  -- 唯一任务名，避免冲突
        current_time::text,  -- 指定当前时间为执行时间，实现立即运行
        'call <postgres_schema>.procedure_two();'
    );
    
    -- 立即调度procedure_three执行
    perform cron.schedule(
        'run-procedure-three-' || current_time,
        current_time::text,
        'call <postgres_schema>.procedure_three();'
    );
    
    -- 可选：执行后删除临时调度任务，避免任务列表冗余
    perform cron.unschedule('run-procedure-two-' || current_time);
    perform cron.unschedule('run-procedure-three-' || current_time);
END;
$procedure$
;

注意：AWS RDS PostgreSQL默认支持pg_cron，但需确保实例已启用该扩展——通过参数组将shared_preload_libraries添加pg_cron后重启实例即可。

2. 使用dblink发起异步连接调用 #

通过dblink扩展建立本地数据库的异步连接，在后台独立执行存储过程：
首先确保dblink扩展已安装：

CREATE EXTENSION IF NOT EXISTS dblink;

然后修改主存储过程：

CREATE OR REPLACE PROCEDURE <postgres_schema>.main_procedure()
LANGUAGE plpgsql
AS $procedure$
begin
    -- 同步执行procedure_one
    call <postgres_schema>.procedure_one();
    
    -- 异步调用procedure_two
    perform dblink_connect('dbname=' || current_database());
    perform dblink_send_query(
        '',  -- 使用当前连接会话
        'call <postgres_schema>.procedure_two();'
    );
    perform dblink_disconnect();
    
    -- 异步调用procedure_three
    perform dblink_connect('dbname=' || current_database());
    perform dblink_send_query(
        '',
        'call <postgres_schema>.procedure_three();'
    );
    perform dblink_disconnect();
END;
$procedure$
;

这种方式会启动新的数据库会话在后台执行存储过程，主过程无需等待其完成即可结束。

3. 基于pg_notify的监听触发模式 #

创建后台监听进程，监听特定通知通道，主过程通过发送通知触发后两个存储过程并行执行：

• 先创建持续监听的过程：

CREATE OR REPLACE PROCEDURE <postgres_schema>.listen_for_tasks()
LANGUAGE plpgsql
AS $procedure$
declare
    rec record;
begin
    loop
        listen 'task_channel';
        fetch pg_notify into rec;
        if rec.channel = 'task_channel' then
            case rec.payload
                when 'run_procedure_two' then call <postgres_schema>.procedure_two();
                when 'run_procedure_three' then call <postgres_schema>.procedure_three();
            end case;
        end if;
    end loop;
END;
$procedure$
;

• 用pg_cron启动该监听过程，确保它持续运行：

select cron.schedule('listen-tasks', '* * * * *', 'call <postgres_schema>.listen_for_tasks();');

• 最后修改主过程发送触发通知：

CREATE OR REPLACE PROCEDURE <postgres_schema>.main_procedure()
LANGUAGE plpgsql
AS $procedure$
begin
    -- 同步执行procedure_one
    call <postgres_schema>.procedure_one();
    
    -- 发送通知触发procedure_two
    perform pg_notify('task_channel', 'run_procedure_two');
    
    -- 发送通知触发procedure_three
    perform pg_notify('task_channel', 'run_procedure_three');
END;
$procedure$
;

关键注意事项 #

• 确保procedure_two和procedure_three是无状态或线程安全的，避免并发执行时出现数据竞争、锁冲突等问题。
• AWS RDS环境中需确认对应扩展（pg_cron、dblink）已启用，且执行用户拥有足够的权限（如cron调度权限、dblink连接权限）。
• 异步执行的存储过程错误不会直接反馈到主过程，需单独实现日志记录或监控机制（比如在存储过程中写入执行状态到日志表）。

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