是一個基于Redis的強大的Nodejs隊列框架
git地址：https://github.com/OptimalBits/bull
下圖是其與其他隊列庫的對比圖，可以看出Bull支持優(yōu)先度，并發(fā)，延遲任務，全局事件，頻率限制，重復任務，原子操作，可視化頁面等功能。

image.png

三個角色

一個隊列Queue有三個角色：任務生產(chǎn)者， 任務消費者， 事件監(jiān)聽者
一般生產(chǎn)者和消費者被分為不同的實例，生產(chǎn)者可以在沒有可用消費者的情況下，依然可以往隊列中加入任務，Queue提供了異步通信。
另外可以讓一個或者多個消費者從隊列中消費任務。
worker,可以在相同或者不同的進程，同一臺或者集群中運行。
Redis作為一個中間者，只要生產(chǎn)者消費之可以連接到Redis，他們就可以分工合作處理任務

一個任務的生命周期

當調(diào)用隊列的 add 的方法時，任務將會有一個生命周期，直到運行成功或者失?。ㄊ〉娜蝿諏恢卦?，將會有一個新的生命周期），如下所示

image.png

當任務被加到隊列后將會是 Wait（等待被處理）或者Delayed狀態(tài) （延遲被處理）。Delayed狀態(tài)的任務不會直接被處理，而是到指定時間后，排到wait隊列后，等到worker空閑后處理
接下來是 Active 狀態(tài)，表示任務正在被處理。當處理完成任務就會進入 completed 狀態(tài)，或者完成失敗進入 Fail 狀態(tài)。
任務在被Active激活的時候會獲取鎖(防止一個任務被多個消費者消費),completed時候會釋放鎖。但是鎖有過期時間（lockDuration）

producer 任務生產(chǎn)者

生產(chǎn)者就是，往隊列里面增加任務，應用示例如下：

// 1.創(chuàng)建隊列
const myFirstQueue = new Bull('my-first-queue');
// 2.往隊列增加任務
const job = await myFirstQueue.add({
  foo: 'bar'
});

底層代碼內(nèi)容包括以下幾個步驟：
1 創(chuàng)建隊列
2 增加任務

• 先看 queue.add，不管是單個任務還是重復性任務，最終都都調(diào)用了Job.create

Queue.prototype.add = function(name, data, opts) {
 ...
  opts = _.cloneDeep(opts || {});
  _.defaults(opts, this.defaultJobOptions);

  if (opts.repeat) {
    return this.isReady().then(() => {
      return this.nextRepeatableJob(name, data, opts, true);  // 后面也是增加了一個延時的job, Job.create(....)
    });
  } else {
    return Job.create(this, name, data, opts);
  }
};

• job.create 調(diào)用了 addJob

Job.create = function(queue, name, data, opts) {
  const job = new Job(queue, name, data, opts);

  return queue
    .isReady()
    .then(() => {
      return addJob(queue, queue.client, job);
    })
    .then(jobId => {
      job.id = jobId;
      debuglog('Job added', jobId);
      return job;
    });
};

• addJob里面是調(diào)用了lua腳本， 增加wait 以及 delayed 隊列，還 publish 進行發(fā)布事件

增加任務后，根據(jù)lua,redis會增加相應的值, 并發(fā)布消息。

• 任務增加成功后，wait 隊列以及 delayed 隊列會增加數(shù)據(jù)

單項任務如下所示：
會增加一個redis值，類型為hash，表示一個任務
key = redis鍵前綴: 隊列名字： 編號
value = 相應任務配置

image.png

重復任務如下所示,會修改redis兩個值
1： repeat隊列，類型為zset，表示可重復任務隊列
key = redis鍵前綴: 隊列名字：'repeat'
value = 各個任務，score為相應任務的下一次執(zhí)行的時間戳

image.png

2 類型hash，表示單個任務的具體信息:
redis鍵前綴：隊列名字：'repeat:' + md5(name + jobId + namespace) + ':' + nextMillis

image.png

consumer 任務消費者

是指處理隊列中的定時任務，應用示例如下：

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

myFirstQueue.process(async (job) => {
  return doSomething(job.data);
});

process 方法會在有任務要執(zhí)行且worker空閑的時候被調(diào)用

底層代碼內(nèi)容包括以下幾個步驟：
1 綁定處理方法handler
2 增加任務

• queue.process 先看process當中主要調(diào)用了 setHandler(綁定handler) _initProcess（注冊監(jiān)聽） start

Queue.prototype.process = function(name, concurrency, handler) {
  switch (arguments.length) {
    case 1:
      handler = name;
      concurrency = 1;
      name = Job.DEFAULT_JOB_NAME;
      break;
    case 2: // (string, function) or (string, string) or (number, function) or (number, string)
      handler = concurrency;
      if (typeof name === 'string') {
        concurrency = 1;
      } else {
        concurrency = name;
        name = Job.DEFAULT_JOB_NAME;
      }
      break;
  }

  this.setHandler(name, handler);

  return this._initProcess().then(() => {
    return this.start(concurrency);
  });
};

• 再看setHandler 將自定義的任務處理方法綁定至監(jiān)聽事件中

