Cara menggunakan nodejs queue library

Bee-Queue is meant to power a distributed worker pool and was built with short, real-time jobs in mind. A web server can enqueue a job, wait for a worker process to complete it, and return its results within an HTTP request. Scaling is as simple as running more workers.

Thanks to the folks at Mixmax, Bee-Queue is once again being regularly maintained!

Celery, Resque, Kue, and Bull operate similarly, but are generally designed for longer background jobs, supporting things like job prioritization and repeatable jobs, which Bee-Queue . Bee-Queue can handle longer background jobs just fine, but they aren't .

• Create, save, and process jobs
• Concurrent processing
• Job timeouts, retries, and retry strategies
• Scheduled jobs
• Pass events via Pub/Sub
  • Progress reporting
  • Send job results back to producers
• Robust design
  • Strives for all atomic operations
  • Retries for at-least-once delivery
  • High code coverage
• Performance-focused
  • Minimizes
  • Uses Lua scripting and pipelining to minimize network overhead
  • favorably against similar libraries
• Fully callback- and Promise-compatible API

Installation

$ npm install bee-queue

You'll also need Redis 2.8+* running somewhere.

* We've been noticing that some jobs get delayed by virtue of an issue with Redis < 3.2, and therefore recommend the use of Redis 3.2+.

• License

Celery is for Python, and Resque is for Ruby, but Kue and Bull already exist for Node, and they're good at what they do, so why does Bee-Queue also need to exist?

In short: we needed to mix and match things that Kue does well with things that Bull does well, and we needed to squeeze out more performance. There's also a long version with more details.

Bee-Queue starts by combining Bull's simplicity and robustness with Kue's ability to send events back to job creators, then focuses heavily on minimizing overhead, and finishes by being strict about code quality and testing. It compromises on breadth of features, so there are certainly cases where Kue or Bull might be preferable (see ).

Bull and Kue do things really well and deserve a lot of credit. Bee-Queue borrows ideas from both, and Bull was an especially invaluable reference during initial development.

Why Bees?

Bee-Queue is like a bee because it:

• is small and simple
• is fast (bees can fly 20mph!)
• carries pollen (messages) between flowers (servers)
• something something "worker bees"

These basic benchmarks ran 10,000 jobs through each library, at varying levels of concurrency, with Node.js (v6.9.1, v6.11.2, v7.6.0, v7.10.1, v8.2.1, v8.3.0) and Redis (v3.2.10, v4.0.1) running directly on an Amazon AWS EC2 m4.large. The numbers shown are averages of 36 runs, 3 for each combination of the aforementioned Redis and Node versions. The raw data collected and code used are available in the benchmark folder.

Check out the Arena web interface to manage jobs and inspect queue health.

Creating Queues

objects are the starting point to everything this library does. To make one, we just need to give it a name, typically indicating the sort of job it will process:

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Queues are very lightweight — the only significant overhead is connecting to Redis — so if you need to handle different types of jobs, just instantiate a queue for each:

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Here, we pass a

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Creating Jobs

Jobs are created using

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Jobs have a chaining API for configuring the Job, and

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

The Job's

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Each Job can be configured with the commands

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

Jobs can later be retrieved from Redis using , but most use cases won't need this, and can instead use .

Advanced: Bulk-Creating Jobs

Normally, creating and saving jobs blocks the underlying redis client for the full duration of an RTT to the Redis server. This can reduce throughput in cases where many operations should occur without delay - particularly when there are many jobs that need to be created quickly. Use

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

Each job in the array provided to saveAll will be mutated with the ID it gets assigned.

Processing Jobs

To start processing jobs, call

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Instead of calling the provided callback, the handler function can return a

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

The handler function is given the job it needs to process, including

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

Progress Reporting

Handlers can send progress reports, which will be received as events on the original job instance:

