bull README

The project's README from github.com/OptimalBits/bull. As one markdown file: README.md.




The fastest, most reliable, Redis-based queue for Node.
Carefully written for rock solid stability and atomicity.


Sponsors · Features · UIs · Install · Quick Guide · Documentation

Check the new Guide!


🚀 Sponsors 🚀

Dragonfly Dragonfly is a new Redis™ drop-in replacement that is fully compatible with BullMQ and brings some important advantages over Redis™ such as massive better performance by utilizing all CPU cores available and faster and more memory efficient data structures. Read more here on how to use it with BullMQ.

📻 News and updates

Bull is currently in maintenance mode, we are only fixing bugs. For new features check BullMQ, a modern rewritten implementation in Typescript. You are still very welcome to use Bull if it suits your needs, which is a safe, battle tested library.

Follow me on Twitter for other important news and updates.

🛠 Tutorials

You can find tutorials and news in this blog: https://blog.taskforce.sh/

Used by

Bull is popular among large and small organizations, like the following ones:

Atlassian Autodesk Mozilla Nest Salesforce

Official FrontEnd

Taskforce.sh, Inc

Supercharge your queues with a professional front end:

Sign up at Taskforce.sh

Bull Features

And coming up on the roadmap...

UIs

There are a few third-party UIs that you can use for monitoring:

BullMQ

Bull v3

Bull <= v2

Monitoring & Alerting

Feature Comparison

Since there are a few job queue solutions, here is a table comparing them:

Feature BullMQ-Pro BullMQ Bull Kue Bee Agenda
Backend redis redis redis redis redis mongo
Observables ✓
Group Rate Limit ✓
Group Support ✓
Batches Support ✓
Parent/Child Dependencies ✓ ✓
Priorities ✓ ✓ ✓ ✓ ✓
Concurrency ✓ ✓ ✓ ✓ ✓ ✓
Delayed jobs ✓ ✓ ✓ ✓ ✓
Global events ✓ ✓ ✓ ✓
Rate Limiter ✓ ✓ ✓
Pause/Resume ✓ ✓ ✓ ✓
Sandboxed worker ✓ ✓ ✓
Repeatable jobs ✓ ✓ ✓ ✓
Atomic ops ✓ ✓ ✓ ✓
Persistence ✓ ✓ ✓ ✓ ✓ ✓
UI ✓ ✓ ✓ ✓ ✓
Optimized for Jobs / Messages Jobs / Messages Jobs / Messages Jobs Messages Jobs

Install

npm install bull --save

or

yarn add bull

Requirements: Bull requires a Redis version greater than or equal to 2.8.18.

Typescript Definitions

npm install @types/bull --save-dev

yarn add --dev @types/bull

Definitions are currently maintained in the DefinitelyTyped repo.

Contributing

We welcome all types of contributions, either code fixes, new features or doc improvements. Code formatting is enforced by prettier. For commits please follow conventional commits convention. All code must pass lint rules and test suites before it can be merged into develop.

Quick Guide

Basic Usage

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' });

Using promises

Alternatively, you can return promises instead of using the done callback:

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'));
});

Separate processes

The process function can also be run in a separate process. This has several advantages:

In order to use this feature just create a separate file with the processor:

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

  return Promise.resolve(result);
}

And define the processor like this:

// 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');

Repeated jobs

A job can be added to a queue and processed repeatedly according to a cron specification:

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

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

As a tip, check your expressions here to verify they are correct: cron expression generator

Pause / Resume

A queue can be paused and resumed globally (pass true to pause processing for just this worker):

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

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

Events

A queue emits some useful events, for example...

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

For more information on events, including the full list of events that are fired, check out the Events reference

Queues performance

Queues are cheap, so if you need many of them just create new ones with different names:

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

However every queue instance will require new redis connections, check how to reuse connections or you can also use named processors to achieve a similar result.

Cluster support

NOTE: From version 3.2.0 and above it is recommended to use threaded processors instead.

Queues are robust and can be run in parallel in several threads or processes without any risk of hazards or queue corruption. Check this simple example using cluster to parallelize jobs across processes:

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();
  });
}

Documentation

For the full documentation, check out the reference and common patterns:

If you see anything that could use more docs, please submit a pull request!

Important Notes

The queue aims for an "at least once" working strategy. This means that in some situations, a job could be processed more than once. This mostly happens when a worker fails to keep a lock for a given job during the total duration of the processing.

When a worker is processing a job it will keep the job "locked" so other workers can't process it.

