آموزش BullMQ و Redis در Node.js؛ اجرای امن Worker

BullMQ کارهای طولانی Node.js را در صف می‌گذارد تا یک Worker مستقل آن‌ها را پردازش کند. وب‌سرور لازم نیست تا پایان تولید گزارش یا پردازش فایل منتظر بماند. Redis وضعیت صف را نگه می‌دارد و برنامه می‌تواند پیشرفت، شکست و تلاش مجدد را پیگیری کند. این جداسازی، کار را ذاتاً سریع‌تر نمی‌کند؛ انتظار کاربر و رقابت منابع را قابل مدیریت می‌کند.

در این آموزش یک صف پردازش متن کوچک می‌سازیم، آن را از درخواست وب جدا می‌کنیم و سپس درباره استقرار، توقف امن و جلوگیری از اجرای تکراری تصمیم می‌گیریم. کد نمونه عمداً عملیات مالی یا ارسال واقعی ایمیل انجام نمی‌دهد. پس از فهم چرخه کار می‌توانید آن را برای پروژه‌ای روی هاست Node.js پاستا توسعه دهید.

چه زمانی صف لازم است و چه زمانی نه؟

تولید فایل خروجی، ساخت Embedding برای چند سند، تبدیل تصویر و همگام‌سازی با سرویس بیرونی از کارهای مناسب صف هستند. درخواست وب فقط پذیرش کار را ثبت می‌کند و یک شناسه پیگیری برمی‌گرداند. کاربر وضعیت را از مسیر احراز هویت‌شده می‌خواند؛ قطع مرورگر نباید به معنی گم‌شدن کار پذیرفته‌شده باشد.

برای یک محاسبه چند میلی‌ثانیه‌ای، افزودن Redis و Worker ممکن است پیچیدگی بی‌فایده بسازد. همچنین کاری که باید قبل از پاسخ نهایی قطعی شود، مثل اعتبارسنجی مجوز خرید، نباید صرفاً برای سریع‌شدن صفحه به آینده موکول شود. صف برای کار قابل تعویق است، نه حذف تضمین‌های کسب‌وکار.

معماری نمونه و پیش‌نیازها

سه جزء داریم: تولیدکننده کار، Redis و مصرف‌کننده. در محیط واقعی، داده اصلی و وضعیت کسب‌وکار در دیتابیس برنامه باقی می‌ماند. پیام صف بهتر است شناسه رکورد و نسخه عملیات را حمل کند، نه کل پرونده مشتری یا کلید API. Worker با همان محدودیت دسترسی موردنیاز، داده را از منبع اصلی می‌خواند.

Node.js پشتیبانی‌شده، Redis در دسترس خصوصی و یک پروژه آزمایشی لازم است. دستورهای زیر در پوشه خالی اجرا می‌شوند. پس از نصب، فایل قفل وابستگی‌ها را نگه دارید و قبل از ارتقای نسخه اصلی BullMQ، راهنمای مهاجرت آن را بخوانید. نمونه از API مشترک Queue و Worker استفاده می‌کند؛ با وجود backendهای دیگر، این راهنما مشخصاً برای Redis است.

npm init -y
npm install bullmq ioredis
# Set REDIS_URL in your local environment; do not commit it.
# Example shape: redis://HOST:6379

۱. تولیدکننده کار را بسازید

فایل producer.mjs یک متن غیرحساس نمونه را وارد صف می‌کند. اتصال تولیدکننده نباید هنگام قطع Redis تا ابد معطل بماند؛ وب‌سرور باید بتواند شکست پذیرش کار را اعلام کند. طبق مستندات اتصال BullMQ، سیاست Retry اتصال تولیدکننده با Worker متفاوت است و برای Redis صف باید سیاست noeviction بررسی شود.

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

if (!process.env.REDIS_URL) throw new Error('REDIS_URL is required');
const connection = new IORedis(process.env.REDIS_URL, {
  maxRetriesPerRequest: 1,
  enableOfflineQueue: false,
});
connection.on('error', () => console.error('Redis connection failed'));
const queue = new Queue('text-demo', { connection });
try {
  await queue.waitUntilReady();
  const job = await queue.add('count-words', { text: 'سلام از صف کار' }, {
    jobId: 'demo-text-v1',
    attempts: 3,
    backoff: { type: 'exponential', delay: 2000 },
    removeOnComplete: { age: 3600, count: 100 },
    removeOnFail: { age: 86400, count: 100 },
  });
  console.log({ acceptedJob: job.id });
} finally {
  await queue.close();
  await connection.quit();
}

شناسه ثابت برای تکرار همین آزمایش مفید است، اما تضمین دائمی اجرای یک‌باره نیست. وقتی رکورد کار از صف پاک شود، همان شناسه می‌تواند دوباره پذیرفته شود. در محصول واقعی، شناسه را از شناسه عملیات تأییدشده سمت سرور بسازید و برای هر مشتری مرز مالکیت مستقل داشته باشید. از ورودی آزاد کاربر برای انتخاب نام صف یا دسترسی به کار دیگران استفاده نکنید.

۲. Worker را جدا اجرا کنید

فایل worker.mjs فقط متن نمونه را می‌شمارد و نتیجه کوچک برمی‌گرداند. در شروع، هم‌زمانی پایین انتخاب شده تا رفتار قابل مشاهده باشد. برای کار CPU‌محور، بالا بردن concurrency در یک پردازش Node.js الزاماً سرعت را بیشتر نمی‌کند؛ ممکن است Event Loop و تمدید قفل کارها را مختل کند.

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

if (!process.env.REDIS_URL) throw new Error('REDIS_URL is required');
const connection = new IORedis(process.env.REDIS_URL, {
  maxRetriesPerRequest: null,
});
connection.on('error', () => console.error('Redis connection failed'));
const worker = new Worker('text-demo', async job => {
  if (job.name !== 'count-words') throw new Error('Unsupported job');
  const text = job.data.text;
  if (typeof text !== 'string' || text.length > 10000) {
    throw new Error('Invalid demo input');
  }
  return { words: text.trim() ? text.trim().split(/\s+/u).length : 0 };
}, { connection, concurrency: 2 });
worker.on('error', () => console.error('Worker connection failed'));
worker.on('completed', (job, result) => console.log({ id: job.id, ...result }));
worker.on('failed', job => console.error({ failedJob: job?.id }));
let closing = false;
async function stop() {
  if (closing) return;
  closing = true;
  await worker.close();
  await connection.quit();
}
for (const signal of ['SIGTERM', 'SIGINT']) {
  process.on(signal, () => stop().catch(() => { process.exitCode = 1; }));
}

در دو ترمینال، ابتدا node worker.mjs و سپس node producer.mjs را اجرا کنید. انتظار دارید تولیدکننده شناسه کار پذیرفته‌شده را چاپ کند و Worker نتیجه شمارش را ثبت کند. این نمونه، آزمون اتصال و چرخه صف است؛ جای آزمون فشار، دوام داده یا مجوزهای محصول را نمی‌گیرد.