const job = addQueue.createJob({x: 2, y: 3}).save();
job.on('progress', (progress) => {
  console.log(
    `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
  );
});

addQueue.process(async (job) => {
  // do some work
  job.reportProgress({page: 3, totalPages: 11});
  // do more work
  job.reportProgress({page: 9, totalPages: 11});
  // do the rest
});

Just like

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Job and Queue Events

There are three classes of events emitted by Bee-Queue objects: , , and . The linked API Reference sections provide a more complete overview of each.

Progress reporting, demonstrated above, happens via Job events. Jobs also emit

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

Queue PubSub events correspond directly to Job events:

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

Queue local events include

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Note that Job events become unreliable across process restarts, since the queue's reference to the associated job object will be lost. Queue-level events are thus potentially more reliable, but Job events are more convenient in places like HTTP requests where a process restart loses state anyway.

Stalling Jobs

Bee-Queue attempts to provide "at least once delivery". Any job enqueued should be processed at least once - and if a worker crashes, gets disconnected, or otherwise fails to confirm completion of the job, the job will be dispatched to another worker for processing.

To make this happen, workers periodically phone home to Redis about each job they're working on, just to say "I'm still working on this and I haven't stalled, so you don't need to retry it." The method finds any active jobs whose workers have gone silent (not phoned home for at least ms), assumes they have stalled, emits a

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

Optimizing Redis Connections

By default, every time you create a queue instance with

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

Let's say for example you have a web application with 30 producer queues and you run 10 webservers & 10 worker servers, each one with 4 processes/server. With the default settings this is going to add up to a lot of Redis connections. Each Redis connection consumes a fairly large chunk of memory, and it adds up quickly!

The producer queues are the ones that run on the webserver and they push jobs into the queue. These queues do not need to receive events so they can all share one redis connection by passing in an instance of .

Example:

$ npm install bee-queue

Note that these "producer queues" above are only relevant for the processes that have to put jobs into the queue, not for the workers that need to actually process the jobs.

In your worker process where you define how to process the job with

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

$ npm install bee-queue

For a more detailed example and explanation see #96

Queue

Settings

The default Queue settings are:

$ npm install bee-queue

The

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

• const baseUrl = 'http://www.google.com/search?q=';
  subQueue.process(10, function (job, done) {
    http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
      // parse the difference out of the response...
      return done(null, difference);
    });
  });

  8: string, default

  const baseUrl = 'http://www.google.com/search?q=';
  subQueue.process(10, function (job, done) {
    http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
      // parse the difference out of the response...
      return done(null, difference);
    });
  });

  9. Useful if the

  const job = addQueue.createJob({x: 2, y: 3}).save();
  job.on('progress', (progress) => {
    console.log(
      `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
    );
  });
  
  addQueue.process(async (job) => {
    // do some work
    job.reportProgress({page: 3, totalPages: 11});
    // do more work
    job.reportProgress({page: 9, totalPages: 11});
    // do the rest
  });

  0 namespace is, for whatever reason, unavailable or problematic on your redis instance.

• const baseUrl = 'http://www.google.com/search?q=';
  subQueue.process(10, function (job, done) {
    http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
      // parse the difference out of the response...
      return done(null, difference);
    });
  });

  1: number, ms; the length of the window in which workers must report that they aren't stalling. Higher values will reduce Redis/network overhead, but if a worker stalls, it will take longer before its stalled job(s) will be retried. A higher value will also result in a lower probability of false-positives during stall detection.

• const job = addQueue.createJob({x: 2, y: 3}).save();
  job.on('progress', (progress) => {
    console.log(
      `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
    );
  });
  
  addQueue.process(async (job) => {
    // do some work
    job.reportProgress({page: 3, totalPages: 11});
    // do more work
    job.reportProgress({page: 9, totalPages: 11});
    // do the rest
  });

  2: number, ms; the window during which delayed jobs will be specifically scheduled using

  const job = addQueue.createJob({x: 2, y: 3}).save();
  job.on('progress', (progress) => {
    console.log(
      `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
    );
  });
  
  addQueue.process(async (job) => {
    // do some work
    job.reportProgress({page: 3, totalPages: 11});
    // do more work
    job.reportProgress({page: 9, totalPages: 11});
    // do the rest
  });

  3 - if all delayed jobs are further out than this window, the Queue will double-check that it hasn't missed any jobs after the window elapses.

