graphile/worker

graphile-worker 镜像文档

镜像概述

graphile-worker 是一个基于 Node.js 和 PostgreSQL 的高性能任务队列，允许在后台运行各类任务（如发送邮件、执行计算、生成PDF等），避免阻塞HTTP响应或应用代码。它可与任何PostgreSQL支持的应用配合使用，尤其适合与PostGraphile或PostgREST搭配。

核心功能和特性

• 独立与嵌入式模式：支持通过CLI独立运行或作为库集成到Node.js代码中
• 低延迟：使用PostgreSQL的LISTEN/NOTIFY机制，任务调度至执行延迟通常低于3ms
• 高性能：采用SKIP LOCKED机制获取任务，提升任务获取速度
• 默认并行执行：支持并发处理任务，可通过配置控制并发数
• 队列机制：同名队列中的任务串行执行，不同队列任务并行执行
• 自动重试：失败任务自动重试，采用指数退避策略
• 可配置重试次数：默认25次重试（约3天周期）
• 定时任务：支持类crontab语法的周期性任务调度（可选回填功能）
• 任务去重：通过唯一job_key实现任务去重
• 灵活的运行时控制：支持复杂的速率限制（可配合graphile-worker-rate-limiter）
• 开发便捷：支持开发模式下的任务文件监听（实验性），无需重启工作进程即可更新任务代码

使用场景和适用范围

• 后台任务处理：需要在后台异步执行的操作（如发送邮件、推送通知）
• 耗时操作：计算密集型任务（数据分析、报表生成）或IO密集型任务（文件处理、API调用）
• 定时任务：周期性执行的任务（数据备份、日志清理、统计报告）
• 任务去重：确保相同任务不会重复执行（如重复订单处理）
• 高可用部署：生产环境就绪，可与主应用部署在同一Node.js进程中降低运维复杂度

环境要求

• PostgreSQL 10+
• Node.js 10+
• 数据库需支持pg***扩展（如未安装，会自动在public模式下安装）

安装与部署

Docker 快速启动

bash
# 基本用法（使用环境变量指定数据库连接）
docker run -d \
  -e DATABASE_URL="postgres://user:password@postgres-host:5432/mydb?ssl=true" \
  -v ./tasks:/app/tasks \
  --name graphile-worker \
  graphile/worker:latest

# 仅初始化数据库 schema
docker run --rm \
  -e DATABASE_URL="postgres://user:password@postgres-host:5432/mydb" \
  graphile/worker:latest \
  --schema-only

# 单次运行模式（执行完所有可运行任务后退出）
docker run --rm \
  -e DATABASE_URL="postgres://user:password@postgres-host:5432/mydb" \
  -v ./tasks:/app/tasks \
  graphile/worker:latest \
  --once

Docker Compose 配置示例

yaml
version: '3.8'

services:
  postgres:
    image: postgres:13
    environment:
      POSTGRES_USER: worker_user
      POSTGRES_PASSWORD: worker_password
      POSTGRES_DB: worker_db
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U worker_user -d worker_db"]
      interval: 5s
      timeout: 5s
      retries: 5

  worker:
    image: graphile/worker:latest
    depends_on:
      postgres:
        condition: service_healthy
    environment:
      DATABASE_URL: "postgres://worker_user:worker_password@postgres:5432/worker_db"
      # 可选环境变量
      # CONCURRENCY: 5  # 并发任务数
      # POLL_INTERVAL: 1000  # 轮询间隔(ms)
    volumes:
      - ./tasks:/app/tasks  # 挂载任务目录
    restart: always

volumes:
  postgres_data:

详细使用方法和配置说明

任务定义

任务需定义在任务目录（默认tasks/）中的JS文件，文件名即为任务标识符。

示例任务文件（tasks/hello.js）：

javascript
// tasks/hello.js
module.exports = async (payload, helpers) => {
  const { name } = payload;
  helpers.logger.info(`Hello, ${name}`);
};

