BullMQ ilg'or imkoniyatlari: flow, prioritet, sandbox va Bull Board

Qisqacha: BullMQ oddiy navbatdan ko'proq narsani biladi. Flow — ota ish bolalari tugaguncha kutadi va ularning natijalarini oladi (hisobotni bo'laklarga bo'lib yig'ish). Prioritet — muhim ish oldinroq; lekin prioritetsiz ishlar prioritetlilardan OLDIN olinadi — kanon buni ko'rdi, BullMQ kodidagi hujjat ham shuni aytadi. Sandboxed processor — CPU og'ir ishni alohida thread yoki jarayonda bajaradi, worker'ning event loop'i bo'sh qoladi. Bull Board — navbatlarni ko'rish va boshqarish paneli, faqat auth ortida.

Bu darsda

• FlowProducer bilan ota va bola ishlar daraxtini yasaysiz.
• Prioritet qanday ishlashini va uning kutilmagan qoidasini bilasiz.
• updateProgress bilan foydalanuvchiga jarayonni ko'rsatasiz.
• Sandboxed processor'ni Windows'dagi yo'l tuzog'i bilan birga ishga tushirasiz.
• Bull Board panelini ulab, nega u ochiq qolmasligi kerakligini tushunasiz.

Oldin bilishingiz kerak: BullMQ ishonchlilik, BullMQ asoslari, worker_threads, Xavfsiz fayl operatsiyalari.

Oldingi darsda ishlarni ishonchli qildik: qayta urinish, limiter, stalled. Endi murakkabroq ish shakllari.

1. Nega bu kerak?

Jasur aka oy oxirida jamoa hisobotini so'raydi: har a'zo uchun bajarilgan vazifalar, vaqt, grafik — keyin bitta PDF. Bitta ish qilsak, u 2 daqiqa ishlaydi, bir joyda yiqilsa — boshidan. Yaxshiroq yo'l: har a'zo uchun alohida bo'lak-ish (parallel), ular tugagach — bitta yig'uvchi ish. Bu flow.