• const job = addQueue.createJob({x: 2, y: 3}).save();
  job.on('progress', (progress) => {
    console.log(
      `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
    );
  });
  
  addQueue.process(async (job) => {
    // do some work
    job.reportProgress({page: 3, totalPages: 11});
    // do more work
    job.reportProgress({page: 9, totalPages: 11});
    // do the rest
  });

  4: number, ms; to avoid unnecessary churn for several jobs in short succession, the Queue may delay individual jobs by up to this amount.

• const job = addQueue.createJob({x: 2, y: 3}).save();
  job.on('progress', (progress) => {
    console.log(
      `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
    );
  });
  
  addQueue.process(async (job) => {
    // do some work
    job.reportProgress({page: 3, totalPages: 11});
    // do more work
    job.reportProgress({page: 9, totalPages: 11});
    // do the rest
  });

  5: object or string, specifies how to connect to Redis. See for the full set of options.

  • const job = addQueue.createJob({x: 2, y: 3}).save();
    job.on('progress', (progress) => {
      console.log(
        `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
      );
    });
    
    addQueue.process(async (job) => {
      // do some work
      job.reportProgress({page: 3, totalPages: 11});
      // do more work
      job.reportProgress({page: 9, totalPages: 11});
      // do the rest
    });

    7: string, Redis host.

  • const job = addQueue.createJob({x: 2, y: 3}).save();
    job.on('progress', (progress) => {
      console.log(
        `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
      );
    });
    
    addQueue.process(async (job) => {
      // do some work
      job.reportProgress({page: 3, totalPages: 11});
      // do more work
      job.reportProgress({page: 9, totalPages: 11});
      // do the rest
    });

    8: number, Redis port.

  • const job = addQueue.createJob({x: 2, y: 3}).save();
    job.on('progress', (progress) => {
      console.log(
        `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
      );
    });
    
    addQueue.process(async (job) => {
      // do some work
      job.reportProgress({page: 3, totalPages: 11});
      // do more work
      job.reportProgress({page: 9, totalPages: 11});
      // do the rest
    });

    9: string, Redis socket to be used instead of a host and port.

  Note that this can also be a node_redis

  const baseUrl = 'http://www.google.com/search?q=';
  subQueue.process(10, function (job, done) {
    http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
      // parse the difference out of the response...
      return done(null, difference);
    });
  });

  4 instance, in which case Bee-Queue will issue normal commands over it. It will

  const baseUrl = 'http://www.google.com/search?q=';
  subQueue.process(10, function (job, done) {
    http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
      // parse the difference out of the response...
      return done(null, difference);
    });
  });

  6 the client for blocking commands and PubSub subscriptions, if enabled. This is advanced usage,

• $ npm install bee-queue

  02: boolean. Disable if this queue will not process jobs.

• $ npm install bee-queue

  03: boolean. Disable if this queue does not need to receive job events.

• $ npm install bee-queue

  04: boolean. Disable if this worker does not need to send job events back to other queues.

• $ npm install bee-queue

  05: boolean. Disable if this worker does not need to associate events with specific

  $ npm install bee-queue

  06 instances. This normally improves memory usage, as the storage of jobs is unnecessary for many use-cases.

• $ npm install bee-queue

  07: boolean. Ensure that the Lua scripts exist in redis before running any commands against redis.

• $ npm install bee-queue

  08: boolean. Activate delayed jobs once they've passed their

  $ npm install bee-queue

  09 timestamp. Note that this must be enabled on at least one

  addQueue.process(function (job, done) {
    console.log(`Processing job ${job.id}`);
    return done(null, job.data.x + job.data.y);
  });

  4 instance for the delayed retry strategies (

  $ npm install bee-queue

  11 and

  $ npm install bee-queue

  12) - this will reactivate them after their computed delay.

• $ npm install bee-queue

  13: boolean. Enable to have this worker automatically remove its successfully completed jobs from Redis, so as to keep memory usage down.

• $ npm install bee-queue

  14: boolean. Enable to have this worker automatically remove its failed jobs from Redis, so as to keep memory usage down. This will not remove jobs that are set to retry unless they fail all their retries.

• $ npm install bee-queue

  15: boolean. Whether to

  $ npm install bee-queue

  16 the redis command client (the client it sends normal operations over) when

  $ npm install bee-queue

  17 is called. This defaults to

  $ npm install bee-queue

  18 for normal usage, and

  $ npm install bee-queue

  19 if an existing

  const baseUrl = 'http://www.google.com/search?q=';
  subQueue.process(10, function (job, done) {
    http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
      // parse the difference out of the response...
      return done(null, difference);
    });
  });

  4 object was provided to the

  const job = addQueue.createJob({x: 2, y: 3}).save();
  job.on('progress', (progress) => {
    console.log(
      `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
    );
  });
  
  addQueue.process(async (job) => {
    // do some work
    job.reportProgress({page: 3, totalPages: 11});
    // do more work
    job.reportProgress({page: 9, totalPages: 11});
    // do the rest
  });

  5 option.

• $ npm install bee-queue

  22: number. For setting the value of the

  $ npm install bee-queue

  23 Redis command used in

  $ npm install bee-queue

  24 for succeeded and failed job types.

Properties

• $ npm install bee-queue

  25: string, the name passed to the constructor.

