如何在使用集群模块的 Node.js 应用程序中运行 Cron Job？

Question

我正在使用node-cron 模块在 Node.js 应用程序中调度任务。我还想使用核心集群模块在多个进程中运行应用程序。

在多个进程中运行应用程序最终会在每个进程中执行计划任务（例如，如果任务要发送电子邮件，则电子邮件将被发送多次）。

与集群模块一起运行 cron 作业的最佳做法/可能的方法是什么？我是否应该创建一些单独的进程来仅处理 cron 作业并且不接受任何请求。如果是，我该如何以正确的方式做到这一点？

Answer 1

经过一些研究，我最终得到了“Distributed locks using Redis”解决方案。 有一个节点模块：node-redis-warlock。

希望这个答案对其他人有用。

更新。最小示例代码：

var Warlock = require('node-redis-warlock'),
    redis = require('redis');

// Establish a redis client
redis = redis.createClient();

// and pass it to warlock
var warlock = new Warlock(redis);

function executeOnce (key, callback) {
    warlock.lock(key, 20000, function(err, unlock){
        if (err) {
            // Something went wrong and we weren't able to set a lock
            return;
        }

        if (typeof unlock === 'function') {
            setTimeout(function() {
                callback(unlock);
            }, 1000);
        }
    });
}

// Executes call back only once
executeOnce('every-three-hours-lock', function(unlock) {
    // Do here any stuff that should be done only once...            
    unlock();          
});

更新 2。更详细的例子：

const CronJob = require('cron').CronJob;
const Warlock = require('node-redis-warlock');
const redis = require('redis').createClient();
const warlock = new Warlock(redis);
const async = require('async');

function executeOnce (key, callback) {
    warlock.lock(key, 20000, function(err, unlock) {
        if (err) {
            // Something went wrong and we weren't able to set a lock
            return;
        }

        if (typeof unlock === 'function') {
            setTimeout(function() {
                callback(unlock);
            }, 1000);
        }
    });
}

function everyMinuteJobTasks (unlock) {
    async.parallel([
        sendEmailNotifications,
        updateSomething,
        // etc...
    ],
    (err) => {
        if (err) {
            logger.error(err);
        }

        unlock();
    });
}

let everyMinuteJob = new CronJob({
    cronTime: '*/1 * * * *',
    onTick: function () {
        executeOnce('every-minute-lock', everyMinuteJobTasks);
    },
    start: true,
    runOnInit: true
});

/* Actual tasks */
let sendEmailNotifications = function(done) {
    // Do stuff here
    // Call done() when finished or call done(err) if error occurred
}

let updateSomething = function(done) {
    // Do stuff here
    // Call done() when finished or call done(err) if error occurred
}

// etc...

Answer 2

如果使用PM2， 您可以使用 PM2 本身提供的名为 NODE_APP_INSTANCE 的环境变量，它需要 PM2 2.5 或更高版本。

NODE_APP_INSTANCE 环境变量可用于确定进程之间的差异，例如您可能只想在一个进程上运行 cronjob，您可以这样做

if(process.env.NODE_APP_INSTANCE == 0) { //schedule your cron job here since this part will be executed for only one cluster } ,

因为两个进程不可能有相同的编号。

PM2官方文档here的更多信息。

Answer 3

我实际上不喜欢 cron-cluster npm 插件中也使用的 redis 方法，因为我不想让那个 redis 服务器在我的机器上运行并维护它。

我想和你讨论这种方法：

亲：我们不需要使用redis 缺点：cron 作业总是在同一个 worker 上运行

我只为此使用消息传递，如果您将其用于其他事情，则要传递的信息

if (cluster.isMaster) {
    // Count the machine's CPUs
    var cpuCount = require('os').cpus().length;;

    // Create a worker for each CPU
    for (var i = 0; i < cpuCount; i += 1) {
        cluster.fork();
    }

    cluster.on('fork', (worker) => {
        console.log("cluster forking new worker", worker.id);
    });

    // have a mainWorker that does the cron jobs.
    var mainWorkerId = null;

    cluster.on('listening', (worker, address) => {
        console.log("cluster listening new worker", worker.id);
        if(null === mainWorkerId) {
            console.log("Making worker " + worker.id + " to main worker");
            mainWorkerId = worker.id;
        worker.send({order: "startCron"});
        }
    });

    // Listen for dying workers if the mainWorker dies, make a new mainWorker
    cluster.on('exit', function (worker, code, signal) {
        console.log('Worker %d died :(', worker.id);

        if(worker.id === mainWorkerId) {
            console.log("Main Worker is dead...");
            mainWorkerId = null;
        }

        console.trace("I am here");
        console.log(worker);
        console.log(code);
        console.log(signal);
        cluster.fork();

    });
// Code to run if we're in a worker process
} else {

    // other code like setup app and stuff

    var doCron = function() {
        // setup cron jobs...
    }

    // Receive messages from the master process.
    process.on('message', function(msg) {
        console.log('Worker ' + process.pid + ' received message from master.', message);
        if(message.order == "startCron") {
            doCron();
        }
    });
}

Answer 4

我也遇到了集群模块的问题，最后我找到了解决问题的示例方法。

让master集群执行cronJob。

我的项目使用 Kue 来管理作业。当 cronJob 运行时，我会得到一个作业列表。

index.js

global.cluster = require('cluster');

if (cluster.isMaster) {
  const cpuCount = require('os').cpus().length;
  for (let i = 0; i < cpuCount; i += 1) {
    cluster.fork();
  }
} else {
  // start your express server here
  require('./server')
}

cluster.on('exit', worker => {
  logger.warn('Worker %d died :(', worker.id);
  cluster.fork();
});

cron.js

const cron = require('cron').CronJob;

const job = new cron('* * * * *', async () => {
  if (cluster.isMaster) {
    console.log('cron trigger');
  }
});

job.start();

希望对您有所帮助。

Answer 5

我认为您可以使用节点集群模块，在那里您可以编写代码仅在主集群中运行

const cluster = require('cluster');

if (cluster.isMaster) {
     // Write your code which you want to execute in the master cluster only
}

这是一种节点方式处理集群，当然你可以使用任何工具如pm2来处理。