使用 NodeJS Bull Queue 独立处理器答案

【问题标题】：Use NodeJS Bull Queue seperate processor使用 NodeJS Bull Queue 独立处理器
【发布时间】：2020-05-13 22:36:28
【问题描述】：

我正在尝试在不同的 processor 文件 itef 中运行 process，就像它在 bull documentation 中一样，我添加了如下文件。

// -------- Queue.js ----------

formatQueue.process(__dirname + "/processors/format-worker.js");


// On Cmplete Handler

formatQueue.on('completed', function(job, result){
    console.log(result, "Format-Complete-job"); // result is undefined
    if(job.data.is_well_format){
        existenceQueue.add(job.data, { attempts: 3, backoff: 1000 });
    }else QueueModel.lastStep(job.data)
});

// -------- Queue.js ends ---------

//format-worker.js
Validator = require("../../validators");
module.exports = (job) => {
    Validator.Format.validate(job.data, (data) => {
        job.data = data;
        return Promise.resolve(data);
    });
}

现在在我之前使用的作业完成时，我曾经使用更新的作业参数获取作业数据。现在我没有得到更新的工作数据。文档中的第二个参数，即result 是undefined。现在如何在这种情况下获取更新的作业数据。

job 和processors 工作正常，我运行process 如下。

formatQueue.process(function(job, done){
    Validator.Format.validate(job.data, (data) => {
        job.data = data;
        done();
    });
});

在这种情况下，工作数据本身正在更新，并且效果也很好。

我有多个队列在做不同的工作，一旦前一个工作成功完成，那么我只需要另一个工作来开始工作。我还有另一个问题，我已经提到了那里的用例。 please check it here.

【问题讨论】：

我最近遇到了类似的问题。您是否尝试将 done 添加到外部处理器的函数并运行它而不是返回 Promise？

标签： node.js queue job-queue bull.js bull-queue

【解决方案1】：

让我帮你，也给你一些提示。首先，您的代码中有一些错误。 process 函数没有返回您在 validate 回调中创建的承诺。现在，我不知道Validator.Format.validate 返回什么，所以为了安全起见，我会这样写：

module.exports = job => {
  return new Promise(resolve => {
    Validator.Format.validate(job.data, data => {
      resolve(data);
    });
  });
};

其次，通常在流程处理程序本身中添加下一个作业而不是在“完成”事件回调中添加下一个作业更健壮，原因是这样做您将获得更健壮的解决方案，在这种情况下添加下一个作业由于某种原因失败，该作业将失败，您将能够重试或检查它失败的原因等等。

module.exports = (job) => {
  return new Promise(resolve => {
    Validator.Format.validate(job.data, (data) => {
      if(data.is_well_format){
        resolve(existenceQueue.add(data, { attempts: 3, backoff: 1000 }));
      }else {
        resolve(QueueModel.lastStep(data))
      }
    });
  }
});

【讨论】：