• $ npm install bee-queue

  26: string, the prefix used for all Redis keys associated with this queue.

• $ npm install bee-queue

  27: a

  $ npm install bee-queue

  28 associating the currently tracked jobs (when

  $ npm install bee-queue

  05 and

  $ npm install bee-queue

  03 are enabled).

• $ npm install bee-queue

  31: boolean, whether the queue instance is paused. Only true if the queue is in the process of closing.

• const job = addQueue.createJob({x: 2, y: 3});
  job
    .timeout(3000)
    .retries(2)
    .save()
    .then((job) => {
      // job enqueued, job.id populated
    });

  5: object, the settings determined between those passed and the defaults

Queue Local Events

ready

Instead of listening to this event, consider calling

$ npm install bee-queue

$ npm install bee-queue

The queue has connected to Redis and ensured that the Lua scripts are cached. You can often get away without checking for this event, but it's a good idea to wait for it in case the Redis host didn't have the scripts cached beforehand; if you try to enqueue jobs when the scripts are not yet cached, you may run into a Redis error.

error

$ npm install bee-queue

Any Redis errors are re-emitted from the Queue. Note that this event will not be emitted for failed jobs.

succeeded

$ npm install bee-queue

This queue has successfully processed

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

retrying

$ npm install bee-queue

This queue has processed

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

failed

$ npm install bee-queue

This queue has processed

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

stalled

$ npm install bee-queue

This queue detected that a job . Note that this might not be the same queue instance that processed the job and ultimately stalled; instead, it's the queue instance that happened to detect the stalled job.

Queue PubSub Events

