如何在 PostgreSQL 中使用 pg_cron 来安排作业的执行

之后你编写的SQL代码在所有地方都是相同的，而这正是它的魅力所在。

Cron语法的快速回顾

pg_cron使用与经典Unix cron相同的五字段调度格式：

┌──────────── 分钟（0–59）
│ ┌────────── 小时（0–23）
│ │ ┌──────── 月份中的日期（1–31，或用$表示最后一天）
│ │ │ ┌────── 月份（1–12）
│ │ │ │ ┌──── 星期几（0–6，周日=0）
│ │ │ │ │
* * * * *

星号表示“所有值”。你可以用逗号组合数值，用连字符表示范围，用斜杠表示间隔时间。有一些调度规则你会经常用到：

*/5 * * * *    每5分钟执行一次
0 * * * *      每小时整点执行
0 3 * * *      每天凌晨3点执行
0 3 * * 1-5    工作日每天凌晨3点执行
30 1 * * 0     每个周日凌晨1:30执行
0 0 1 * *      每个月1号午夜执行

pg_cron还在标准语法中添加了两个常规cron所没有的功能。

你可以在“月份中的日期”字段中使用$来表示该月的最后一天，而在标准cron中表达这一点其实相当麻烦：

0 23 $ * *     每个月最后一天晚上11点执行

而对于那些需要每分钟多次执行的作业，你可以使用1到59秒之间的时间间隔来设置调度规则：

crontab.guru可以将Cron表达式翻译成通俗易懂的英语。需要注意的是，pg_cron会按照cron.timezone设置的时区来解析调度规则，默认时区为GMT。

如何安排你的第一个作业
核心函数是cron.schedule()。它有两种形式：一种带有名称，另一种没有名称。
建议使用带名称的形式，因为这样可以让作业更容易被查找、更新或删除：
SELECT cron_schedule(
  'delete-expired-sessions',                          -- 作业名称
  '0 3 * * *',                                        -- 调度规则
  $$DELETE FROM sessions WHERE expires_at < now()$$   -- 命令
);

该函数会返回作业的ID：
 schedule
----------
        1
(1 row)

还有一些需要注意的地方：
命令需要用$$ ... $$括起来，这是PostgreSQL特有的引用方式。这样就不需要在SQL代码中为单引号进行转义处理了。对于不需要引号的命令，使用普通的字符串字面量即可。
作业会在调用cron_schedule()的数据库中运行，而且会以调用该函数的用户的权限来执行任务。调度系统中不存在任何提升权限的功能——如果你的用户没有权限从sessions表中删除数据，那么作业也同样无法执行这个操作。
如果你再次使用相同的作业名称调用cron_schedule()，pg_cron会更新现有的作业而不是创建重复项。这样一来，调度任务就具备了幂等性，这对于在数据库迁移过程中定义作业来说非常方便。
实用的pg_cron示例
让我们来看一些能够涵盖大多数实际应用场景的例子。这些示例都可以直接被你借鉴和使用。
示例1：每晚清除旧数据行
那些用于存储临时数据的表格——如会话信息、令牌、审计记录、通知日志等——如果不进行清理，这些数据就会不断累积。因此，每晚删除这些旧数据是一项非常常见的操作，这也是pg_cron的首个常用任务：
SELECT cron.schedule(
  'purge-old-events',
  '0 2 * * *',
  $$DELETE FROM events WHERE created_at < now() - interval '90 days'$$
);

每晚凌晨2点，所有创建时间超过90天的数据行都会被删除。如果表格的数据量非常大，可以考虑将删除操作封装在一个函数中，这样每次执行该函数的耗时就会较短，然后再安排这个函数定期运行。
示例2：每小时刷新一次物化视图
物化视图是一种用于缓存复杂聚合数据的优秀工具，但PostgreSQL本身并不会自动更新这些视图。pg_cron可以解决这个问题：
SELECT cron.schedule(
  'refresh-daily-sales',
  '5 * * * *',
  'REFRESH MATERIALIZED VIEW CONCURRENTLY daily_sales_summary'
);

