Bull队列未完成

Question

我是bull的新手，我已经尝试过基于他们的文档代码运行公牛。进程正在启动，但是我的任务没有完成，或者不确定它是否触发了complete事件？我不确定我哪里弄错了

将我的代码附加到下面

const Queue = require('bull');

const myFirstQueue = new Queue('my-first-queue', 
{
  redis: { 
      port: Config.redis.port, 
      host: Config.redis.host, 
      password: Config.redis.password 
  },
});



(async function ad() {
    const job = await myFirstQueue.add({
    foo: 'bar',
  });
})();

myFirstQueue.process(async (job, data) => {
  log.debug({ job, data }, 'Job data');
  let progress = 0;
  for (let i = 0; i < 10; i++) {
    await doSomething(data);
    progress += 10;
    job.progress(progress).catch(err => {
      log.debug({ err }, 'Job progress err');
    });
    log.debug({ progress }, 'After await');
  }

  return job;
});

const doSomething = data => {
  return new Promise((resolve, reject) => {
    return resolve(data);
  });
};

myFirstQueue.on('completed', (job, result) => {
  log.debug(`Job completed with result ${job}`);
});


myFirstQueue.on('progress', (job, progress) => {
log.debug(`Job progress with result ${job} ${progress}`);
});

我可以看到日志，它在进度事件处理程序中，但是完成事件没有被触发。任何帮助我们都将不胜感激

Answer 1

您需要从process调用done()，然后只会触发已完成的事件。

myFirstQueue.process(async (job, done) => {
  const data = job.data;
  let progress = 0;
  for (let i = 0; i < 10; i++) {
    await doSomething(data);
    progress += 10;

    job.progress(progress).catch(err => {
      log.debug({ err }, 'Job progress err');
    });
  }

  done();
});

Answer 2

您可以从作业对象本身获取数据，不需要向外部传递数据，处理回调需要job和data()作为参数。最后调用data()回调完成一个任务。您可以传递数据，如果某些验证失败，还可以传递错误。更好地解释https://github.com/OptimalBits/bull/blob/develop/REFERENCE.md#queueprocess

  // call done when finished
  done();

  // or give a error if error
  done(new Error('error transcoding'));

  // or pass it a result
  done(null, { framerate: 29.5 /* etc... */ });

如果不想调用done()回调，可以直接使用promise，ref-https://github.com/OptimalBits/bull#using-promises

对你的代码做了一些小的修改，希望它能有所帮助。

const Queue = require('bull');

const myFirstQueue = new Queue('my-first-queue');



(async function ad() {
    const job = await myFirstQueue.add({
    foo: 'bar',
  });
})();

myFirstQueue.process(async (job, done) => {
  log.debug( 'Job data ' + job.data);
  let progress = 0;
   for (let i = 0; i < 10; i++) {
    await doSomething(job.data);
    progress += 10;
    job.progress(progress).catch(err => {
      log.debug({ err }, 'Job progress err');
    });
    log.debug({ progress }, 'After await');
  }

  done(null, {done: job.data}); //we need this if we are not using promise
});

const doSomething = data => {
  return new Promise((resolve, reject) => {
    return resolve(data);
  });
};

myFirstQueue.on('completed', (job, result) => {
  log.debug(`Job completed with result ${job}`);
});


myFirstQueue.on('progress', (job, progress) => {
log.debug(`Job progress with result ${job} ${progress}`);
});