bull:基于 Redis 的作业队列项目

🚀 赞助商 🚀

Dragonfly 是一个全新的Redis™即插即用替代品，与BullMQ完全兼容，并且相较于Redis™提供了一些重要的优势，例如通过利用所有可用的CPU核心来提供巨大的性能改进，以及更快速、更节省内存的数据结构。阅读更多关于如何与BullMQ一起使用它的信息这里。

📻 新闻与更新

Bull目前处于维护模式，我们只修复bug。对于新特性，请查看 BullMQ，这是一个用Typescript重新编写的现代实现。如果你需要，仍然非常欢迎使用Bull，这是一个安全、经过实战检验的库。

关注我的 Twitter，获取其他重要新闻和更新。

🛠 教程

你可以在以下博客中找到教程和新闻：https://blog.taskforce.sh/

---

使用者

Bull在大小组织中都很受欢迎，以下是一些使用者：

---

---

官方前端

为您的队列增强专业前端：

• 获取所有队列的完整概览。
• 检查任务，搜索，重试或提升延迟任务。
• 度量和统计。
• 以及更多特性。

在 Taskforce.sh 注册。

---

Bull 特性

• 由于无轮询设计，最小化CPU使用。
• 基于Redis的健壮设计。
• 延迟任务。
• 根据cron规范调度和重复任务。
• 任务速率限制器。
• 重试。
• 优先级。
• 并发。
• 暂停/恢复——全局或本地。
• 每个队列支持多种任务类型。
• 线程化（沙盒化）处理函数。
• 自动从进程崩溃中恢复。

即将在路线图上出现的...

• 任务完成确认（你可以暂时使用消息队列pattern）。
• 父/子任务关系。

---

用户界面

有一些第三方UI可用于监控：

BullMQ

• Taskforce

Bull v3

• Taskforce
• bull-board
• bull-repl
• bull-monitor
• Monitoro

Bull <= v2

• Matador
• react-bull
• Toureiro

---

监控与警报

• 使用Prometheus Bull Queue Exporter

---

特性比较

由于有一些工作队列解决方案，以下是它们的比较表格：

特性	BullMQ-Pro	BullMQ	Bull	Kue	Bee	Agenda
后端	redis	redis	redis	redis	redis	mongo
观察者	✓
分组速率限制	✓
分组支持	✓
批次支持	✓
父/子依赖关系	✓	✓
优先级	✓	✓	✓	✓		✓
并发	✓	✓	✓	✓	✓	✓
延迟任务	✓	✓	✓	✓		✓
全局事件	✓	✓	✓	✓
速率限制器	✓	✓	✓
暂停/恢复	✓	✓	✓	✓
沙盒化工作者	✓	✓	✓
可重复任务	✓	✓	✓			✓
原子操作	✓	✓	✓		✓
持久化	✓	✓	✓	✓	✓	✓
UI	✓	✓	✓	✓		✓
优化为	任务 / 消息	任务 / 消息	任务 / 消息	任务	消息	任务

安装指南

npm install bull --save

或者

yarn add bull

需求： Bull 需要一个大于或等于 2.8.18 版本的 Redis。

Typescript 定义

npm install @types/bull --save-dev

请提供需要翻译的英文文本以及期望的中文翻译风格（通俗、专业、优雅或流畅），这样我才能根据您的要求进行翻译。如果您没有提供具体文本，我无法进行翻译工作。谢谢。

yarn add --dev @types/bull

定义目前维护在 DefinitelyTyped 仓库中。

贡献指南

我们欢迎各种类型的贡献，无论是代码修复、新功能添加还是文档改进。 代码格式化由 prettier 强制执行。 对于提交信息，请遵循传统的 提交规范。 所有代码在合并到开发分支前都必须通过代码规范检查和测试套件。

---

快速指南

基本用法

const Queue = require('bull');

const videoQueue = new Queue('video transcoding', 'redis://127.0.0.1:6379');
const audioQueue = new Queue('audio transcoding', { redis: { port: 6379, host: '127.0.0.1', password: 'foobared' } }); // Specify Redis connection using object
const imageQueue = new Queue('image transcoding');
const pdfQueue = new Queue('pdf transcoding');

videoQueue.process(function (job, done) {

  // job.data contains the custom data passed when the job was created
  // job.id contains id of this job.

  // transcode video asynchronously and report progress
  job.progress(42);

  // call done when finished
  done();

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

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

  // If the job throws an unhandled exception it is also handled correctly
  throw new Error('some unexpected error');
});

audioQueue.process(function (job, done) {
  // transcode audio asynchronously and report progress
  job.progress(42);

  // call done when finished
  done();

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

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

  // If the job throws an unhandled exception it is also handled correctly
  throw new Error('some unexpected error');
});

imageQueue.process(function (job, done) {
  // transcode image asynchronously and report progress
  job.progress(42);

  // call done when finished
  done();

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

  // or pass it a result
  done(null, { width: 1280, height: 720 /* etc... */ });

  // If the job throws an unhandled exception it is also handled correctly
  throw new Error('some unexpected error');
});

pdfQueue.process(function (job) {
  // Processors can also return promises instead of using the done callback
  return pdfAsyncProcessor();
});

videoQueue.add({ video: 'http://example.com/video1.mov' });
audioQueue.add({ audio: 'http://example.com/audio1.mp3' });
imageQueue.add({ image: 'http://example.com/image1.tiff' });

使用承诺