Queue.prototype.setHandler = function(name, handler) {
  if (!handler) {
    throw new Error('Cannot set an undefined handler');
  }
  if (this.handlers[name]) {
    throw new Error('Cannot define the same handler twice ' + name);
  }

  this.setWorkerName();

  if (typeof handler === 'string') {
    // 當處理方法定義在另一個文件中
    ...
  } else {
    handler = handler.bind(this);

    if (handler.length > 1) {
      this.handlers[name] = promisify(handler);
    } else {
      this.handlers[name] = function() {
        try {
          return Promise.resolve(handler.apply(null, arguments));
        } catch (err) {
          return Promise.reject(err);
        }
      };
    }
  }
};

• 再看 _initProcess， 為注冊監(jiān)聽

Queue.prototype._initProcess = function() {
  if (!this._initializingProcess) {
    this.delayedTimestamp = Number.MAX_VALUE;
    this._initializingProcess = this.isReady()  
      .then(() => {
        return this._registerEvent('delayed');
      })
      .then(() => {
        return this.updateDelayTimer();
      });

    this.errorRetryTimer = {};
  }

  return this._initializingProcess;
};

• start方法調(diào)用了run，如下其實質(zhì)是調(diào)用了processJobs

Queue.prototype.run = function(concurrency) {
  const promises = [];

  return this.isReady()
    .then(() => {
      return this.moveUnlockedJobsToWait();
    })
    .then(() => {
      return utils.isRedisReady(this.bclient);
    })
    .then(() => {
      while (concurrency--) {
        promises.push(
          new Promise(resolve => {
            this.processJobs(concurrency, resolve);
          })
        );
      }

      this.startMoveUnlockedJobsToWait();

      return Promise.all(promises);
    });
};

• processJobs 調(diào)用 _processJobOnNextTick

Queue.prototype.processJobs = function(index, resolve, job) {
  const processJobs = this.processJobs.bind(this, index, resolve);
  process.nextTick(() => {
    this._processJobOnNextTick(processJobs, index, resolve, job);
  });
};

• _processJobOnNextTick 調(diào)用 processJob

Queue.prototype._processJobOnNextTick = function(
  processJobs,
  index,
  resolve,
  job
) {
  if (!this.closing) {
    (this.paused || Promise.resolve())
      .then(() => {
        const gettingNextJob = job ? Promise.resolve(job) : this.getNextJob();  // 獲取job

        return (this.processing[index] = gettingNextJob
          .then(this.processJob)                                        // 調(diào)用processJob
          .then(processJobs, err => {
            this.emit('error', err, 'Error processing job');
            clearTimeout(this.errorRetryTimer[index]);
            this.errorRetryTimer[index] = setTimeout(() => {
              processJobs();
            }, this.settings.retryProcessDelay);

            return null;
          }));
      })
      .catch(err => {
        this.emit('error', err, 'Error processing job');
      });
  } else {
    resolve(this.closing);
  }
};

• 再看看getNextJob, 利用brpoplpush來獲取消息，超時時間為 DrainDelay

Queue.prototype.getNextJob = function() {
  if (this.closing) {
    return Promise.resolve();
  }

  if (this.drained) {
    //
    // Waiting for new jobs to arrive
    //
    return this.bclient
      .brpoplpush(this.keys.wait, this.keys.active, this.settings.drainDelay)
      .then(
        jobId => {
          if (jobId) {
            return this.moveToActive(jobId);
          }
        },
        err => {
          // Swallow error
          if (err.message !== 'Connection is closed.') {
            console.error('BRPOPLPUSH', err);
          }
        }
      );
  } else {
    return this.moveToActive();
  }
};

• 再看 processJob
  根據(jù)事件循環(huán)獲取事件，并用綁定的handler去處理該任務

Queue.prototype.processJob = function(job, notFetch = false) {
  let lockRenewId;
  let timerStopped = false;

  if (!job) {
    return Promise.resolve();
  }
  ...
  const handleCompleted = result => {
    return job.moveToCompleted(result, undefined, notFetch).then(jobData => {
      this.emit('completed', job, result, 'active');
      return jobData ? this.nextJobFromJobData(jobData[0], jobData[1]) : null;
    });
  };

  const handleFailed = err => {
    const error = err;

    return job.moveToFailed(err).then(jobData => {
      this.emit('failed', job, error, 'active');
      return jobData ? this.nextJobFromJobData(jobData[0], jobData[1]) : null;
    });
  };

  lockExtender();
  const handler = this.handlers[job.name] || this.handlers['*'];

  if (!handler) {
    return handleFailed(
      new Error('Missing process handler for job type ' + job.name)
    );
  } else {
    let jobPromise = handler(job);     // 調(diào)用綁定好的handler

    if (timeoutMs) {
      jobPromise = pTimeout(jobPromise, timeoutMs);
    }

    // Local event with jobPromise so that we can cancel job.
    this.emit('active', job, jobPromise, 'waiting');

    return jobPromise
      .then(handleCompleted)
      .catch(handleFailed)
      .finally(() => {
        stopTimer();
      });
  }
};

在執(zhí)行完processJob后有重復執(zhí)行processJobs, 就是一個循環(huán)，如下圖：

image.png

Listeners 監(jiān)聽者

應用示例如下：

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

// Define a local completed event
myFirstQueue.on('completed', (job, result) => {
  console.log(`Job completed with result ${result}`);
})