如何在公牛队列中收听已完成的事件-仅针对当前工作

Question

运行下面的代码时会打印太多结果。

我怀疑completed 事件监听了当前队列实例中所有先前的作业。

如何管理已完成事件以仅监听当前作业完成？

producer.js

生产者创建一个具有默认数字 id 的作业并侦听全局完成，以便在作业完成时返回响应。

const BullQ = require('bull');

let bullQ = BullQ('my-first-queue', {
  redis: {
    host: process.env.REDISHOST || 'localhost',
    password: process.env.REDISPASSWORD || ''
  }
});

app.get('/search/:term', async (req, res) => {
  const job = await bullQ.add({
    searchTerm: req.params.term
  });
  
  // Listen to the global completion of the queue in order to return result.
  bullQ.on('global:completed', (jobId, result) => {
    // Check if id is a number without any additions
    if (/^\d+$/.test(jobId) && !res.headersSent) {
      console.log(`Producer get: Job ${jobId} completed! Result: ${result}`);
      res.json(`Job is completed with result: ${result}`);
    }
  });
});

consumer.js

消费者有两个角色。

1. 按本应有的方式使用工作
2. 根据上一个作业的结果创建新作业。

const BullQ = require('bull');

let bullQ = BullQ('my-first-queue', {
  redis: {
    host: process.env.REDISHOST || 'localhost',
    password: process.env.REDISPASSWORD || ''
  }
});

bullQ.process((job, done) => {
  // Simulate asynchronous server request.
  setTimeout(async () => {
    // Done the first job and return an answer to the producer after the timeout.
    done(null, `Search result for ${job.data.searchTerm}`);
    // next job run
    if (counter < 10) {
      // For the first run the id is just a number if not changed via the jobId in JobOpts,
      // the next time the job id will be set to {{id}}_next_{{counter}} we need only the first number in order not to end with a long and not clear concatenated string.
      let jobID = (/^\d+$/.test(job.id)) ? job.id : job.id.replace(/[^\d].*/,'');
      await createNextJob(jobID, ++counter);
    }
  }, 100);
});

// Create next job and add it to the queue.
// Listen to the completed jobs (locally)
const createNextJob = async (id, counter) => {
  const nextJob = bullQ.add({
    searchTerm: "Next job"
  }, {
    jobId: `${id}_next_${counter}`
  });

  await bullQ.on('completed', (job, result) => {
    job.finished();
    console.log(`Consumer(next): Job ${job.id} completed! Result: ${result}`);
  });
};

Answer 1

您可以通过await job.finished() 获取特定于由queue.add() 返回的job 对象的结果。

这是一个简单的、可运行的示例，无需 Express 来说明：

const Queue = require("bull"); // "bull": "^3.22.6"

const sleep = (ms=1000) =>
  new Promise(resolve => setTimeout(resolve, ms))
;

const queue = new Queue("test", process.env.REDIS_URL);
queue.process(4, async job => {
  await sleep(job.data.seconds * 1000); // waste time
  return Promise.resolve(`job ${job.id} complete!`);
});

(async () => {
  const job = await queue.add({seconds: 5});
  const result = await job.finished();
  console.log(result); // => job 42 complete!
  await queue.close();
})();

有快递：

const Queue = require("bull");
const express = require("express");

const sleep = (ms=1000) =>
  new Promise(resolve => setTimeout(resolve, ms))
;

const queue = new Queue("test", process.env.REDIS_URL);
queue.process(4, async job => {
  await sleep(job.data.seconds * 1000);
  return Promise.resolve(`job ${job.id} complete!`);
});

const app = express();
app
  .set("port", process.env.PORT || 5000)
  .get("/", async (req, res) => {
    try {
      const job = await queue.add({
        seconds: Math.abs(+req.query.seconds) || 10,
      });
      const result = await job.finished();
      res.send(result);
    }
    catch (err) {
      res.status(500).send({error: err.message});
    }
  })
  .listen(app.get("port"), () => 
    console.log("app running on port", app.get("port"))
  )
;

示例运行：

$ curl localhost:5000?seconds=2
job 42 complete!