或者，您可以选择返回承诺（promises），而不是使用 done 回调函数：

videoQueue.process(function (job) { // don't forget to remove the done callback!
  // Simply return a promise
  return fetchVideo(job.data.url).then(transcodeVideo);

  // Handles promise rejection
  return Promise.reject(new Error('error transcoding'));

  // Passes the value the promise is resolved with to the "completed" event
  return Promise.resolve({ framerate: 29.5 /* etc... */ });

  // If the job throws an unhandled exception it is also handled correctly
  throw new Error('some unexpected error');
  // same as
  return Promise.reject(new Error('some unexpected error'));
});

分离进程

进程函数也可以在独立的进程中运行。这样做有几个优点：

• 进程被沙盒化，因此即使崩溃也不会影响工作进程。
• 你可以运行阻塞代码，而不会影响队列（任务不会停滞）。
• 更好地利用多核 CPU。
• 减少与 Redis 的连接数。

为了使用这个特性，只需创建一个包含处理器的独立文件：

// processor.js
module.exports = function (job) {
  // Do some heavy work

  return Promise.resolve(result);
}

并按如下方式定义处理器：

// Single process:
queue.process('/path/to/my/processor.js');

// You can use concurrency as well:
queue.process(5, '/path/to/my/processor.js');

// and named processors:
queue.process('my processor', 5, '/path/to/my/processor.js');

循环任务

一项任务可以添加到队列中，并根据cron规范进行重复处理：

  paymentsQueue.process(function (job) {
    // Check payments
  });

  // Repeat payment job once every day at 3:15 (am)
  paymentsQueue.add(paymentsData, { repeat: { cron: '15 3 * * *' } });

以下是一些建议，请检查这里的表达式以验证它们是否正确： cron表达式生成器

暂停 / 恢复

队列可以全局暂停和恢复（为仅此工作进程暂停处理传递 true）：

queue.pause().then(function () {
  // queue is paused now
});

queue.resume().then(function () {
  // queue is resumed now
})

事件

队列会触发一些有用的事件，例如...

.on('completed', function (job, result) {
  // Job completed with output result!
})

了解更多关于事件的信息，包括触发事件完整列表，请查阅事件参考。

队列性能

队列的成本低廉，因此如果您需要许多队列，只需创建具有不同名称的新队列即可：

const userJohn = new Queue('john');
const userLisa = new Queue('lisa');
.
.
.

然而，每个队列实例都需要新的 Redis 连接。请查看如何重用连接，或者您也可以使用命名处理器来实现类似的效果。

集群支持

注意：从版本 3.2.0 开始，建议使用多线程处理器替代。

队列健壮，可以在多个线程或进程中并行运行，而不会有任何风险或队列损坏。请查看这个使用集群来跨进程并行化任务的简单示例：

const Queue = require('bull');
const cluster = require('cluster');

const numWorkers = 8;
const queue = new Queue('test concurrent queue');

if (cluster.isMaster) {
  for (let i = 0; i < numWorkers; i++) {
    cluster.fork();
  }

  cluster.on('online', function (worker) {
    // Let's create a few jobs for the queue workers
    for (let i = 0; i < 500; i++) {
      queue.add({ foo: 'bar' });
    };
  });

  cluster.on('exit', function (worker, code, signal) {
    console.log('worker ' + worker.process.pid + ' died');
  });
} else {
  queue.process(function (job, jobDone) {
    console.log('Job done by worker', cluster.worker.id, job.id);
    jobDone();
  });
}

default

文档

查阅完整的文档，请参考以下指南和常见模式：

• 指南 —— 使用 Bull 开发的基础入门。
• 参考 —— 包含所有可用对象和方法的参考文档。
• 模式 —— 常见模式的示例集合。
• 许可 —— Bull 的许可协议，采用 MIT 许可。

如果您发现任何需要更多文档的内容，请提交一个 pull request！

---

重要提示

队列旨在实现“至少一次”的工作策略。这意味着在某些情况下，一个任务可能会被处理多次。这通常发生在工作者在处理期间无法为特定任务保持锁时。

当工作者正在处理一个任务时，它会将任务保持为“锁定”状态，以防止其他工作者处理它。

理解锁的工作方式至关重要，这样可以防止您的任务丢失锁——导致 停滞 ——并因此重新启动。锁是通过在 lockRenewTime（通常是 lockDuration 的一半）间隔上为 lockDuration 创建一个锁来内部实现的。如果在锁能够续期之前 lockDuration 已经过期，任务将被视为停滞并自动重新启动；它将被 双重处理。以下情况下可能会发生这种情况：

1. 运行您任务处理器的 Node 进程意外终止。
2. 您的任务处理器过于占用 CPU，导致 Node 事件循环停滞，因此 Bull 无法续期任务锁（参见 #488 了解我们如何更好地检测这种情况）。您可以修复这个问题，方法是将任务处理器分解成更小的部分，以避免任何单个部分阻塞 Node 事件循环。或者，您可以传递一个较大的 lockDuration 设置值（这种做法的权衡是识别实际停滞任务需要更长的时间）。

因此，您应该始终监听 stalled 事件并将其记录到错误监控系统中，因为这意味着您的任务可能正在被双重处理。

为了确保问题任务不会无限次地重新启动（例如，如果任务处理器总是导致其 Node 进程崩溃），任务将从停滞状态中恢复，最多 maxStalledCount 次（默认值：1）。