[Node][Agenda]Node Agenda 中文文档 定时任务调度系统[基础篇]
发布于 2 年前 作者 Bingmang 3209 次浏览 来自 分享

CSDN(鄙人)

Agenda简介

Agenda是一个定时任务管理模块,它将node中的定时任务存储在数据库中(官方文档推荐使用mongoDB),通过事件回调与监听实现定时任务调度。 Agenda

使用步骤概述

总的来说分4步:

  • 初始化(new Agenda(..))
  • 定义任务(agenda.define(..))
  • 配置任务(agenda.every(..))
  • 最后设置事件监听(agenda.on(..)
const Agenda = require('agenda')
const connection_options = {
  db: {
    address: 'mongodb://127.0.0.1:27017/agenda',
    collection: 'agendaJobs',
    options: { server: { auto_reconnect: true } },
  },
}
// 初始化agenda
let agenda = new Agenda(connection_options)
agenda
  .name('AGENDA TEST - ' + process.pid)
  .defaultConcurrency(5)
  .defaultLockLifetime(10000)
// 定义任务
agenda.define('updateCampaignTimeout', { priority: 'high', concurrency: 3 }, (job, done) => {
  dbp_job.updateCampaignTimeoutTick()
    .then(() => done())
    .catch((error) => { throw error })
})
// 配置任务(需要在ready事件中完成)
agenda.on('ready', () => {
  agenda.every('00 09 18 * * *', 'updateCampaignTimeout', {}, {timezone: 'Asia/Shanghai'})
  console.log('agenda测试开始,启动完毕')
  agenda.start()
})
// 设置监听
agenda.on('start', (job) => {
  console.log('检测到job启动: ', job.attrs.name)
})

agenda.on('complete', (job) => {
  console.log('检测到job完成: ', job.attrs.name)
})

agenda.on('success', (job) => {
  console.log('检测到job成功: ', job.attrs.name)
})

agenda.on('fail', (job) => {
  console.log('检测到job失败: ', job.attrs.name)
  console.log('失败时间: ', job.attrs.failedAt)
  console.log('失败原因: ', job.attrs.failReason)
  agenda.stop()
})
// 最后,优雅的退出方案
function graceful() {
  agenda.stop(() => {
    console.log('检测到退出')
    process.exit(0);
  });
}

process.on('SIGTERM', graceful);
process.on('SIGINT', graceful);

步骤详述

初始化

完整参数表点我

初始化有两种方案:

  • 通过完整的参数传递
  • 链式初始化(推荐使用)

部分说明:

  • .database(url, [collectionName])连接数据库,例如agenda.database('mongodb://127.0.0.1:27017/agenda', 'agendaJobs'),其中collectionName可省略,默认值为agendaJobs
  • .name(string)jobs存储于数据库中会有lastModifiedBy字段,存储调度这个任务的agenda名字
  • .processEvery(interval)用于设置agenda轮询数据库检测是否有新的任务加入的扫描时间
  • .defaultLockLimit(number)设置默认值,agenda调度任务时会锁定该任务,防止其他agenda重复执行任务,当任务超出lockLifetime或调用done时会解锁任务,define新任务时如果设置lockLimit(number)则会覆盖默认值。该值用于设置锁定任务的数量,默认为0,即无限制
  • .defaultLockLifetime(number)上面说了,当任务被调度时会被锁定,当任务调用完毕执行done()后会解锁,或者执行的时间很长,超出了lockLifetime也会解锁,同样,define其他任务时如果设置了lockLifetime参数,会覆盖掉默认值

定义任务

agenda.define(jobName, [options], fn)

agenda.define('updateCampaignTimeout', { priority: 'high', concurrency: 3 }, (job, done) => {
  dbp_job.updateCampaignTimeoutTick()
    .then(() => done())
    .catch((error) => { throw error })
})

####参数说明:

jobName 类型String,用于存储于数据库中的name字段,同时配置任务和设置监听会使用到该参数

options 可配置concurrencylockLimitlockLifetimepriority,前文都已说过, priority为任务优先级:lowest|low|normal|high|highest|number,前五个是字符串,传入后会被转为数字,依次对应:-20|-10|0|10|20

fn(job, done)

  • 配置任务时,会设置每次执行任务需要传入的参数,该参数都被放在了job.attrs.data中,
  • 任务出错,会调用job.fail(reason)方法,其中的reason可以是字符串或Error,当调用此方法时,会设置job.attrs.failedAt为当前时间(ISO),job.attrs.failReasonreasonError.message
  • 对于异步任务,必须调用done方法告知任务结束
  • 对于同步任务,省略为fn(job)即可

配置任务

agenda.every(interval, name, [data], [options], [cb])

agenda.every('00 09 18 * * *', 'updateCampaignTimeout', {}, {timezone: 'Asia/Shanghai'})
agenda.every('30 seconds', 'updateCampaignOverBudget')
agenda.every('1 minutes', 'synchronizeBudget')

参数说明

interval 可以是字符串,或cron时间,注意:官方Agenda文档说明的cron时间格式已过时,现在的cron时间格式为

* * * * * *
| | | | | | 
| | | | | +-- Day of the Week   (range: 0-6, 0 standing for Monday)
| | | | +---- Month             (range: 0-11)
| | | +------ Day of the Month  (range: 1-31)
| | +-------- Hour              (range: 0-23)
| +---------- Minute            (range: 0-59)
+------------ Second            (range: 0-59)

设置每隔几分钟的任务用'5 minutes'就好了,但是如果设置每天的什么时候运行,就要使用cron设置,且要注意时区(在options中设置)!!

name 就是任务的名字,一定要对应,可以是字符串数组,这样的话多个任务就使用同一个配置

data 是一个Object,在define任务时通过job.attrs.data获取该对象

options 注意,要设置options必须先设置data,所以data可以传空Object{},在options中可以设置的有timezone,必须是可识别的时区字符串,默认为'America/New_York,而在我们的时区,必须要设置对时区,才能正常运行定时任务!我们的时区字符串是Asia/Shanghai

完整时区查询戳我(维基百科)

cbjob成功被加入数据库时会调用回调函数


设置监听

agenda.on('ready', () => {
  // agenda.every('5 seconds', 'say hello')
  agenda.every('00 09 18 * * *', 'updateCampaignTimeout', {}, {timezone: 'Asia/Shanghai'})
  agenda.every('30 seconds', 'updateCampaignOverBudget')
  agenda.every('1 minutes', 'synchronizeBudget')
  agenda.purge((err, numRemoved) => {
    console.log('旧任务被移除: ', numRemoved)
  })
  console.log('agenda测试开始,启动完毕')
  agenda.start()
})

agenda.on('fail', (job) => {
  console.log('检测到job失败: ', job.attrs.name)
  console.log('失败时间: ', job.attrs.failedAt)
  console.log('失败原因: ', job.attrs.failReason)
  agenda.stop()
})

可以监听的事件有:

  • 'ready', function()
  • 'start', function(job)
  • 'start:job name', function(job)
  • 'complete', function(job)
  • 'complete:job name', function(job)
  • 'success', function(job)
  • 'success:job name', function(job)
  • 'fail', function(err, job)
  • 'fail: job name', function(err, job)

其中比较重要的事件就是ready了,当agenda初始化完毕成功连接数据库后,会发出ready事件,此时就要通过监听该事件,在ready时间中对define过的任务配置时间等参数,最后调用agenda.start()启动agenda

我们发现,在ready事件中,我们还调用了agenda.purge(cb),这个函数是清理数据库中的旧任务(即这此运行程序没有被define到的job)


注意事项

  • agenda在mongoDB中的存储方式:
{
	"_id" : ObjectId("5981983d9b85994179fc98c9"),
	"name" : "updateCampaignTimeout",
	"type" : "single",
	"data" : {
		
	},
	"priority" : 10,
	"repeatInterval" : "00 09 18 * * *",
	"repeatTimezone" : "Asia/Shanghai",
	"lastModifiedBy" : "AGENDA TEST - 97807",
	"nextRunAt" : ISODate("2017-08-03T10:09:00.011Z"),
	"lockedAt" : null,
	"lastRunAt" : ISODate("2017-08-02T10:09:00.011Z"),
	"lastFinishedAt" : ISODate("2017-08-02T10:09:00.076Z"),
	"failReason" : "failed to calculate nextRunAt due to invalid repeat interval",
	"failCount" : 1,
	"failedAt" : ISODate("2017-08-02T09:44:54.551Z")
}
  • 重复启动agenda时,会更新对应的项,而不是覆盖,只要任务名相同,就不会在数据库中创建新的数据项,也不会删除旧的job(该次执行程序没有define到的job),除非在ready事件中调用agenda.purge

  • 在配置任务(agenda.every)时一定要注意时区的设置

  • agenda.every配置失败时(传入的interval不对等原因),会返回undefined,正常情况下返回job对象

  • 关于job对象:

Job {
  agenda:
   EventEmitter {
     _name: 'AGENDA TEST - 97807',
     _processEvery: 5000,
     _defaultConcurrency: 5,
     _maxConcurrency: 20,
     _defaultLockLimit: 0,
     _lockLimit: 0,
     _definitions:
      { updateCampaignTimeout: [Object],
        updateCampaignOverBudget: [Object],
        synchronizeBudget: [Object] },
     _runningJobs: [],
     _lockedJobs: [],
     _jobQueue: [],
     _defaultLockLifetime: 10000,
     _isLockingOnTheFly: false,
     _jobsToLock: [],
     _events:
      { ready: [Function],
        start: [Function],
        complete: [Function],
        success: [Function],
        fail: [Function] },
     _eventsCount: 5,
     _mdb:
      Db {
        domain: null,
        _events: {},
        _eventsCount: 0,
        _maxListeners: undefined,
        s: [Object],
        serverConfig: [Getter],
        bufferMaxEntries: [Getter],
        databaseName: [Getter] },
     _collection: Collection { s: [Object] } },
  attrs:
   { name: 'updateCampaignTimeout',
     data: {},
     type: 'single',
     priority: 10,
     nextRunAt: moment.parseZone("2017-08-02T18:09:00.009+08:00"),
     repeatInterval: '00 09 18 * * *',
     repeatTimezone: 'Asia/Shanghai' } }
  • 官方README的CRON格式过时了,最新的CRON格式请见上文

待补充……

1 回复

弱弱的问一句,这是一个带定时功能的消息队列吗?我是这么理解的

回到顶部