It's important to understand how locking works to prevent your jobs from losing their lock - becoming stalled - and being restarted as a result. Locking is implemented internally by creating a lock for lockDuration on interval lockRenewTime (which is usually half lockDuration). If lockDuration elapses before the lock can be renewed, the job will be considered stalled and is automatically restarted; it will be double processed. This can happen when:

  1. The Node process running your job processor unexpectedly terminates.
  2. Your job processor was too CPU-intensive and stalled the Node event loop, and as a result, Bull couldn't renew the job lock (see #488 for how we might better detect this). You can fix this by breaking your job processor into smaller parts so that no single part can block the Node event loop. Alternatively, you can pass a larger value for the lockDuration setting (with the tradeoff being that it will take longer to recognize a real stalled job).

As such, you should always listen for the stalled event and log this to your error monitoring system, as this means your jobs are likely getting double-processed.

As a safeguard so problematic jobs won't get restarted indefinitely (e.g. if the job processor always crashes its Node process), jobs will be recovered from a stalled state a maximum of maxStalledCount times (default: 1).

bull (OptimalBits/bull)

Searching Database…

16.3K ·GitHub
Bull
Что это такое?Bull — это пакет очередей на основе Redis для Node.js, предназначенный для распределённой обработки задач и сообщений: фоновых операций, работы по расписанию и обмена сообщениями. Библиотека написана с упором на надёжность и атомарность операций.
Это библиотека, которую я добавляю в своё Node-приложение, или что-то, что разворачивается отдельно?Это npm-библиотека, которую вы подключаете внутри своего Node.js-приложения — отдельного сервера Bull разворачивать не нужно. Состояние очереди хранится в вашем Redis, а задачи обрабатывают воркеры в вашем коде.
Как добавить его в свой проект?Установите пакет командой npm install bull, создайте именованную очередь и подключите функцию-обработчик, которая выполняет каждую задачу. В README есть ссылки на руководство, полную справку по API и распространённые паттерны.
Нужно ли запускать Redis рядом с ним?Да — Bull хранит всё состояние очереди в Redis, версия которого должна быть 2.8.18 или новее, а также требуется Node.js 12 или выше. Определения типов для TypeScript поддерживаются отдельно в репозитории DefinitelyTyped.
Может ли задача случайно выполниться больше одного раза?Да — Bull работает по принципу «at least once» (минимум один раз): если воркер теряет блокировку (падение или ресурсоёмкая задача, блокирующая event loop), задача перезапускается и может выполниться дважды. Перезапуски зависших задач прекращаются по достижении maxStalledCount (по умолчанию 1); следите за событием stalled.
В слогане Bull назван «Premium Queue» — пользоваться им платно?Нет — Bull — проект с открытым исходным кодом под лицензией MIT, бесплатный для использования. Taskforce.sh — необязательный хостинг-интерфейс для очередей — это отдельный сервис, к которому можно подключиться.
Что он умеет из коробки?Отложенные задачи, повторяющиеся задачи по cron-расписанию, повторные попытки, приоритеты, управление параллелизмом, ограничитель частоты выполнения задач, паузу и возобновление как глобально, так и для отдельных воркеров, несколько типов задач в одной очереди, изолированные (sandboxed) обработчики и автоматическое восстановление после падения процесса.
Есть ли интерфейс, чтобы следить за своими очередями?Да — несколько сторонних инструментов перечислены в README: bull-board, bull-repl, bull-monitor, Matador, Toureiro и другие, а также хостинг-интерфейс от Taskforce.sh с обзором очередей, инспекцией задач и метриками. Есть ещё Prometheus-экспортёр для алертинга.
Чем он отличается от Kue, Bee или Agenda?Все это очереди задач, но Kue и Bee работают на Redis, а Agenda хранит задачи в MongoDB. Согласно сравнительной таблице из README, Bull — единственный из этих трёх, у кого есть ограничитель частоты, пауза/возобновление и изолированные воркеры; у Bee нет приоритетов, а у Agenda нет глобальных событий и паузы/возобновления.
Могут ли обработчики задач работать вне основного Node-процесса?Да — обработчик может находиться в отдельном файле и выполняться изолированно (sandboxed): его падение не уронит вашего воркера, блокирующий код не остановит очередь, а использование многоядерного процессора улучшится при меньшем числе соединений с Redis. Начиная с версии 3.2.0 рекомендуется использовать обработчики в потоках (threaded).
What are similar projects?rq, effect-mq, xxl-job, celery, sidekiq, redisson, asynq, tork
Which topics is it in?job, job-queue, job-scheduler, jobs, queue, background-jobs, message-queue, event-driven, cron, scheduler, messaging, concurrency
Creator
OptimalBits
Size
4.1 MB
Stars
16.3K
Updated
CommitMay 10Releasev4.16.5Dec 18
Age
13y
Commits · 90 Days