These events are all reported by some worker queue (with

$ npm install bee-queue

$ npm install bee-queue

If the

$ npm install bee-queue

Note that Queue PubSub events pass the

$ npm install bee-queue

job succeeded

$ npm install bee-queue

Some worker has successfully processed job

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

job retrying

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Some worker has processed job

$ npm install bee-queue

job failed

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Some worker has processed

$ npm install bee-queue

$ npm install bee-queue

job progress

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Some worker is processing job

$ npm install bee-queue

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Queue Delayed Job activation

The

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

$ npm install bee-queue

$ npm install bee-queue

The promptness of the job activation is controlled with the

const job = addQueue.createJob({x: 2, y: 3}).save();
job.on('progress', (progress) => {
  console.log(
    `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
  );
});

addQueue.process(async (job) => {
  // do some work
  job.reportProgress({page: 3, totalPages: 11});
  // do more work
  job.reportProgress({page: 9, totalPages: 11});
  // do the rest
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

const job = addQueue.createJob({x: 2, y: 3}).save();
job.on('progress', (progress) => {
  console.log(
    `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
  );
});

addQueue.process(async (job) => {
  // do some work
  job.reportProgress({page: 3, totalPages: 11});
  // do more work
  job.reportProgress({page: 9, totalPages: 11});
  // do the rest
});

$ npm install bee-queue

The

const job = addQueue.createJob({x: 2, y: 3}).save();
job.on('progress', (progress) => {
  console.log(
    `Job ${job.id} reported progress: page ${progress.page} / ${progress.totalPages}`
  );
});

addQueue.process(async (job) => {
  // do some work
  job.reportProgress({page: 3, totalPages: 11});
  // do more work
  job.reportProgress({page: 9, totalPages: 11});
  // do the rest
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

$ npm install bee-queue

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Methods

Queue(name, [settings])

Used to instantiate a new queue; opens connections to Redis.

Queue#createJob(data)

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Returns a new with the associated user data.

Queue#getJob(jobId, [cb])

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Looks up a job by its

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

Be careful with this method; most potential uses would be better served by job events on already-existing job instances. Using this method indiscriminately can lead to increasing memory usage when the

$ npm install bee-queue

$ npm install bee-queue

Queue#getJobs(type, page, [cb])

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Looks up jobs by their queue type. When looking up jobs of type

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

$ npm install bee-queue

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

$ npm install bee-queue

Note that large values of the attributes of

$ npm install bee-queue

Queue#process([concurrency], handler(job, done))

Begins processing jobs with the provided handler function.

The

$ npm install bee-queue

$ npm install bee-queue

The optional

$ npm install bee-queue

The handler function should either:

• Return a

  addQueue
    .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
    .then((errors) => {
      // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
    });

  7 that eventually resolves or rejects, or
• Call

  addQueue.process(function (job, done) {
    console.log(`Processing job ${job.id}`);
    return done(null, job.data.x + job.data.y);
  });

  2 exactly once
  • Use

    $ npm install bee-queue

    41 to indicate job failure
  • Use

    $ npm install bee-queue

    88 or

    $ npm install bee-queue

    36 to indicate job success

    • $ npm install bee-queue

      35 must be JSON-serializable (for

      $ npm install bee-queue

      91)
• Never ever block the event loop (for very long). If you do, the stall detection might think the job stalled, when it was really just blocking the event loop.

N.B. If the handler returns a

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Queue#checkStalledJobs([interval], [cb])

Checks for jobs that appear to be stalling and thus need to be retried, then re-enqueues them.

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

What happens after the check is determined by the parameters provided:

• $ npm install bee-queue

  94 only:

  $ npm install bee-queue

  94 is called

• $ npm install bee-queue

  96 only: a timeout is set to call the method again in

  $ npm install bee-queue

  96 ms

• $ npm install bee-queue

  94 and

  $ npm install bee-queue

  96: a timeout is set, then

  $ npm install bee-queue

  94 is called

Bee-Queue automatically calls this method once when a worker begins processing, so it will check once if a worker process restarts. You should also make your own call with an interval parameter to make the check happen repeatedly over time; see for an explanation why.

The maximum delay from when a job stalls until it will be retried is roughly

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

$ npm install bee-queue

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

Note that for calls that specify an interval, you must provide a callback if you want results from each subsequent check - the returned

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

$ npm install bee-queue

$ npm install bee-queue

addQueue.process(async (job) => {
  console.log(`Processing job ${job.id}`);
  return job.data.x + job.data.y;
});

Queue#checkHealth([cb])

Check the "health" of the queue. Returns a promise that resolves to the number of jobs in each state (

$ npm install bee-queue

$ npm install bee-queue

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

$ npm install bee-queue

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

$ npm install bee-queue

$ npm install bee-queue

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Queue#close([timeout], [cb])

Closes the queue's connections to Redis. Idempotent.

The recommended pattern for gracefully shutting down your worker is:

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Queue#isRunning()

Returns

$ npm install bee-queue

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Queue#ready([cb])

Promise resolves to the queue (or callback is called wth

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Queue#removeJob(jobId, [cb])

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Removes a job by its

$ npm install bee-queue

This may have unintended side-effect, e.g. if the job is currently being processed by another worker, so only use this method when you know it's safe.

Returns the

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Queue#destroy([cb])

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Removes all Redis keys belonging to this queue (see ). Idempotent.

It goes without saying that this should be used with great care.

Returns the number of keys removed.

Job

Properties

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  24: string, Job ID unique to each job. Not populated until

  const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  25 calls back. Can be overridden with

  const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  26.

• const job = addQueue.createJob({x: 2, y: 3});
  job
    .timeout(3000)
    .retries(2)
    .save()
    .then((job) => {
      // job enqueued, job.id populated
    });

  7: object; user data associated with the job. It should:
  • Be JSON-serializable (for

    $ npm install bee-queue

    91)
  • Never be used to pass large pieces of data (100kB+)
  • Ideally be as small as possible (1kB or less)

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  29: object used by Bee-Queue to store timeout, retries, stack traces, etc.
  • Do not modify directly; use job methods instead.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  30: the Queue responsible for this instance of the job. This is either:
  • the queue that called

    const Queue = require('bee-queue');
    const addQueue = new Queue('addition');

    31 to make the job,
  • the queue that ran

    const Queue = require('bee-queue');
    const addQueue = new Queue('addition');

    32 to fetch the job from redis, or
  • the queue that called

    $ npm install bee-queue

    82 to process it

• addQueue.process(function (job, done) {
    console.log(`Processing job ${job.id}`);
    return done(null, job.data.x + job.data.y);
  });

  7: number; progress between 0 and 100, as reported by

  const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  35.

Job Events

These are all Pub/Sub events like and are disabled when

$ npm install bee-queue

succeeded

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

The job has succeeded. If

$ npm install bee-queue

$ npm install bee-queue

retrying

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

The job has failed, but it is being automatically re-enqueued for another attempt.

$ npm install bee-queue

failed

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

The job has failed, and is not being retried.

progress

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

The job has sent a of

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Methods

Each Job can be configured with the chainable commands

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

addQueue
  .saveAll([addQueue.createJob({x: 3, y: 4}), addQueue.createJob({x: 4, y: 5})])
  .then((errors) => {
    // The errors value is a Map associating Jobs with Errors. This will often be an empty Map.
  });

Job#setId(id)

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Explicitly sets the ID of the job. If a job with the given ID already exists, the Job will not be created, and

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Avoid passing a numeric job ID, as it may conflict with an auto-generated ID.

Job#retries(n)

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Sets how many times the job should be automatically retried in case of failure.

Stored in

$ npm install bee-queue

Defaults to 0.

Job#backoff(strategy, delayFactor)

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Sets the backoff policy when handling retries.

This setting is stored in

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Defaults to

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Job#delayUntil(date|timestamp)

const subQueue = new Queue('subtraction', {
  redis: {
    host: 'somewhereElse',
  },
  isWorker: false,
});

Delay the job until the given Date/timestamp passes. See the

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Defaults to enqueueing the job for immediate processing.

Job#timeout(ms)

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Sets a job runtime timeout in milliseconds; if the job's handler function takes longer than the timeout to call

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Defaults to no timeout.

Job#save([cb])

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Saves a job, queueing it up for processing. After the callback fires (and associated Promise resolves),

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Job#reportProgress(n)

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Reports job progress when called within a handler function. Causes a

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

$ npm install bee-queue

$ npm install bee-queue

addQueue.process(function (job, done) {
  console.log(`Processing job ${job.id}`);
  return done(null, job.data.x + job.data.y);
});

Job#remove([cb])

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

Removes a job from the queue. Idempotent.

This may have unintended side-effect, e.g. if the job is currently being processed by another worker, so only use this method when you know it's safe.

Note that this method will call with the job id, so if you don't have the job in memory, but knows its id, it's much more efficient to use

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Returns the

$ npm install bee-queue

Defaults

Defaults for Queue

const job = addQueue.createJob({x: 2, y: 3});
job
  .timeout(3000)
  .retries(2)
  .save()
  .then((job) => {
    // job enqueued, job.id populated
  });

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Each Queue uses the following Redis keys:

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  68: Integer, incremented to determine the next Job ID.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  69: Hash from Job ID to a JSON string containing its data and options.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  70: List of IDs of jobs waiting to be processed.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  71: List of IDs jobs currently being processed.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  72: Set of IDs of jobs which succeeded.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  73: Set of IDs of jobs which failed.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  74: Ordered Set of IDs corresponding to delayed jobs - this set maps delayed timestamp to IDs.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  75: Set of IDs of jobs which haven't 'checked in' during this interval.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  76: Set of IDs of jobs which haven't 'checked in' during this interval.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  77: Pub/Sub channel for workers to send out job results.

• const Queue = require('bee-queue');
  const addQueue = new Queue('addition');

  78: When a new delayed job is added prior to all other jobs, the script creating the job will publish the job's timestamp over this Pub/Sub channel.

Bee-Queue is non-polling, so idle workers are listening to receive jobs as soon as they're enqueued to Redis. This is powered by brpoplpush, which is used to move jobs from the waiting list to the active list. Bee-Queue generally follows the "Reliable Queue" pattern described here.

The

$ npm install bee-queue

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

$ npm install bee-queue

$ npm install bee-queue

The stalling set is a snapshot of the active list from the beginning of the latest stall interval. During each stalling interval, workers remove their job IDs from the stalling set, so at the end of an interval, any jobs whose IDs are left in the stalling set have missed their window (stalled) and need to be rerun. When

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

Bee-Queue requires the user to start the repeated checks on their own because if we did it automatically, every queue instance in the system would be doing the check. Checking from all instances is less efficient and provides weaker guarantees than just checking from one or two. For example, a

const baseUrl = 'http://www.google.com/search?q=';
subQueue.process(10, function (job, done) {
  http.get(`${baseUrl}${job.data.x}-${job.data.y}`, function (res) {
    // parse the difference out of the response...
    return done(null, difference);
  });
});

Pull requests are welcome; just make sure

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Some significant non-features include:

• Worker tracking: Kue does this.
• All-workers pause-resume: Bull does this.
• Job priority: multiple queues get the job done in simple cases, but Kue has first-class support. Bull provides a wrapper around multiple queues.

Some of these could be worthwhile additions; please comment if you're interested in using or helping implement them!

Testing

You'll need a local Redis server to run the tests. Note that running the tests may delete some keys in the form of

const Queue = require('bee-queue');
const addQueue = new Queue('addition');

Alternatively, if you have Docker available, you can run tests or do forensic work in an ephemeral container with its own Redis server, e.g.:

Node.js digunakan untuk apa saja?

Node.js menggunakan bahasa apa?