BullMQ 初体验 | 月空人

import { Queue } from 'bullmq';

const myQueue = new Queue('foo');

async function addJobs() {
    await myQueue.add('myJobName', { foo: 'bar' });
    await myQueue.add('myJobName', { qux: 'baz' }, { delay: 5000 }); // 延迟作业，该任务将至少等待 5 秒钟才会进行处理
}

await addJobs();

消费

import { Worker } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis({ maxRetriesPerRequest: null });

const worker = new Worker(
    'foo',
    async (job) => {
    // 先打印  { foo: 'bar'}
    // 然后打印 { qux: 'baz' }
        console.log(job.data);
    },
    { connection },
);

监听任务状态

// 成功
worker.on('completed', (job) => {
    console.log(`${job.id} has completed!`);
});

// 失败
worker.on('failed', (job, err) => {
    console.log(`${job.id} has failed with ${err.message}`);
});

自动删除已完成的任务

await myQueue.add(
    'test',
    { foo: 'bar' },
    { removeOnComplete: true, removeOnFail: true }, // 成功和失败的任务都删除
);

添加任务是异步的，我们可以使用 addBulk 进行批量添加

import { Queue } from 'bullmq';

const queue = new Queue('paint');

const name = 'jobName';
const jobs = await queue.addBulk([
    { name, data: { paint: 'car' } },
    { name, data: { paint: 'house' } },
    { name, data: { paint: 'boat' } },
]);

速率限制

我们可以配置在一段时间内处理多少个作业

全局

import { Queue } from 'bullmq';

// 1 秒处理一个
await queue.setGlobalRateLimit(1, 1000);

单个任务

import { QueueScheduler, Worker } from 'bullmq';

const worker = new Worker('painter',
    async job => paintCar(job),
    {
        limiter: {
            max: 10,
            duration: 1000,
        },
    });

在失败时重试

什么情况下 BullMQ 认为是失败了

Worker 中抛出异常
Worker 陷入停滞

如何配置失败时重试？

固定重试3次，每次延迟3秒

import { Queue } from 'bullmq';

const myQueue = new Queue('foo');

await queue.add(
    'test-retry',
    { foo: 'bar' },
    {
        attempts: 3,
        backoff: {
            type: 'fixed',
            delay: 1000,
        },
    },
);

可以配置 jitter 来实现随机抖动。延迟时间会在 1000ms ~ 1000 * 0.5 ms 之间随机

import { Queue } from 'bullmq';

const myQueue = new Queue('foo');

await queue.add(
    'test-retry',
    { foo: 'bar' },
    {
        attempts: 8,
        backoff: {
            type: 'fixed',
            delay: 1000,
            jitter: 0.5,
        },
    },
);

扩展

使用 bull-board 监控和管理

bull-board

它可以结合很多 node 框架使用。比如有常见的 express, fastify, koa, nestjs, hono, h3, elysia 等

我这几天看 hono 有点多，这里使用 hono 扩展。

text

pnpm install @bull-board/hono

routes/admin/queue.tsts

import { createBullBoard } from '@bull-board/api'
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter'
import { HonoAdapter } from '@bull-board/hono'
import { serveStatic } from '@hono/node-server/serve-static'

import { emailQueue } from '@/queues/email.queue'

const serverAdapter = new HonoAdapter(serveStatic)

serverAdapter.setBasePath('/admin/queues')

createBullBoard({ queues: [new BullMQAdapter(emailQueue)], serverAdapter })

export const queueUI = serverAdapter.registerPlugin()

app.tsts

import { createApp } from '@/lib/create-app'
import { queueUI } from '@/routes/admin/queue'

const app = createApp()

app.route('/admin/queues', queueUI)

export { app }