BullMQ：基于Redis的Node.js消息队列库的使用与功能介绍

BullMQ 是一个基于 Redis 的 Node.js 消息队列库，用于处理后台任务和作业调度。它提供了强大的功能，如任务队列、延迟任务、重试机制、并发控制等。BullMQ 是 Bull 的继任者，提供了更好的性能和更现代的 API。

安装

首先，你需要安装 BullMQ 和 Redis 客户端库：

npm install bullmq ioredis

基本用法

1. 创建队列

首先，你需要创建一个队列。队列是 BullMQ 的核心概念，用于存储和管理任务。

const { Queue } = require('bullmq');
const { Redis } = require('ioredis');

const redisConnection = new Redis({
  host: 'localhost',
  port: 6379,
});

const queue = new Queue('myQueue', { connection: redisConnection });

2. 添加任务

你可以向队列中添加任务。任务是一个 JavaScript 对象，可以包含任意数据。

async function addJob() {
  await queue.add('myJob', { foo: 'bar' });
  console.log('Job added to the queue');
}

addJob();

3. 处理任务

你需要创建一个工作进程来处理队列中的任务。

const { Worker } = require('bullmq');

const worker = new Worker('myQueue', async job => {
  console.log('Processing job:', job.id);
  console.log('Data:', job.data);
  // 在这里处理任务
}, { connection: redisConnection });

worker.on('completed', job => {
  console.log(`Job ${job.id} completed`);
});

worker.on('failed', (job, err) => {
  console.error(`Job ${job.id} failed with error:`, err);
});

4. 延迟任务

你可以添加延迟任务，任务会在指定的时间后执行。

async function addDelayedJob() {
  await queue.add('myDelayedJob', { foo: 'bar' }, { delay: 5000 }); // 5秒后执行
  console.log('Delayed job added to the queue');
}

addDelayedJob();

5. 重试机制

BullMQ 支持自动重试失败的任务。

async function addRetryJob() {
  await queue.add('myRetryJob', { foo: 'bar' }, {
    attempts: 3, // 最多重试3次
    backoff: {
      type: 'exponential', // 指数退避
      delay: 1000, // 初始延迟1秒
    },
  });
  console.log('Retry job added to the queue');
}

addRetryJob();

6. 并发控制

你可以控制工作进程的并发数。

const worker = new Worker('myQueue', async job => {
  console.log('Processing job:', job.id);
  // 在这里处理任务
}, {
  connection: redisConnection,
  concurrency: 5, // 最多同时处理5个任务
});

高级功能

1. 事件监听

BullMQ 提供了丰富的事件监听功能，你可以监听队列和工作进程的各种事件。

queue.on('waiting', jobId => {
  console.log(`Job ${jobId} is waiting`);
});

queue.on('active', job => {
  console.log(`Job ${job.id} is now active`);
});

queue.on('completed', job => {
  console.log(`Job ${job.id} has completed`);
});

queue.on('failed', (job, err) => {
  console.error(`Job ${job.id} has failed with error:`, err);
});

2. 队列统计

你可以获取队列的统计信息，如等待中的任务数、已完成的任务数等。

async function getQueueStats() {
  const counts = await queue.getJobCounts();
  console.log('Queue stats:', counts);
}

getQueueStats();

3. 清理队列

你可以清理队列中的已完成或失败的任务。

async function cleanQueue() {
  await queue.clean(1000, 'completed'); // 清理1000个已完成的任务
  await queue.clean(1000, 'failed'); // 清理1000个失败的任务
  console.log('Queue cleaned');
}

cleanQueue();

总结

BullMQ 是一个功能强大且灵活的消息队列库，适用于处理后台任务和作业调度。通过 Redis 作为后端存储，BullMQ 提供了高性能和可靠性。你可以根据需求使用 BullMQ 的各种功能，如延迟任务、重试机制、并发控制等，来构建高效的后台任务处理系统。