@ better-bull-board/客户端
开发者将Better Bulk Board 与您的 BullMQ 工人整合的文件 .
安装
pnpm add @better-bull-board/client bullmq
bullmq 属于对等关系( e)^6) (中文(简体) ). 安装您的应用程序使用的版本; 它不需要匹配此软件包的针 .
使用量
要使用更好的公牛板 与你的公牛MQ工人, 你需要:
- 导入 BBB 客户端
- 使用单独的处理器 (取消职务所需的散装箱处理器)
基本设置
1. 创建处理器文件
创建一个单独的处理器文件(例如, processor.ts 或者说 processor.js:
import { patch } from "@better-bull-board/client"
import type { SandboxedJob } from "bullmq"
import { redis } from "./lib/redis" // Your Redis connection
export default patch(async (job: SandboxedJob) => {
console.log(`Processing job ${job.id}`)
// Your job processing logic here
await processYourJob(job.data)
console.log(`Job ${job.id} completed`)
return { status: "done" }
}, redis)
2. 创建工人
使用您主文档中更好的公牛板工人班:
import path from "node:path"
import { Worker } from "@better-bull-board/client"
import { redis } from "./lib/redis" // Your Redis connection
const processorFile = path.join(__dirname, "processor.cjs") // Note: .cjs extension for built files
new Worker("your-queue-name", processorFile, {
connection: redis,
ioredis: redis,
useWorkerThreads: true,
concurrency: 10,
getJobTags(job) {
// Optional: Return tags for better organization
return ["your-queue", "production"]
},
})
关键要求
- 单独的处理器: 您必须使用沙盒处理器( 分离文件) 才能正常工作
- Redis 连接: 提供两者
connection财务报告和审定财务报表ioredis选项指向相同的 Redis 实例 - 工人线索设置
useWorkerThreads: true适当隔离 - 文件扩展名: 使用
.cjs引用已建文件时处理器文件路径的扩展名
完整示例
这里有一个完整的工作例子 依据我们的演示:
处理器.ts
import { patch } from "@better-bull-board/client"
import type { SandboxedJob } from "bullmq"
import { redis } from "../lib/redis"
export default patch(async (job: SandboxedJob) => {
console.log(`Processing job ${job.id}`)
// Simulate work
await new Promise((resolve) => setTimeout(resolve, 10_000))
console.log(`Job ${job.id} processed`)
return { status: "done" }
}, redis)
工 作 员
import path from "node:path"
import { fileURLToPath } from "node:url"
import { Worker } from "@better-bull-board/client"
import { redis } from "./lib/redis"
const __filename = fileURLToPath(import.meta.url)
const __dirname = path.dirname(__filename)
const processorFile = path.join(__dirname, "processor.cjs")
new Worker("demo-queue", processorFile, {
connection: redis,
ioredis: redis,
useWorkerThreads: true,
concurrency: 10,
getJobTags() {
return ["demo-queue", "test"]
},
})
高级特性
任务取消
这个 patch 函数提供内置任务取消支持。 工作可以通过Better Bulk Board UI取消.
工作日志
处理器的所有控制台输出都会自动被捕获并被送到Better Bull Board进行监控.
直接写出字的伐木工,比如 皮诺绕行 console。给 Pino 工作日志目标,所以每行也到达当前任务 :
import { createPinoJobLogDestination } from "@better-bull-board/client"
import { pino } from "pino"
export const logger = pino(
{ level: "info" },
pino.multistream([
{ stream: process.stdout, level: "trace" },
{ stream: createPinoJobLogDestination({ ignore: ["trace_id"] }), level: "trace" },
]),
)
目的地保留 msg,将其他字段作为 JSON 和打印 err 犹如被抛弃的迷误一样code, (中文(简体) ), data(......) ; (......); ignore 在工作日志中丢弃您不想要的密钥( r)pid 财务报告和审定财务报表 hostname 已经是). 在工作之外,它是一个不开放的,所以一个日志实例既服务于工人,也服务于它的处理器.
任何其他日志可以调用 relayJobLog({ level, message }) 从它的运输:它将线导到任务上,绑定在当前的同步上下文中,而别无他法.
任务标记
使用 getJobTags 在更好的公牛板界面中组织和过滤您的工作的函数:
getJobTags(job) {
return [
job.queueName,
job.data.priority || "normal",
process.env.NODE_ENV || "development"
];
}
API 参考
patch(processor, redis)
用更好的公牛板功能包扎工作处理器.
processor: 您的同步工作处理功能redis: Redis 连接实例
createPinoJobLogDestination(options?)
一个将每个日志行转发到当前工作日志的皮诺目的地.
options.ignore: 从中继消息中丢出的额外密钥
relayJobLog({ level, message })
将一个日志行转发到绑定在当前同步上下文(如果有的话)的工作. level 这是其中之一 log, (中文(简体) ), debug, (中文(简体) ), info, (中文(简体) ), warn, (中文(简体) ), error。 。 。 .
Worker
延伸公牛MQ Worker 与更好的公牛板整合.
所需的选项 :
ioredis: Redis 连接实例useWorkerThreads: 必须是truegetJobTags: 可选函数返回工作标签
解决问题
- 无法工作的工作取消: 确保您使用沙盒处理器( 分离文件)
- 未出现的工作: 检查您的 Redis 连接是否配置得当
- 建立问题: 确保参考
.cjs用于已建处理器的文件