PostgreSQL函数内循环读取pg_stat_activity数据重复如何解决

PostgreSQL的单个事务内会复用启动时生成的MVCC快照，你编写的PL/pgSQL函数运行在调用它的父事务中，所以整个循环执行期间读取的pg_stat_activity系统视图都是事务启动时的快照数据，同时now()函数返回的也是事务启动时间，才会出现所有采集数据都和首秒一致的问题。

下面提供3种适配PostgreSQL 13的可行方案，优先推荐使用存储过程方案： #

方案1：改用支持事务提交的存储过程（推荐） #

PostgreSQL 11及以上版本支持的存储过程（PROCEDURE）允许在代码内部显式提交事务，每次提交后下一次循环会启动新事务、获取最新快照，完全满足你的需求。

修改后代码 #

CREATE OR REPLACE PROCEDURE public.sp_activity(
    IN p_collect_count integer
)
LANGUAGE plpgsql
AS $BODY$
DECLARE
    counter integer := 0;
BEGIN
    WHILE counter < p_collect_count LOOP
        RAISE NOTICE '当前采集次数: %', counter;
        counter := counter + 1;
        
        INSERT INTO public.pg_stat_activity_log(
            "time",datid,datname,pid,leader_pid,usesysid,usename,application_name,client_addr,client_hostname,
            client_port,backend_start,xact_start,query_start,state_change,wait_event_type,wait_event,state,backend_xid,backend_xmin,query,backend_type
        )
        SELECT 
            clock_timestamp()::time(0), -- 用clock_timestamp获取当前真实时间，替代事务级固定的now()
            datid,datname,pid,leader_pid,usesysid,usename,application_name,client_addr,client_hostname,client_port,
            backend_start,xact_start,query_start,state_change,wait_event_type,wait_event,state,backend_xid,backend_xmin,query,backend_type
        FROM pg_stat_activity;
        
        COMMIT; -- 显式提交当前事务，下一次循环使用新的事务快照
        
        RAISE NOTICE '第%次采集完成', counter;
        PERFORM pg_sleep(1);
    END LOOP;
END;
$BODY$;

调用方式 #

TRUNCATE TABLE public.pg_stat_activity_log;
CALL public.sp_activity(60); -- 采集60次

方案2：用dblink模拟自治事务 #

如果必须保留函数调用的方式，可以使用dblink扩展通过本地独立连接查询pg_stat_activity，每次查询都是独立会话、独立事务，能拿到最新数据。

操作步骤 #

1. 安装dblink扩展

CREATE EXTENSION IF NOT EXISTS dblink;

2. 修改函数代码

CREATE OR REPLACE FUNCTION public.fn_activity(
    integer
) RETURNS void 
LANGUAGE plpgsql
AS $BODY$
DECLARE 
    counter integer := 0;
    v_local_conn text := 'dbname='||current_database(); -- 本地数据库连接串
BEGIN
    WHILE counter < $1 LOOP
        RAISE NOTICE '当前采集次数: %', counter;
        counter := counter + 1;
        
        INSERT INTO public.pg_stat_activity_log(
            "time",datid,datname,pid,leader_pid,usesysid,usename,application_name,client_addr,client_hostname,
            client_port,backend_start,xact_start,query_start,state_change,wait_event_type,wait_event,state,backend_xid,backend_xmin,query,backend_type
        )
        SELECT * FROM dblink(v_local_conn,
            'SELECT clock_timestamp()::time(0),datid,datname,pid,leader_pid,usesysid,usename,application_name,client_addr,client_hostname,client_port,
            backend_start,xact_start,query_start,state_change,wait_event_type,wait_event,state,backend_xid,backend_xmin,query,backend_type
            FROM pg_stat_activity'
        ) AS t(
            "time" time,datid oid,datname name,pid integer,leader_pid integer,usesysid oid,usename name,application_name text,client_addr inet,client_hostname text,
            client_port integer,backend_start timestamptz,xact_start timestamptz,query_start timestamptz,state_change timestamptz,
            wait_event_type text,wait_event text,state text,backend_xid xid,backend_xmin xid,query text,backend_type text
        );
        
        RAISE NOTICE '第%次采集完成', counter;
        PERFORM pg_sleep(1);
    END LOOP;
    RETURN;
END;
$BODY$;

方案3：外部脚本调度采集 #

如果不想修改数据库内部代码，可以使用外部脚本（Shell、Python等）每秒发起一次独立数据库连接执行插入操作，每次连接都是独立事务，天然避免快照复用问题。

Shell脚本示例 #

#!/bin/bash
# 替换为实际的数据库连接信息
DB_USER="你的用户名"
DB_NAME="你的数据库名"

psql -U ${DB_USER} -d ${DB_NAME} -c "TRUNCATE TABLE public.pg_stat_activity_log;"
for i in {1..60}
do
  psql -U ${DB_USER} -d ${DB_NAME} -c "INSERT INTO public.pg_stat_activity_log(\"time\",datid,datname,pid,leader_pid,usesysid,usename,application_name,client_addr,client_hostname,client_port,backend_start,xact_start,query_start,state_change,wait_event_type,wait_event,state,backend_xid,backend_xmin,query,backend_type) SELECT clock_timestamp()::time(0),datid,datname,pid,leader_pid,usesysid,usename,application_name,client_addr,client_hostname,client_port,backend_start,xact_start,query_start,state_change,wait_event_type,wait_event,state,backend_xid,backend_xmin,query,backend_type FROM pg_stat_activity;"
  sleep 1
done

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