bull 源碼閱讀筆記

是一個(gè)基于Redis的強(qiáng)大的Nodejs隊(duì)列框架
git地址：https://github.com/OptimalBits/bull
下圖是其與其他隊(duì)列庫的對(duì)比圖涌韩，可以看出Bull支持優(yōu)先度，并發(fā)不瓶，延遲任務(wù)钠四，全局事件甘改，頻率限制，重復(fù)任務(wù)美尸，原子操作鞠柄，可視化頁面等功能。

image.png

三個(gè)角色

一個(gè)隊(duì)列Queue有三個(gè)角色：任務(wù)生產(chǎn)者械馆，任務(wù)消費(fèi)者胖眷，事件監(jiān)聽者
一般生產(chǎn)者和消費(fèi)者被分為不同的實(shí)例，生產(chǎn)者可以在沒有可用消費(fèi)者的情況下霹崎，依然可以往隊(duì)列中加入任務(wù)珊搀，Queue提供了異步通信。
另外可以讓一個(gè)或者多個(gè)消費(fèi)者從隊(duì)列中消費(fèi)任務(wù)尾菇。
worker,可以在相同或者不同的進(jìn)程境析，同一臺(tái)或者集群中運(yùn)行。
Redis作為一個(gè)中間者派诬，只要生產(chǎn)者消費(fèi)之可以連接到Redis劳淆，他們就可以分工合作處理任務(wù)

一個(gè)任務(wù)的生命周期

當(dāng)調(diào)用隊(duì)列的 add 的方法時(shí)，任務(wù)將會(huì)有一個(gè)生命周期默赂，直到運(yùn)行成功或者失斉嫱摇（失敗的任務(wù)將會(huì)被重試，將會(huì)有一個(gè)新的生命周期）缆八，如下所示

image.png

當(dāng)任務(wù)被加到隊(duì)列后將會(huì)是 Wait（等待被處理）或者Delayed狀態(tài) （延遲被處理）曲掰。Delayed狀態(tài)的任務(wù)不會(huì)直接被處理，而是到指定時(shí)間后奈辰，排到wait隊(duì)列后栏妖，等到worker空閑后處理
接下來是 Active 狀態(tài)，表示任務(wù)正在被處理奖恰。當(dāng)處理完成任務(wù)就會(huì)進(jìn)入 completed 狀態(tài)吊趾，或者完成失敗進(jìn)入 Fail 狀態(tài)。
任務(wù)在被Active激活的時(shí)候會(huì)獲取鎖(防止一個(gè)任務(wù)被多個(gè)消費(fèi)者消費(fèi)),completed時(shí)候會(huì)釋放鎖瑟啃。但是鎖有過期時(shí)間（lockDuration）

producer 任務(wù)生產(chǎn)者

生產(chǎn)者就是趾徽，往隊(duì)列里面增加任務(wù)，應(yīng)用示例如下：

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

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

先看 queue.add翰守，不管是單個(gè)任務(wù)還是重復(fù)性任務(wù)孵奶，最終都都調(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);  // 后面也是增加了一個(gè)延時(shí)的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 隊(duì)列蜡峰，還 publish 進(jìn)行發(fā)布事件

增加任務(wù)后了袁，根據(jù)lua,redis會(huì)增加相應(yīng)的值, 并發(fā)布消息朗恳。

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

單項(xiàng)任務(wù)如下所示：
會(huì)增加一個(gè)redis值载绿，類型為hash粥诫，表示一個(gè)任務(wù)
key = redis鍵前綴: 隊(duì)列名字：編號(hào)
value = 相應(yīng)任務(wù)配置

image.png

重復(fù)任務(wù)如下所示,會(huì)修改redis兩個(gè)值
1： repeat隊(duì)列，類型為zset崭庸，表示可重復(fù)任務(wù)隊(duì)列
key = redis鍵前綴: 隊(duì)列名字：'repeat'
value = 各個(gè)任務(wù)怀浆，score為相應(yīng)任務(wù)的下一次執(zhí)行的時(shí)間戳

image.png

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

image.png

consumer 任務(wù)消費(fèi)者

是指處理隊(duì)列中的定時(shí)任務(wù)怕享，應(yīng)用示例如下：

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

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

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

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

queue.process 先看process當(dāng)中主要調(diào)用了 setHandler(綁定handler) _initProcess（注冊(cè)監(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 將自定義的任務(wù)處理方法綁定至監(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') {
    // 當(dāng)處理方法定義在另一個(gè)文件中
    ...
  } 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执赡，為注冊(cè)監(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，如下其實(shí)質(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來獲取消息函筋，超時(shí)時(shí)間為 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去處理該任務(wù)

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后有重復(fù)執(zhí)行processJobs, 就是一個(gè)循環(huán)，如下圖：

image.png

Listeners 監(jiān)聽者

應(yīng)用示例如下：

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

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