这个示例会每小时5分更新一次daily_sales_summary物化视图。由于使用了CONCURRENTLY选项，因此即使在更新过程中也有数据被读取，只要该视图具有唯一的索引，这种操作就不会出现问题。
示例3：生成每日汇总表
另一种常见的用法是生成每日汇总表：而不是在每次查看仪表盘时都重新计算数百万条原始数据，可以每天预先计算一次这些数值。
SELECT cron.schedule(
  'rollup-daily-orders',
  '15 0 * * *',
  $$
  INSERT INTO daily_order_stats (day, order_count, total_amount)
  SELECT created_at::date, count(*), sum(amount)
  FROM orders
  WHERE created_at >= current_date - 1
    AND created_at < current_date
  GROUP BY created_at::date
  ON CONFLICT (day) DO UPDATE
    SET order_count = EXCLUDED.order_count,
        total_amount = EXcluded.total_amount
  $$
);

每晚凌晨15分，昨天所有的订单信息会被汇总到一张表格中。ON CONFLICT子句确保了这个任务可以安全地重复执行——即使它被执行了两次，也会只覆盖原有的数据而不会产生重复记录。
示例4：每30秒运行一次任务
有些操作需要更频繁地执行，而cron默认的1分钟间隔显然无法满足这种需求。例如，需要定期清空缓冲表、从待处理队列中取出数据、推进轻量级任务的执行等。pg_cron支持每30秒执行一次这样的任务：
SELECT cron.schedule(
  'process-outbox',
  '30 seconds',
  'CALL process_outbox_batch()'
);

请记住之前提到的保证：当某个作业还在运行时，pg_cron不会启动该作业的第二个实例。如果某个批处理任务偶尔需要45秒才能完成，那么下一次执行时会等待轮到自己再开始运行，而不会抢在前面执行。
示例5：在每月的最后一天执行维护任务
在标准的cron系统中，处理月末任务会遇到一些麻烦，因为不同月份的天数长度各不相同。而pg_cron的$语法让这种问题变得非常简单：
SELECT cron.schedule(
  'month-end-vacuum',
  '0 23 $ * *',
  'VACUUM ANALYZE orders'
);

这条命令会在每个月的28日、29日、30日或31日晚上11点，对orders表执行VACUUM ANALYZE操作——也就是在那个月的最后一天执行这个任务。
如何查看和监控你的作业
pg_cron所记录的所有信息都存储在cron模式下的两个表中，你可以像查询其他表格一样来查询这些表。
如果你想查看已经安排好的作业任务，可以查询cron.job表：
SELECT jobid, jobname, schedule, command, active
FROM cron.job;

 jobid |         jobname         |  schedule  |            command             | active
-------+-------------------------+------------+--------------------------------+--------
     1 | delete-expired-sessions | 0 3 * * *  | DELETE FROM sessions WHERE ... | t
     2 | refresh-daily-sales     | 5 * * * *  | REFRESH MATERIALIZED VIEW ...  | t
(2 rows)

如果你想了解这些作业实际是如何运行的，可以查询cron.job_run_details表：
SELECT jobid, status, return_message, start_time, end_time
FROM cron.job_run_details
ORDER BY start_time DESC
LIMIT 10;

每一行记录了一次作业的执行情况：包括作业是成功完成还是失败了、返回了什么信息，以及作业开始和结束的具体时间。如果作业失败了，其status字段会显示为‘failed’，同时还会显示错误信息，因此调试工作通常都是从这个表开始的。
需要注意的一点是：pg_cron本身永远不会自动清理这个表。如果某个作业每30秒执行一次，那么每天就会产生将近三千条记录。解决这个问题的方法也很简单——再安排一个pg_cron作业来清除这些历史记录：
SELECT cron.schedule(
  'purge-cron-history',
  '0 12 * * *',
  $$DELETE FROM cron.job_run_details
    WHERE end_time < now() - interval '14 days'$$
);

