@ agendajs/postgres-后端

PostgreSQL 的后端 议程项目 在实时工作处理中,有 optEN/NOTIFY 支持的工作调度器.

特征

• 议事日程任务的全部 PostgreSQL 存储后端
• 使用 PostgreSQL 倾听/NOTIFY 的实时工作通知
• 具有优化索引的自动创建计划
• 连接集合通过 pg 图书馆
• 带有完整类型定义的脚本类型支持

安装

npm install @agendajs/postgres-backend
# or
pnpm add @agendajs/postgres-backend
# or
yarn add @agendajs/postgres-backend

使用量

基本用途

import { Agenda } from 'agenda';
import { PostgresBackend } from '@agendajs/postgres-backend';

// Create agenda with PostgreSQL backend
const agenda = new Agenda({
  backend: new PostgresBackend({
    connectionString: 'postgresql://user:pass@localhost:5432/mydb'
  })
});

// Define jobs
agenda.define('send-email', async (job) => {
  const { to, subject, body } = job.attrs.data;
  await sendEmail(to, subject, body);
});

// Start processing
await agenda.start();

// Schedule jobs
await agenda.every('5 minutes', 'send-email', { to: '[email protected]', subject: 'Hello', body: 'World' });
await agenda.schedule('in 10 minutes', 'send-email', { to: '[email protected]', subject: 'Hi', body: 'There' });

配置选项

import { PostgresBackend } from '@agendajs/postgres-backend';

const backend = new PostgresBackend({
  // PostgreSQL connection string (required unless pool/poolConfig is provided)
  connectionString: 'postgresql://user:pass@localhost:5432/mydb',

  // Or use pool configuration (creates a new pool)
  poolConfig: {
    host: 'localhost',
    port: 5432,
    database: 'mydb',
    user: 'user',
    password: 'pass',
    max: 20 // max pool size
  },

  // Table name for jobs (default: 'agenda_jobs')
  tableName: 'agenda_jobs',

  // Channel name for LISTEN/NOTIFY (default: 'agenda_jobs')
  channelName: 'agenda_jobs',

  // Name to identify this Agenda instance (stored as lastModifiedBy)
  name: 'worker-1',

  // Whether to create the table and indexes on connect (default: true)
  ensureSchema: true,

  // Sort order for job queries
  sort: {
    nextRunAt: 'asc',
    priority: 'desc'
  }
});

使用现有池

如果您的应用程序已有 PostgreSQL 连接池, 您可以直接通过 。 游泳池会 没有 当议程断开时关闭 :

import { Pool } from 'pg';
import { PostgresBackend } from '@agendajs/postgres-backend';

// Your app's existing pool
const pool = new Pool({ connectionString: 'postgresql://...' });

const backend = new PostgresBackend({ pool });

// After agenda.stop(), your pool is still usable

如何运作

存储

后端将工作存储在 PostgreSQL 表格中,其结构如下:

• id - UUID主要钥匙
• name - 职称
• priority - 优先工作(较高=更紧急)
• next_run_at - 当工作应该运行下一个
• type - 工作类型(“ 正常” 或“ 单一” )
• locked_at - 执行分配的锁定时间戳
• data - JSONB工作数据的有效载荷
• 其他元数据字段...

实时通知

与MongoDB不同,PostgreSQL通过Listen/NOTIFY拥有内置的酒吧/子. 保存工作时,会向所有倾听《议程》的事例发布通知,立即启动工作处理,而不等待下一次投票间隔.

这意味着:

• 执行职务的延迟度较低
• 减少数据库投票费用
• 二. 有效的跨进程协调

分布式锁定

后端使用 PostgreSQL FOR UPDATE SKIP LOCKED 以解剖方式锁定工作进行处理,防止在多个《议程》中重复执行.

数据库索引

何时 ensureSchema: true (默认),创建下列索引:

-- Main index for finding and locking next job
CREATE INDEX agenda_jobs_find_and_lock_idx
  ON agenda_jobs (name, next_run_at, priority DESC, locked_at, disabled)
  WHERE disabled = FALSE;

-- Index for single jobs (upsert operations)
CREATE UNIQUE INDEX agenda_jobs_single_job_idx
  ON agenda_jobs (name) WHERE type = 'single';

-- Index for locked_at (stale lock detection)
CREATE INDEX agenda_jobs_locked_at_idx
  ON agenda_jobs (locked_at) WHERE locked_at IS NOT NULL;

-- Index for next_run_at queries
CREATE INDEX agenda_jobs_next_run_at_idx
  ON agenda_jobs (next_run_at) WHERE next_run_at IS NOT NULL;

任务日志

PostgresBackend 包含内置 PostgresJobLogger 在专门表格中存储结构化工作生命周期事件(agenda_logs (默认情况下). 伐木机是轻量级的,只在第一次使用时创建出它的桌子.

自动使用

当您在议事日程中启用日志时, 后端的内置登录器会被自动使用 :

const agenda = new Agenda({
  backend: new PostgresBackend({ connectionString: 'postgresql://...' }),
  logging: true  // Uses PostgresJobLogger automatically
});

独立使用

您还可以使用 PostgresJobLogger 独立——例如,在使用不同的后端存储时登录到 PostgreSQL :

import { PostgresJobLogger } from '@agendajs/postgres-backend';
import { Pool } from 'pg';

const logger = new PostgresJobLogger({
  pool: new Pool({ connectionString: 'postgresql://...' }),
  logTableName: 'agenda_logs'       // Optional: table name (default: 'agenda_logs')
});

// Use with any backend
import { MongoBackend } from '@agendajs/mongo-backend';
const agenda = new Agenda({
  backend: new MongoBackend({ mongo: db }),
  logging: logger
});

见 议程项目研究 用于完整记录文件.

手册Schema 管理

如果您愿意自己管理这个方案( 设置 ) ensureSchema: false:

import { getCreateTableSQL, getCreateIndexesSQL } from '@agendajs/postgres-backend';

const tableName = 'agenda_jobs';

// Create table
await pool.query(getCreateTableSQL(tableName));

// Create indexes
for (const sql of getCreateIndexesSQL(tableName)) {
  await pool.query(sql);
}

测试

测试需要一个 PostgreSQL 数据库。 最简单的方法是使用多克:

# Start PostgreSQL container and run tests
pnpm test:docker

# Or manually:
pnpm docker:up        # Start PostgreSQL container
pnpm test:postgres    # Run tests
pnpm docker:down      # Stop container

您也可以使用现有的 PostgreSQL 数据库 :

POSTGRES_TEST_URL=postgresql://user:pass@localhost:5432/agenda_test pnpm test

可用的脚本

所需资源

• 节点.js 18.0.0
• PostgreSQL 12 (用于 gen_random_uuid() 支助)
• 议程项目6.0.0

许可证

麻省理工学院