如何在另一个作业之后运行 Postgres pg_cron 作业？

Question

我晚上使用 pg_cron 扩展在我的 postgres 数据库上运行一些自动化任务。我正在将某些旧记录移动到存档数据库表中。我在 5 个不同的后台工作人员同时运行 5 个存储过程，因此它们都同时启动并在不同的工作人员上运行（我假设这类似于在 Java 中的不同线程上运行不同的任务） .这 5 个存储过程是独立的（将记录移动到归档表中），因此它们可以同时运行。我使用类似的命令安排他们每个人

cron.schedule (myJob1,
    '* * * * *',
    'call my_stored_proc_1()'
);

cron.schedule (myJob2,
    '* * * * *',
    'call my_stored_proc_2()'
);

.
..
...

cron.schedule (myJob5,
    '* * * * *',
    'call my_stored_proc_5()'
);

现在，我想运行一些更多 依赖存储过程。但是他们需要在这 5 个作业完成/完成后运行，因为他们正在执行一些 DELETE... sql 操作。

如何让第二个存储过程（执行 DELETE 查询的那个）作业在我的前 5 个存储过程作业完成后运行？我不想设置 CRON 表达式对于第二个存储过程执行 DELETES，因为我不知道前 5 个存储过程什么时候完成...

下面我提供了一个关于当前如何触发作业以及我希望它如何工作的示意图（如果可能的话）：

Answer 1

前言：我如何理解问题

我希望我理解 OP 描述的问题。

如果我错了，那么下面的所有内容都会无效。

我想这是关于 CPU 和/或 IO 繁重的周期性夜间任务。

例如：

• 有用于归档数据的任务 A-C
• 也许任务 D-E 用于重建聚合/刷新垫视图
• 最后是在整个 DB 上运行重新索引/分析的任务 F

因此，仅在任务 A-E 完成后运行任务 F 才有意义。

每个任务在一段时间内只需要运行一次：

• 一天、一小时或一周一次，或者仅在周末的晚上一次
• 最好不要在服务器负载过重的时候运行

是否符合 OP 要求 - IDK。

为简单起见，我们假设每个任务每晚只运行一次。很容易扩展到其他时期/要求。

数据驱动方法

1。添加日志表

例如

CREATE TABLE job_log (
  log_id bigint,
  job_name text,
  log_date timestamptz
) 

任务 A-E

开始时

为每个工作职能做检查：

IF  EXISTS(
  SELECT 1 FROM job_log 
    WHERE
      job_name = 'TaskA' # TaskB-TaskE for each functiont
      AND log_date::DATE = NOW()::DATE # check that function already executed  this night
) OR  EXISTS(
   SELECT 1 FROM pg_stat_activity 
     WHERE 
       query like 'SELECT * FROM jobA_function();'  # check that job not executing right now
) THEN RETURN;
END IF;

可能还可以添加其他条件：查找连接数、是否存在锁等。

这样可以保证函数执行的频率不会超过需要的频率。

完成

INSERT INTO job_log
SELECT
   (SELECT MAX(log_id) FROM job_log) + 1 # or use sequences/other autoincrements
  ,'TaskA'
  ,NOW()

Cronjob 计划

意思变了。

现在是：“尝试启动任务的执行”。

可以安全地在所选时段之间的每个小时安排一次，甚至更频繁。

Cronjob 无法知道服务器是否处于负载状态，表上是否有锁，或者可能有人手动开始执行任务。

工作职能在这方面可能更聪明。

任务 F

与上述相同，但在开始时检查是否完成其他任务。

例如

IF NOT EXISTS(
  SELECT 1 FROM job_log 
     WHERE 
       job_name = 'TaskA'
       AND log_date::DATE = NOW()::DATE
) OR NOT EXISTS(  
  SELECT 1 FROM job_log 
     WHERE 
       job_name = 'TaskB'  
       AND log_date::DATE = NOW()::DATE
)
....  # checks for completions of other tasks
OR EXISTS(
  SELECT 1 FROM job_log 
    WHERE
      job_name = 'TaskF' # TaskB-TaskE for each functiont
      AND log_date::DATE = NOW()::DATE # check that function already executed  this night
) OR  EXISTS(
   SELECT 1 FROM pg_stat_activity 
     WHERE 
       query like 'SELECT * FROM jobF_function();'  # check that job not executing right now
) THEN RETURN;

完成时

与其他函数一样写入job_log。

更新。 Cronjob 时间表

在 cronjob 中创建多个计划。

例如

假设 A-E 任务将运行大约 10-15 分钟。

其中一两个可能工作 30-45-60 分钟。

为任务 F 创建一个计划以尝试每 5 分钟启动一次。

这将如何运作：

• 尝试 1：任务 A 已完成，其他仍在工作 -> 退出
• 尝试 2：任务 A-C 完成 -> 退出
• 尝试 3：任务 A-E 完成 -> 启动任务 F
• 尝试 4：任务 A-E 已完成，但在 pg_stat_activity 中有一个正在执行的任务 F -> 退出
• 尝试 5：任务 A-E 已完成，pg_stat_activity 为空，但在日志中我们看到任务 F 已执行 -> 无需工作 -> 退出
• ...所有其他尝试将保持不变，直到第二天晚上

总结

很容易扩展这种方法以满足任何要求：

• 另一个周期性
• 或使其完全不定期。例如。用触发器制作一个表并在更改时开始执行
• 任何深度的依赖和/或“模糊”依赖
• ...几乎所有内容

概念保持不变：

• cronjob 计划意味着“尝试运行”
• 是否运行的决定是数据驱动的

我很高兴听到任何形式的批评 - 谁知道我可能忽略了一些东西。

Answer 2

您可以使用pg_stat_activity view 来确保没有像您的工作 1-5 这样的活动查询。

注意：

超级用户和内置角色pg_read_all_stats（另见第 21.5 节）的成员可以查看所有会话的所有信息

...
while (
    select count(*) > 0 
    from pg_stat_activity 
    where query in ('call my_stored_proc_1()', 'call my_stored_proc_2()', ...))
loop
    perform pg_sleep(1);
    perform pg_stat_clear_snapshot(); -- needs to retrieve the fresh data
end loop;
...

只需将此代码插入您的 stored proc 6 的开头并在作业 1-5 之后调用它几秒钟。

注1：

可以使用正则表达式简化和概括条件：

when query ~ 'my_stored_proc_1|my_stored_proc_2|...'

注2：

您可以使用clock_timestamp() 函数实现超时：

...
is_timedout := false;
timeout := '10 min'::interval; -- stop waiting after 10 minutes
start_time := clock_timestamp();
while (...)
loop
    perform pg_sleep(1);
    perform pg_stat_clear_snapshot(); -- needs to retrieve the fresh data
    if clock_timestamp() - start_time > timeout then
        is_timedout := true;
        break;
    end if;
end loop;

if is_timedout then
    ...
else
    ...
end if;
...

注3：

查看pg_stat_activity 的其他列。您可能也需要使用它们。