如果你根本不希望记录作业的执行历史，可以在配置中将cron.log_run = off这个参数设置成关闭状态。
如何更新和删除作业
如果要修改现有的作业任务，可以使用cron.alter_job()函数，并传入该作业的ID。只有你传递的参数会被修改，其他所有设置都会保持不变：
-- 将作业1的执行时间从凌晨3点改为4点
SELECT cron.alter_job(1, schedule := '0 4 * * *');

-- 暂停某个作业而不删除它
SELECT cron.alter_job(1, active := false);

-- 后来再恢复该作业的运行
SELECT cron.alter_job(1, active := true);

使用active := false来暂停某个作业，这一做法其实被低估了。在发生故障或进行大规模数据迁移时，你可以暂时停止某个运行中的作业，等任务完成后再重新启动它，而该作业的配置信息并不会因此丢失。
如果要永久删除某个作业，可以使用cron.unschedule()命令，只需提供作业的名称或ID即可：
SELECT cron.un_schedule('delete-expired-sessions');
-- 或者
SELECT cron.un.schedule(1);

无论使用哪种方式，当命令执行成功并删除了相应的作业时，都会返回true值。
如何在其他数据库中运行作业
需要记住的是，pg_cron在每个数据库实例中只会被安装一次，通常安装在postgres数据库中。如果你的数据库实例托管了多个数据库，你并不需要在每个数据库中都安装pg_cron——你可以只在一个数据库中配置作业调度任务，然后使用cron.schedule_in_database()命令在其他数据库中执行这些作业：
SELECT cron.schedule_in_database(
  'analytics-nightly-vacuum',
  '0 4 * * *',
  'VACUUM ANALYZE page_views',
  'analytics_db'
);

这些作业虽然是在中央数据库中配置的，但实际会在analytics_db数据库中执行。此外，这个函数还允许用户指定用户名（如果希望作业以其他用户的身份运行），以及设置active参数来控制作业是处于暂停状态还是正常运行。
这种设计方式使得所有的调度任务都集中在一个数据库的同一个模式中，因此进行审计非常方便：只需执行一条SELECT * FROM cron.job命令，就能查看整个实例中所有已安排的作业。
如何允许其他用户安排作业
默认情况下，只有超级用户才能使用调度相关的功能。如果你想让某个应用角色能够自行管理它的作业，就需要授予它对cron模式的使用权：
GRANT USAGE ON SCHEMA cron TO app_user;

这种权限设置既合理又安全：


作业是按照安排它们的用户的权限来执行的，不会超出这些权限范围。


由于cron.job表采用了行级安全机制，因此用户只能看到和修改自己安排的作业；超级用户则可以查看所有作业。


每个用户还可以自行删除cron.job_run_details表中与自己相关的记录，这样一来，即使没有超级用户权限，也能正常执行清理任务。



在实际使用中，我建议专门为作业调度创建一个专用角色，而不是让个人账户来负责这些任务。因为如果负责安排作业的工程师离职了，他们的角色也会被删除，而此时就不应该让之前的作业安排继续生效。
何时使用pg_cron，何时避免使用它
当需要执行与数据库相关的操作时，pg_cron会显得非常有用。以下是一些适合使用它的场景：


数据维护：清理会话记录、日志文件、事件信息以及令牌表中的过期数据。


数据聚合：更新物化视图并生成汇总报表。


系统维护：执行有针对性的VACUUM ANALYZE操作，重新计算统计信息，并管理数据库分区（这与pg_partman配合使用效果最佳）。



简单的数据处理流程：在不同的表之间迁移数据，处理待发送的消息，以及删除那些被标记为“软删除”的记录。


共同点在于：所有的任务都可以用SQL语句或存储过程来表示，而且这些任务不会涉及数据库之外的任何内容。
在以下情况下，你应该考虑使用其他工具：


如果任务需要调用外部系统。 pg_cron只能执行SQL语句，它无法发送HTTP请求、加入队列或发送电子邮件。这类任务应该由你的应用程序或工作流引擎来处理。


如果任务需要具备重试机制、延迟执行功能以及警报提醒。 pg_cron虽然会记录失败情况，但不会自动重试或通知用户。对于那些必须完成的任务来说，使用Temporal这样的工具或专门的工作流管理系统才是更合适的选择。