Bir vaqtda navbatda 3 000 ta eslatma va bitta "parolni tiklash" xati turibdi. Parol xati 3 000 ta eslatmani kutmasligi kerak. Bu prioritet (yoki alohida navbat — pastda ko'ramiz).

PDF yasash CPU'ni 5 soniya band qiladi — worker'ning event loop'i to'xtaydi, qulf yangilanmaydi va ish yolg'on "stalled" bo'ladi (oldingi dars). Bu sandboxed processor muammosi.

2. Flow: ota va bola ishlar

2.1 Kanon tajribasi

Kanon kodi (telefon uchun formatlab):

import { FlowProducer, Worker } from "bullmq";

const fp = new FlowProducer({ connection, prefix: P });

const w = new Worker("hisobot", async (j) => {
  if (j.name === "yig") {
    const b = await j.getChildrenValues();
    const jami = Object.values(b).reduce((a, x) => a + x, 0);
    const soni = Object.keys(b).length;
    console.log(`ota: bolalar ${soni} ta, jami ${jami}`);
    return "tayyor";
  }
  await kut(20);
  return j.data.n;
}, { connection, prefix: P, concurrency: 5 });

const daraxt = await fp.add({
  name: "yig", queueName: "hisobot",
  children: [1, 2, 3, 4].map((n) => ({
    name: "bo'lak", queueName: "hisobot", data: { n },
  })),
});

Kanon chiqishi:

flow: ota: bolalar 4 ta, jami 10

flowchart TB
  O["yig - ota ish<br/>waiting-children"]
  B1["bo'lak n=1"] -->|1| O
  B2["bo'lak n=2"] -->|2| O
  B3["bo'lak n=3"] -->|3| O
  B4["bo'lak n=4"] -->|4| O
  O -->|getChildrenValues - jami 10| N["tayyor"]

Diagrammaga qarang: strelkalar bola natijasi otaga qarab oqadi. Ota bolalardan oldin yaratiladi, lekin eng oxirida bajariladi.

Qanday ishladi:

1. fp.add(daraxt) hamma ishlarni bitta atomik amal bilan qo'shdi: 4 ta bola waiting da, ota esa waiting-children holatida.
2. Worker bolalarni oldi (concurrency 5 — to'rttasi parallel), har biri o'z n ini qaytardi.
3. Oxirgi bola tugagach, BullMQ otani waiting ga o'tkazdi.
4. Ota getChildrenValues() bilan bolalar natijasini oldi: 1 + 2 + 3 + 4 = 10.

Bolalar boshqa navbatlarda ham bo'lishi mumkin (queueName) — masalan, rasm bo'laklari rasm navbatida, yig'uvchi hisobot da.

2.2 Model

Mexanizmni brauzerda ko'raylik — bu BullMQ emas, rekursiv model:

// Flow modeli: ota ish bolalari tugaguncha "waiting-children" da.
async function flow(daraxt, ishlovchi) {
  const bolalar = {};
  for (const [i, bola] of (daraxt.children ?? []).entries()) {
    bolalar[`${bola.name}:${i + 1}`] = await flow(bola, ishlovchi);
  }
  return ishlovchi(daraxt, bolalar);
}

const daraxt = {
  name: "yig",
  children: [1, 2, 3, 4]
    .map((n) => ({ name: "bo'lak", data: { n } })),
};

const natija = await flow(daraxt, async (ish, bolalar) => {
  if (ish.name !== "yig") return ish.data.n;
  const qiymatlar = Object.values(bolalar);
  console.log("bolalar:", JSON.stringify(bolalar));
  return qiymatlar.reduce((a, x) => a + x, 0);
});
console.log("jami", natija);

Konsolda:

bolalar: {"bo'lak:1":1,"bo'lak:2":2,"bo'lak:3":3,"bo'lak:4":4}
jami 10

Haqiqiy getChildrenValues() da kalitlar — bolalarning Redis kalitlari (<prefiks>:<navbat>:<id>), shuning uchun kanon Object.values dan foydalandi, kalit nomiga tayanmadi. Modelimiz bolalarni ketma-ket bajaradi, BullMQ esa parallel — natija bir xil, vaqt boshqa.

2.3 Bola yiqilsa

Sukutda bola failed bo'lsa, ota waiting-children da qolib ketadi — hisobot hech qachon tugamaydi. Bola ishiga qo'shimcha sozlama beriladi (hujjat bo'yicha, kanon sinamagan): failParentOnFailure: true — ota ham yiqiladi; ignoreDependencyOnFailure: true — ota yiqilgan bolasiz davom etadi. Qaysi biri kerakligini biznes hal qiladi: hisobotda bitta a'zo yetishmasa — chiqaramizmi yoki yo'q?

2.4 Qachon flow kerak emas

Bolalar bir-biriga bog'liq bo'lmasa va natijasi kerak bo'lmasa — oddiy addBulk yetarli. Flow — "hammasi tugagach, natijalar bilan nimadir qil" kerak bo'lganda.

3. Prioritet

3.1 Kutilmagan qoida

Kanon uch ishni shu tartibda qo'shdi va concurrency 1 li worker bilan bajardi:

await q.add("past", { n: "past" }, { priority: 10 });
await q.add("yuqori", { n: "yuqori" }, { priority: 1 });
await q.add("oddiy", { n: "prioritetsiz" });

prioritet tartibi: prioritetsiz → yuqori → past (prioritetsiz ish prioritetlilardan OLDIN)

priority: 1 — eng yuqori (kichik son — oldin). Bu kutilgan. Kutilmagani — prioritetsiz ish birinchi bo'ldi. Bu xato emas, hujjatlashtirilgan xatti-harakat. bullmq 6.3.19 manbasidagi izoh (job.js): "0 means no explicit priority, and jobs with no explicit priority are processed before prioritized jobs". Ichkarida: prioritetsiz ishlar oddiy ro'yxatda (wait), prioritetlilar — alohida sorted set'da (prioritized). Worker avval ro'yxatdan oladi, u bo'sh bo'lsagina sorted set'ga qaraydi (moveToActive Lua skripti).

// BullMQ qaysi ishni oladi (soddalashtirilgan): avval "wait" ro'yxati
// (prioritetsiz ishlar, FIFO), u bo'sh bo'lsa — "prioritized"
// (kichik son — oldin; tenglarda — qo'shilish tartibi).
function tartib(ishlar) {
  const wait = ishlar.filter((i) => !i.priority);
  const pri = ishlar
    .map((i, n) => ({ ...i, n }))
    .filter((i) => i.priority)
    .sort((a, b) => a.priority - b.priority || a.n - b.n);
  return [...wait, ...pri].map((i) => i.nom);
}

console.log(tartib([
  { nom: "past", priority: 10 },
  { nom: "yuqori", priority: 1 },
  { nom: "prioritetsiz" },
]).join(" → "));

Konsolda:

prioritetsiz → yuqori → past

Amaliy xulosa: bitta navbatda prioritet ishlatsangiz — hamma ishga prioritet bering. Aks holda "oddiy" ishlar "muhim"larni kutdiradi. Prioritet diapazoni 1 dan 2 097 151 gacha; hujjatga ko'ra, prioritet ishlashga biroz narx qo'shadi (sorted set — O(log n)).

3.2 Muqobil: alohida navbat

Ko'pincha prioritet o'rniga alohida navbat yaxshiroq: muhim va oddiy navbatlari, har biriga o'z worker'i. Afzalligi — oddiy ishlar qanchalik ko'p bo'lmasin, muhim navbatning worker'i band emas. Kanon modulida aynan shunday: eslatmalar, yetkazishlar, tizim (BullMQ asoslari).

Jadvaldagi ikki so'z: ochlik (starvation) — past prioritetli ish navbatga doim yangi muhim ishlar kelgani uchun hech qachon olinmay qolishi; SLA — "bu ish ko'pi bilan qancha vaqtda bajarilishi kerak" degan va'da.

4. Progress

Uzoq ish (eksport, hisobot) foydalanuvchiga "45%" ko'rsatsa — u sahifani yopmaydi. Worker:

for (const [i, qator] of qatorlar.entries()) {
  await yoz(qator);
  if (i % 1000 === 0) {
    await j.updateProgress(Math.round((i / qatorlar.length) * 100));
  }
}

updateProgress qiymatni ishga yozadi va progress hodisasini tarqatadi (asoslar darsida kanon chiqishida progress:1(50) edi). Qiymat son yoki obyekt bo'lishi mumkin ({ qator: 4000, jami: 10000 }). API uni QueueEvents dan olib, SSE bilan brauzerga uzatadi.

Har qatorda chaqirmang — har chaqiruv Redis'ga yozuv va hodisa. 1–5 soniyada bir marta yetarli.

5. Sandboxed processor

5.1 Muammo va yechim

Processor CPU'ni uzoq band qilsa, worker'ning event loop'i to'xtaydi: qulf yangilanmaydi, boshqa ishlar olinmaydi. Sandboxed processor — processor funksiyasini alohida fayl sifatida berish. BullMQ uni alohida jarayonda (child_process, sukut) yoki thread'da (useWorkerThreads: true) ishga tushiradi (worker_threads).

Kanon 30 million sonni qo'shadigan processor faylini yozdi:

// sandbox-ishlov.mjs — alohida fayl
export default async (job) => {
  let s = 0;
  for (let i = 0; i < 3e7; i++) s += i;
  return process.pid + ":" + s;
};

Va worker'ga funksiya o'rniga fayl yo'lini berdi:

import { pathToFileURL } from "node:url";

const ws = new Worker("sandbox", pathToFileURL(fayl), {
  connection, prefix: P, useWorkerThreads: true,
});

Chiqishda pid bir xil qismiga qarang — u processor qayerda ishlaganini aytadi:

sandboxed (worker_threads): natija 449999985000000, jarayon pid bir xil: true (thread — event loop bloklanmaydi)

Nima bo'ldi? pid bir xil: true — useWorkerThreads bilan processor o'sha jarayonning boshqa threadida ishladi. Asosiy thread'ning event loop'i bo'sh: qulf yangilanadi, hodisalar keladi.

5.2 Windows tuzog'i

Kanon birinchi urinishda oddiy mutlaq yo'l satrini berdi (C:\Users\…\sandbox-ishlov.mjs). Natija — ERR_UNSUPPORTED_ESM_URL_SCHEME. Sabab: processor .mjs (ESM) va uni yuklashda yo'l import() ga boradi. ESM yuklagichi esa mutlaq yo'lni URL deb o'qiydi: C: — "C" sxemali URL, Node uni tanimaydi. pathToFileURL(fayl) yo'lni file:///C:/… ga aylantiradi — shundan keyin ishladi.

Bu tuzoq Linux'da ko'rinmaydi (/home/… — sxemasiz), shuning uchun Linux serverda ishlagan kod Windows'dagi dasturchi kompyuterida yiqiladi. Yo'lni import() ga beradigan har joyda pathToFileURL ishlating (Xavfsiz fayl operatsiyalari).

5.3 Qachon sandbox

6. Metrikalar

Worker'ga metrics: { maxDataPoints: MetricsTime.ONE_WEEK } berilsa, BullMQ daqiqa sayin nechta ish tugaganini yig'adi. Kanon o'qidi:

metrikalar: {"count":0,"meta":3}

count: 0 — 3 ta ish tugagan bo'lsa ham. Sabab: metrikalar daqiqalik nuqtalarga yig'iladi, sinov esa bir necha soniya davom etdi — birinchi nuqta hali yozilmagan. Qisqa sinovda metrika bo'sh chiqishi normal. Production'da ular grafik uchun: "soatiga nechta email" — o'sib-kamayishi ko'rinadi. Batafsil kuzatuv (Prometheus, OpenTelemetry) — keyingi qismlar mavzusi.

7. Bull Board

7.1 Ulash

Bull Board — navbatlarni ko'rish paneli: har navbatdagi ishlar, holatlar, xato sabablari, ma'lumot; tugmalar — qayta urinish, o'chirish, navbatni to'xtatish. Kanon versiyasi 9.10.4, Express adapteri bilan:

import { createBullBoard } from "@bull-board/api";
import { BullMQAdapter } from "@bull-board/api/bullMQAdapter";
import { ExpressAdapter } from "@bull-board/express";

const sa = new ExpressAdapter();
sa.setBasePath("/admin/navbat");
createBullBoard({
  queues: [new BullMQAdapter(q)],
  serverAdapter: sa,
});
app.use("/admin/navbat", sa.getRouter());

Kanon panelni ochib, sahifa va API javobini tekshirdi:

UI: HTTP 200, 2109 bayt; API: navbatlar hisobot(completed 5)
Eslatma: Bull Board ishni qayta urinish/o'chirish imkonini beradi — auth'siz ochiq qoldirish = navbatni boshqarishni berish.

UI sahifasi ochildi, API hisobot navbatida 5 ta tugagan ishni ko'rsatdi — flow tajribasining 4 bolasi va 1 otasi.

7.2 Nega auth ortida

Kanon panelni auth'siz ochdi — bu faqat tajriba. Panel orqali:

• ishlarning data si ko'rinadi — email manzillar, vazifa matnlari;
• istalgan ishni qayta yurgizish mumkin — "to'lovni qaytarish" ishini ikkinchi marta;
• navbatni tozalash yoki to'xtatish mumkin.

Ya'ni panel — admin huquqi. Uni marshrutga faqat admin rolini tekshiradigan middleware ortida ulang (rollar — 28-qism), qo'shimcha ravishda faqat ichki tarmoqdan. Hech qachon app.use("/admin/navbat", …) ni himoyasiz qoldirmang.

Tekshirib ko'ring: Bull Board'ni faqat NODE_ENV !== "production" da ulash yetarli himoyami?

8. Loyiha qadami: kanon tajribasi, modulga kirmadi

Bu darsda navbat.ts ga kod qo'shilmaydi. Kanon flow, prioritet, sandbox va Bull Board'ni tajribada sinadi, lekin modulga olmadi:

Commit hali yo'q — modul Cron darsida tugaydi. O'zingiz sinab ko'ring: tajriba papkasida (npm i bullmq@6.3.19 ioredis@6.0.0, Redis ≥ 5) uchta ishni prioritet bilan va bittasini prioritetsiz qo'shib, tartibni yozib oling. Keyin hammasiga prioritet bering va solishtiring.

9. Ko'p uchraydigan xatolar

10. Mashqlar

Mashqlar kurs/mashqlar/29/47-ilgor/ da.

1-mashq (oson): bo'shliqlarni to'ldiring

• Ota va bola ishlarni bitta amal bilan qo'shadigan klass —
• Bolalarni kutayotgan ota ishning holati —
• Bolalar natijalarini oladigan metod —
• BullMQ'da eng yuqori prioritet — eng son
• Sandboxed processor'ni thread'da yurgizish sozlamasi —

2-mashq (o'rta): navbat tartibini oldindan ayting

Navbatga shu tartibda olti ish tushdi. Darsdagi tartib funksiyasi bilan worker ularni qaysi tartibda olishini chiqaring. Keyin prioritetsiz ishlarga priority: 3 berib, tartibni yana chiqaring.

const ishlar = [
  { nom: "hisobot", priority: 5 },
  { nom: "eslatma-a" },
  { nom: "to'lov", priority: 1 },
  { nom: "eksport", priority: 5 },
  { nom: "eslatma-b" },
  { nom: "parol", priority: 1 },
];

const ishlar = [
  { nom: "hisobot", priority: 5 },
  { nom: "eslatma-a" },
  { nom: "to'lov", priority: 1 },
  { nom: "eksport", priority: 5 },
  { nom: "eslatma-b" },
  { nom: "parol", priority: 1 },
];

function tartib(royxat) {
  const wait = royxat.filter((i) => !i.priority);
  const pri = royxat
    .map((i, n) => ({ ...i, n }))
    .filter((i) => i.priority)
    .sort((a, b) => a.priority - b.priority || a.n - b.n);
  return [...wait, ...pri].map((i) => i.nom);
}

console.log(tartib(ishlar).join(" → "));
const hammasiga = ishlar
  .map((i) => ({ ...i, priority: i.priority ?? 3 }));
console.log(tartib(hammasiga).join(" → "));

eslatma-a → eslatma-b → to'lov → parol → hisobot → eksport
to'lov → parol → eslatma-a → eslatma-b → hisobot → eksport

3-mashq (qiyin): ikki qavatli flow

Hisobot ikki bo'limdan, har bo'lim ikki bo'lakdan iborat. Bo'lak n * 100 qaytaradi, bo'lim va hisobot — bolalari yig'indisini va <nom>: <yig'indi> qatorini chiqaradi. Rekursiv flow(ish, ishlovchi) yozing (bolalar natijalari massiv bo'lsin).

async function flow(ish, ishlovchi) {
  const bolalar = [];
  for (const bola of ish.children ?? []) {
    bolalar.push(await flow(bola, ishlovchi));
  }
  return ishlovchi(ish, bolalar);
}

const bolak = (n) => ({ name: "bo'lak", data: { n } });
const daraxt = {
  name: "hisobot",
  children: [
    {
      name: "bo'lim", data: { nom: "A" },
      children: [bolak(1), bolak(2)],
    },
    {
      name: "bo'lim", data: { nom: "B" },
      children: [bolak(3), bolak(4)],
    },
  ],
};

const jami = await flow(daraxt, async (ish, bolalar) => {
  if (ish.name === "bo'lak") return ish.data.n * 100;
  const s = bolalar.reduce((a, x) => a + x, 0);
  const nom = ish.data ? `${ish.name} ${ish.data.nom}` : ish.name;
  console.log(`${nom}: ${s}`);
  return s;
});
console.log("natija:", jami);

bo'lim A: 300
bo'lim B: 700
hisobot: 1000
natija: 1000

11. Real ishda

• Flow — video qayta ishlash (bo'laklar → birlashtirish), hisobot va eksport, ko'p bosqichli import.
• Alohida navbatlar prioritetdan ko'ra ko'p uchraydi: "transactional email" va "marketing email" — har doim alohida.
• Sandbox — PDF, Excel, arxivlash, ishonchsiz kutubxonalar.
• Bull Board (yoki Taskforce.sh kabi tijoriy panel) — har BullMQ loyihada; kirish — faqat admin.
• Intervyuda: "Bir necha ish tugagach bitta ish qanday ishga tushiriladi?", "CPU og'ir ish worker'ni qanday bloklaydi?", "Admin panelni qanday himoyalaysiz?"

Xulosa

• Flow: FlowProducer.add — ota waiting-children da, getChildrenValues() bilan natijalar (kanon: 4 bola, jami 10).
• Prioritet: kichik son — oldin, lekin prioritetsiz ishlar prioritetlilardan oldin (kanon tartibi va BullMQ hujjati); hammaga prioritet bering yoki alohida navbat qiling.
• updateProgress — uzoq ishlar uchun, siyrak chaqiring.
• Sandboxed processor — CPU ishini thread/jarayonga chiqaradi; Windows'da pathToFileURL.
• Metrikalar daqiqalik — qisqa sinovda count: 0.
• Bull Board — kuchli boshqaruv paneli; faqat admin auth ortida.

Keyingi dars: Cron va rejali vazifalar — "har kuni 03:30" ifodasi, vaqt zonasi tuzog'i va ikki server nusxasida vazifa ikki marta ishlamasligi; navbat.ts modulini yakunlaymiz.

Manbalar

• BullMQ — Flows: https://docs.bullmq.io/guide/flows; Fail parent / ignore dependency: https://docs.bullmq.io/guide/flows/fail-parent
• BullMQ — Prioritized jobs: https://docs.bullmq.io/guide/jobs/prioritized
• BullMQ — Sandboxed processors: https://docs.bullmq.io/guide/workers/sandboxed-processors
• BullMQ — Metrics: https://docs.bullmq.io/guide/metrics
• Bull Board: https://github.com/felixmosh/bull-board
• bullmq 6.3.19 manba kodi (job.js prioritet izohi, moveToActive Lua skripti) — paket ichidan, 2026-10-28 da o'qildi
• Node.js — ESM, import() va fayl URL'lari: https://nodejs.org/docs/latest-v24.x/api/esm.html#urls
• Kanon tajribasi (IlmHamroh, 2026-10-27): bullmq 6.3.19, @bull-board 9.10.4, Redis 8.10.2 (Docker), Node 24.21.0, Windows 11. Modellar va mashq yechimlari — biz ishga tushirdik (2026-10-28).