[大数据]node.js -redis -bull

初始化queue

// bullTest.js
'use strict';
const Queue = require('bull');
const queue = new Queue('nike', {
  // redis数据库
  redis: {
    port: 6379,
    host: '127.0.0.1',
    db: 4,
    password: null,
  },
  prefix: 'nike_', // 队列文件
  defaultJobOptions: {
    attempts: 1,
    removeOnComplete: true,
    backoff: false,
    delay: 0,
  },
  // sequelize
  // 1000 ms 最多20000 个作业
  limiter: {
    max: 20000,
    duration: 1000,
  },
  settings: {
    maxStalledCount: 1,
    guardInterval: 1, // 重新调度延迟
    retryProcessDelay: 500, // delay before processing next job in case of internal error.
    // drainDelay: 50000 // 空队列时brpoplpush的等待时间
  },
});
module.exports = queue;

队列基本使用

// test.js
'use strict';
const queue = require('./bullTest');
const random = require('random-string');
const log4js = require('log4js');
const logger = log4js.getLogger();
logger.level = 'info';
queue.on('global:progress', function (jobId, progress) {
  logger.info(`Job ${jobId} is ${progress * 100}% ready!`);
});
queue.on('global:completed', jobId => {
  logger.info(`global:completed Job with id ${jobId} has been completed`);
});
queue.process((job, done) => {
  console.log('消耗-----', job.data);
  done();
});
// queue.on('completed', job => {
//   console.log(`================Job with id ${job.id} has been completed`);
// });
const main = async () => {
  for (let i = 0; i < 7; i++) {
    const job = await queue.add({
      key: random(10),
    }, {
      delay: (3 - i) * (3 - i) * 1000 + 10000, // 表示添加进来的数据延迟多长时间做
      // repeat: {
      //   every: 1000,
      //   limit: 3,
      // },
    });
    logger.info('生产者:', job.data, await queue.count());
  }
  // await queue.add({
  //   key: '154964',
  // }, {
  //   // 重复执行 三次，每次间隔10000
  //   repeat: {
  //     every: 10000,
  //     limit: 3,
  //   },
  // });
  // for (let i = 0; i < 17; i++) {
  //   const job = await queue.add({
  //     key: random(10),
  //   }, {
  //     delay: 5000,
  //   });
  //   logger.info('生产者:', job.data, await queue.count());
  // }
};
main();
// lifo true表示后进先出
// queue.add({foo:'bar'}, {lifo:true});

// 运行
node test.js