如果任务耗时较长、计算量较大。在主OLTP数据库中运行长达四小时的批处理任务会占用大量的CPU资源、内存以及锁资源。因此，应该将这类任务安排在其他地方执行。


如果任务之间存在复杂的依赖关系。

 “只有当A任务成功完成后才能执行B任务，然后再依次执行C和D任务”这种逻辑属于工作流管理的范畴，而不是pg_cron的适用范围。


一个合理的经验法则是：pg_cron适用于替代那些在某些被遗忘的服务器上通过crontab来执行的、仅用于运行`psql -c "..."`命令的任务。但它不能替代你的工作流队列或工作流管理系统。
使用pg_cron的最佳实践
养成以下一些习惯，可以让你的定时任务更加高效、易于管理：
为每个任务起一个名称：仅通过ID来标识的匿名任务在六个月后会变得难以管理。给任务起名字还可以确保`cron.schedule()`函数的幂等性，这样你在进行数据库迁移时就可以安全地定义这些任务。
要明确设置时区：pg_cron的默认时区是GMT，如果你不想遇到“凌晨3点的任务实际上在凌晨4点才执行”这样的问题，可以在一开始就通过`cron.timezone`参数来指定正确的时区。
缩短每次任务的执行时间：将大规模的数据删除操作封装到批处理存储过程中去。那些能在几秒钟内完成的任务会占用较少的系统资源，也不会在系统中留下过多的等待队列。
确保任务具有幂等性：服务器可能会重启，任务也可能在执行中途失败。使用`ON CONFLICT`语句、时间窗口条件等机制，可以确保重新执行任务时不会产生问题。
定期清理不必要的任务：在表格的大小还没有大到足以引起麻烦之前，就通过调度清理任务来保持数据库的整洁。
不仅要关注任务是否失败，还要注意它们是否完全停止了执行：失败的任务会在`job_run_details`表中留下记录，但那些被彻底取消调度的任务则不会留下任何痕迹。定期检查每个关键任务是否最近有成功执行的记录，就可以及时发现这两种情况：
SELECT j.jobname, max(d.end_time) AS last_success
FROM cron.job j
LEFT JOIN cron.job_run_details d
  ON d.jobid = j.jobid AND d.status = 'succeeded'
GROUP BY j.jobname
HAVING max(d.end_time) < now() - interval '1 day'
   OR max(d.end_time) IS NULL;

这个查询返回的任何任务都已经超过一天没有成功执行了，因此值得我们关注。
结论
pg_cron将PostgreSQL转变成了一个独立的调度系统。你可以通过SQL语言定义这些任务，数据库会负责执行它们，而整个系统的运行情况——包括任务的定义、执行历史以及失败记录——都可以通过普通的查询语句来查看。
在本教程中，你了解了这个扩展模块的内部工作原理，学会了如何在自己的服务器或托管服务上安装它，掌握了如何编写调度任务（包括pg_cron提供的秒级调度功能以及每月最后一天的特殊处理机制），也知道了如何利用它来处理所有数据库都会遇到的维护工作，比如数据优化、汇总分析、数据更新以及磁盘清理等。此外，你还学会了如何监控这些任务的执行情况，如何管理相关权限，以及何时应该使用任务队列或专门的调度工具来更高效地完成任务。
如果你的基础设施中有一台专门用于通过crontab运行psql命令的服务器，那么现在你知道应该如何让这台服务器“退休”了。
感谢您的阅读！我主要撰写关于PostgreSQL和后端开发的文章。您可以通过LinkedIn或X与我保持联系。
Related Posts:
如何在 Node.js 中使用 Scalar 创建 REST API 文档
如何在 Go 语言和 PostgreSQL 中实现“待发邮件模式”
如何在C#中使用“OneOf”来实现类型安全的联合体
在 Flutter 中使用 IndexedStack 实现高效的状态管理
PostgresML：通过 PostgreSQL 集成简化 AI 模型部署
如何将 PostgreSQL 用作缓存、队列及搜索引擎