启动工作进程

通过CLI启动（Docker环境）

bash
# 基本启动（使用环境变量DATABASE_URL和默认任务目录/tasks）
docker run -e DATABASE_URL="postgres:///my_db" -v ./tasks:/app/tasks graphile/worker:latest

# 指定数据库连接字符串和并发数
docker run -v ./tasks:/app/tasks graphile/worker:latest -c "postgres://user:pass@host:port/db?ssl=true" -j 5

# 定时任务模式（指定crontab文件）
docker run -v ./tasks:/app/tasks -v ./crontab:/app/crontab graphile/worker:latest --crontab /app/crontab

通过代码集成（Node.js库模式）

javascript
const { run, quickAddJob } = require("graphile-worker");

async function main() {
  // 启动工作进程
  const runner = await run({
    connectionString: "postgres:///my_db",
    concurrency: 5,  // 并发数
    noHandleSignals: false,  // 处理系统信号（SIGINT, SIGTERM等）以优雅关闭
    pollInterval: 1000,  // 轮询间隔(ms)
    taskList: {
      hello: async (payload, helpers) => {
        const { name } = payload;
        helpers.logger.info(`Hello, ${name}`);
      },
    },
  });

  // 添加任务
  await quickAddJob(
    { connectionString: "postgres:///my_db" },
    "hello",  // 任务标识符
    { name: "Bobby Tables" }  // 任务参数
  );

  // 等待工作进程完成
  await runner.promise;
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});

添加任务

通过SQL添加任务

连接PostgreSQL数据库执行以下SQL：

sql
-- 添加任务
SELECT graphile_worker.add_job('hello', json_build_object('name', 'Bobby Tables'));

-- 添加定时任务（5分钟后执行）
SELECT graphile_worker.add_job(
  'hello', 
  json_build_object('name', 'Bobby Tables'),
  run_at => NOW() + INTERVAL '5 minutes'
);

通过代码添加任务

javascript
const { makeWorkerUtils } = require("graphile-worker");

async function addTask() {
  const workerUtils = await makeWorkerUtils({
    connectionString: "postgres:///my_db",
  });
  
  try {
    // 添加任务
    await workerUtils.addJob(
      "hello",  // 任务标识符
      { name: "Bobby Tables" },  // 任务参数
      { 
        queueName: "greetings",  // 队列名称（同名队列串行执行）
        priority: 10,  // 优先级（数值越小优先级越高）
        runAt: new Date(Date.now() + 5000),  // 5秒后执行
        maxAttempts: 5,  // 最大重试次数
        jobKey: "unique-hello-bobby"  // 任务去重键
      }
    );
  } finally {
    await workerUtils.release();
  }
}

addTask().catch(console.error);

配置参数

环境变量

CLI选项

Options:
      --help                    显示帮助信息
      --version                 显示版本号
  -c, --connection              数据库连接字符串，默认使用DATABASE_URL环境变量
  -s, --schema                  数据库模式（graphile-worker元数据存储位置）
                                           [默认: "graphile_worker"]
      --schema-only             仅安装/更新数据库模式，然后退出
      --once                    执行完所有可运行任务后退出
  -w, --watch                   开发模式：监听任务文件变化（实验性）
      --crontab                 指定crontab文件路径
  -j, --jobs                    并发任务数 [默认: 1]
  -m, --max-pool-size           PostgreSQL连接池最大大小 [默认: 10]
      --poll-interval           轮询间隔(毫秒) [默认: 2000]
      --no-prepared-statements  禁用预编译语句（如用于pgBouncer兼容）

状态与要求

• 生产就绪：已在生产环境中使用
• 版本说明：0.x版本，表示仍在快速迭代，次要版本更新可能需要少量代码调整
• 依赖要求：PostgreSQL 10+，Node.js 10+
• 数据库扩展：自动安装pg***扩展（如需自定义扩展位置，可能需要创建公共模式别名函数）