@ agendajs/mongo-后端
MongoDB 的后端 议程项目 工作调度员.
安装
npm install agenda @agendajs/mongo-backend
# or
pnpm add agenda @agendajs/mongo-backend
# or
yarn add agenda @agendajs/mongo-backend
所需经费:
- 节点.js 18+
- 蒙戈DB 4+ 导弹
使用量
import { Agenda } from 'agenda';
import { MongoBackend } from '@agendajs/mongo-backend';
// Via connection string
const agenda = new Agenda({
backend: new MongoBackend({
address: 'mongodb://localhost/agenda'
})
});
// Or via existing MongoDB connection
import { MongoClient } from 'mongodb';
const client = await MongoClient.connect('mongodb://localhost');
const db = client.db('agenda');
const agenda = new Agenda({
backend: new MongoBackend({ mongo: db })
});
// Define jobs
agenda.define('send-email', async (job) => {
const { to, subject } = job.attrs.data;
await sendEmail(to, subject);
});
// Start processing
await agenda.start();
// Schedule jobs
await agenda.every('1 hour', 'send-email', { to: '[email protected]', subject: 'Hello' });
配置选项
interface IMongoBackendConfig {
/** MongoDB connection string */
address?: string;
/** Existing MongoDB database instance */
mongo?: Db;
/** Collection name for jobs (default: 'agendaJobs') */
collection?: string;
/** MongoDB client options */
options?: MongoClientOptions;
/** Name to identify this Agenda instance (stored as lastModifiedBy) */
name?: string;
/** Whether to create indexes on connect (default: true) */
ensureIndex?: boolean;
/** Sort order for job queries */
sort?: { [key: string]: SortDirection };
}
带选项的完整示例
const agenda = new Agenda({
backend: new MongoBackend({
address: 'mongodb://localhost/agenda',
collection: 'jobs', // Custom collection name
sort: { nextRunAt: 'asc', priority: 'desc' }
}),
processEvery: '30 seconds', // Job polling interval
maxConcurrency: 20, // Max concurrent jobs
defaultConcurrency: 5 // Default per job type
});
实时通知
MongoBackend只提供存储,并使用投票来检查新的工作. 对于跨分布式系统的实时工作处理,您有几种选择:
备选案文1:MongoDB变化流(国内)
如果您的 MongoDB 部署是复制件集( 甚至单节点复制件集) , 您可以使用 MongoChangeStreamNotificationChannel 本地实时通知中没有任何外部依赖性:
import { Agenda } from 'agenda';
import { MongoBackend, MongoChangeStreamNotificationChannel } from '@agendajs/mongo-backend';
const agenda = new Agenda({
backend: new MongoBackend({ mongo: db }),
notificationChannel: new MongoChangeStreamNotificationChannel({ db })
});
// Jobs are processed immediately when created (no polling delay)
await agenda.start();
await agenda.now('myJob'); // Triggers instant processing
所需经费:
- MongoDB 复制套件(单节点复制套件工作精细)
- 有线老虎存储引擎(自MongoDB 3.2起默认)
配置选项 :
const channel = new MongoChangeStreamNotificationChannel({
db: mongoDb, // Required: MongoDB database instance
collection: 'agendaJobs', // Optional: collection name (default: 'agendaJobs')
resumeToken: savedToken, // Optional: resume from specific point
fullDocument: true // Optional: include full document on updates (default: true)
});
备选案文2:记忆(单一进程)
import { Agenda, InMemoryNotificationChannel } from 'agenda';
import { MongoBackend } from '@agendajs/mongo-backend';
// For single-process apps (testing/development)
const agenda = new Agenda({
backend: new MongoBackend({ mongo: db }),
notificationChannel: new InMemoryNotificationChannel()
});
备选案文3:Redis Pub/Sub(多程序)
import { Agenda } from 'agenda';
import { MongoBackend } from '@agendajs/mongo-backend';
import { RedisBackend } from '@agendajs/redis-backend';
// For production: use Redis for notifications
const redisBackend = new RedisBackend({ connectionString: 'redis://localhost:6379' });
const agenda = new Agenda({
backend: new MongoBackend({ mongo: db }),
notificationChannel: redisBackend.notificationChannel
});
数据库索引
默认情况下, MongoBackend 自动创建 findAndLockNextJobIndex 用于优化任务查询性能的连接索引。 该指数包括:
{
"name": 1,
"nextRunAt": 1,
"priority": -1,
"lockedAt": 1,
"disabled": 1
}
要禁用自动索引创建( 例如, 如果您单独管理索引 ) :
const backend = new MongoBackend({
mongo: db,
ensureIndex: false // Skip automatic index creation
});
附加索引
如果您使用 议程项目sh中,添加此索引,用于更好的仪表板性能:
db.agendaJobs.createIndex({
"nextRunAt": -1,
"lastRunAt": -1,
"lastFinishedAt": -1
}, { name: "agendash" })
如果您有上千个实例的工作定义, 此索引可以改进锁定查询 :
db.agendaJobs.createIndex({
"name": 1,
"disabled": 1,
"lockedAt": 1
}, { name: "findAndLockDeadJobs" })
任务日志
MongoBackend 包含内置 MongoJobLogger 将结构化的工作生命周期活动储存在专门的MongoDB收藏中(agenda_logs (默认情况下). 伐木机是轻量级的,并且只在第一次使用时创建它的收藏.
自动使用
当您在议事日程中启用日志时, 后端的内置登录器会被自动使用 :
const agenda = new Agenda({
backend: new MongoBackend({ mongo: db }),
logging: true // Uses MongoJobLogger automatically
});
独立使用
您还可以使用 MongoJobLogger 独立 – 例如, 在使用不同的后端存储时登录到 MongoDB :
import { MongoJobLogger } from '@agendajs/mongo-backend';
const logger = new MongoJobLogger({
mongo: db, // Required: MongoDB database instance
logCollection: 'agenda_logs' // Optional: collection name (default: 'agenda_logs')
});
// Use with any backend
import { RedisBackend } from '@agendajs/redis-backend';
const agenda = new Agenda({
backend: new RedisBackend({ connectionString: 'redis://...' }),
logging: logger
});
查询日志
const { entries, total } = await agenda.getLogs({
jobName: 'send-email',
level: 'error',
limit: 50,
sort: 'desc'
});
// Clear old logs
await agenda.clearLogs({ to: new Date(Date.now() - 30 * 86400000) });
见 议程项目研究 用于完整记录文件.
相关软件包
- 议程 - 核心排程器
- @ agendajs/postgres-后端 - PostgreSQL 后端,配有Listen/NOTIFY
- @ agendajs/ redis-后端 - 与 Pub/ Sub 重接后端
许可证
麻省理工学院