BullMQ asoslari: Queue, Worker va ish hayot sikli

Qisqacha: BullMQ — Node uchun Redis ustidagi navbat kutubxonasi. Uch asosiy klass bor: Queue (ish qo'shadi — producer), Worker (ishni bajaradi — consumer) va QueueEvents (ishlar hodisalarini tinglaydi). Har ish holatlardan o'tadi: waiting → active → completed yoki failed; kechiktirilgani avval delayed. Worker'ni API bilan bitta jarayonda emas, alohida jarayonda yurgizing. Ulanishda maxRetriesPerRequest: null shart, Redis 5.0 dan past bo'lsa BullMQ ishlamaydi.

Bu darsda

• BullMQ'ni o'rnatib, Queue, Worker va QueueEvents bilan birinchi navbatni yasaysiz.
• Ish holatlarini va ular orasidagi o'tishlarni bilasiz.
• Worker'ni alohida jarayonda yurgizish nega muhimligini tushunasiz.
• ioredis va node-redis klientlari masalasini bilasiz.
• Loyiha moduli navbat.ts ning birinchi qismini — vazifa muddati eslatmasini — o'qiysiz.

Oldin bilishingiz kerak: Nega navbat kerak, Redis persistence va operatsiya, Pub/Sub va Redis Streams, child_process.

Oldingi darsda navbat nega kerakligini o'lchov bilan ko'rdik. Endi uni haqiqiy kutubxona bilan quramiz.

Maslahat: Bu darsdagi BullMQ kodi Redis talab qiladi va saytdagi «Ishga tushir» tugmasida ishlamaydi. Bunday bloklarni o'z kompyuteringizda yurgizasiz. Brauzerda ishlaydigan o'quv modellari esa jonli.

1. Nega bu kerak?

SKIP LOCKED darsida navbatni Postgres jadvalida o'zimiz yozgan edik: ish olish, qayta urinish, kechiktirish, "osilib qolgan" ishlarni qaytaruvchi qorovul. Har loyihada buni qaytadan yozish — vaqt va xato manbai. Kutubxona bularni tayyor beradi.

Node dunyosida eng ko'p ishlatiladigan navbat kutubxonasi — BullMQ. U eski Bull kutubxonasining TypeScript'da qayta yozilgan vorisi, ma'lumotni Redis'da saqlaydi va har muhim amalni Lua skript bilan atomik bajaradi (Atomiklik: MULTI, Lua va pipelining). 30-qismda o'rganadigan NestJS freymvorki ham navbat uchun aynan BullMQ'ni o'raydi (Navbatlar: @nestjs/bullmq).

Bizning loyihada birinchi ish — vazifa muddati eslatmasi: Malika vazifaga "ertaga soat 12:00" muddat qo'ydi, 30 daqiqa oldin unga eslatma ketishi kerak. Bu kechiktirilgan ish — BullMQ'ning delay imkoniyati.

2. O'rnatish va Redis talablari

Kanon versiyalari: bullmq 6.3.19, ioredis 6.0.0, Redis 8.10.2 (Node 24.21.0):

npm i bullmq@6.3.19 ioredis@6.0.0

Redis'dan ikki talab bor:

1. Versiya ≥ 5.0. Windows'da ko'p uchraydigan eski "Redis on Windows" porti 3.0 versiyasida qolgan. Biz uni ishga tushirib, BullMQ'ga ulandik (bullmq 6.3.19):

Error: Redis version needs to be greater or equal than 5.0.0 Current: 3.0.504

BullMQ Redis'ning versiyasini ulanishda tekshiradi va eski serverda ishlamaydi. Kanon Redis 8 ni Docker'da ishlatdi (Redis nima va ulanish).

2. maxmemory-policy noeviction. Xotira to'lganda Redis kalitlarni o'chirib yubormasin — kesh uchun bu normal, navbat uchun esa ish yo'qolishi degani. Redis persistence darsida ko'rgandik: navbat — noeviction, kesh — allkeys-lru. Kanon Redis --appendonly yes --maxmemory-policy noeviction bilan ishga tushirilgan. Boshqa siyosatda BullMQ ogohlantirish yozadi.

2.1 ioredis yoki node-redis

27-qismda Redis bilan redis paketi (node-redis) orqali ishladik. BullMQ esa tarixan ioredis ishlatadi va hujjatining asosiy yo'li ham shu. Kanon ham ioredis ni tanladi.

BullMQ 6 da o'zgarish bor: ioredis endi ixtiyoriy peer bog'liqlik — BullMQ uni o'zi bilan olib kelmaydi, loyihangiz alohida o'rnatishi kerak. Uni o'rnatmasangiz, BullMQ "could not load the optional 'ioredis' package" xatosini beradi. Muqobili — node-redis adapteri createNodeRedisClient. Kanon uni sinadi:

import { Queue, Worker, QueueEvents, RedisConnection,
  createNodeRedisClient } from "bullmq";
import { createClient } from "redis";

RedisConnection.clientFactory = (o) => createNodeRedisClient(
  createClient({ url: `redis://${o.host}:${o.port}` }));

node-redis orqali natija: 42

Ish (x: 21, natija x * 2) node-redis orqali bajarildi. Bitta loyihada ikkita Redis klienti bo'lishi ortiqcha — kesh node-redis'da, navbat ioredis'da. Lekin bu kursda BullMQ hujjati va kanon kodi ioredis bilan, shuning uchun darslarda ioredis. Bitta klientga o'tish — loyiha qarori, uni vazifalar-api ga ulash bosqichida qilamiz. README'ga ko'ra node-redis adapteri uchun redis ≥ 5.0.0 kerak.

3. Uch klass: Queue, Worker, QueueEvents

3.1 Ish qo'shish

import { Queue } from "bullmq";

const connection = {
  host: "127.0.0.1", port: 6379, maxRetriesPerRequest: null,
};
const q = new Queue("asos", { connection, prefix: "sinov" });

const a = await q.add("j", { x: 1 });
const b = await q.add("j", { x: "yiqil" }, { attempts: 1 });
const c = await q.add("j", { x: 2 }, { delay: 300 });

q.add(nom, data, opts) uch narsa oladi:

prefix — Redis kalitlari oldidagi qism (sukut bull): sinov:asos:1, sinov:asos:wait va hokazo. Bir Redis'da bir nechta ilova yoki test bo'lsa, prefiks ularni ajratadi.

maxRetriesPerRequest: null — worker Redis'ning bloklovchi buyruqlarida (ish kelguncha kutish) uzoq turadi. ioredis sukutda 20 urinishdan keyin so'rovni xato bilan tugatadi — worker uchun bu noto'g'ri. BullMQ worker ulanishida buni talab qiladi.

3.2 Ishni bajarish

import { Worker } from "bullmq";

const w = new Worker("asos", async (j) => {
  await j.updateProgress(50);
  if (j.data.x === "yiqil") throw new Error("ataylab");
  return "ok-" + process.pid;
}, { connection, prefix: "sinov" });

Worker'ga bitta funksiya — processor beriladi. Uning shartnomasi oddiy:

• Promise qaytsa — ish completed, qaytgan qiymat ish natijasi (returnvalue) bo'lib saqlanadi.
• Xato otsa — ish failed (yoki qayta urinishga — keyingi darsda), failedReason — xato xabari.
• j.updateProgress(50) — ish qanchalik bajarilganini bildiradi; hodisa sifatida tarqaladi.

Worker yaratilishi bilan ishlay boshlaydi (autorun). U Redis'dan ish kelishini kutadi, bitta oladi, processor'ni chaqiradi, natijani yozadi va keyingisini oladi.

3.3 Hodisalarni tinglash

import { QueueEvents } from "bullmq";

const ev = new QueueEvents("asos", { connection, prefix: "sinov" });
await ev.waitUntilReady();
ev.on("completed", ({ jobId, returnvalue }) => { /* ... */ });
ev.on("failed", ({ jobId, failedReason }) => { /* ... */ });

const natija = await a.waitUntilFinished(ev);

QueueEvents ichida Redis Streams ishlaydi: BullMQ har o'zgarishni navbatning events oqimiga XADD qiladi, QueueEvents uni XREAD bilan o'qiydi. Shuning uchun u boshqa jarayonda, boshqa mashinada ham ishlaydi. waitUntilFinished(ev) — ish tugashini kutib natijani qaytaradi (testlarda qulay, so'rov ichida esa ishlatmang — navbatning ma'nosi yo'qoladi).

Uch klass bir-biri bilan to'g'ridan gaplashmaydi — hammasi Redis orqali:

sequenceDiagram
  participant Q as Queue (API)
  participant R as Redis
  participant W as Worker
  participant E as QueueEvents
  Q->>R: add (ish hash + wait ro'yxati)
  R->>W: ish beriladi
  W->>R: progress 50
  R-->>E: events oqimi (progress)
  W->>R: natija, completed
  R-->>E: events oqimi (completed)

4. Ish holatlari

stateDiagram-v2
  [*] --> waiting: add
  [*] --> delayed: add delay bilan
  [*] --> prioritized: add priority bilan
  delayed --> waiting: vaqti keldi
  waiting --> active: worker oldi
  prioritized --> active: worker oldi
  active --> completed: processor qaytdi
  active --> failed: xato, urinish qolmadi
  active --> delayed: xato, backoff bilan qayta
  active --> waiting: stalled
  completed --> [*]
  failed --> [*]

BullMQ 6 dagi holatlar (JobState turi): waiting, active, delayed, prioritized, waiting-children, completed, failed. Bu darsda birinchi beshtasini ko'ramiz, prioritized va waiting-children — ilg'or imkoniyatlar darsida, stalled — keyingi darsda.

Holat Redis'da ma'lumot tuzilmasi bilan ifodalanadi: waiting — list, delayed va prioritized — sorted set (vaqt yoki prioritet bo'yicha), completed/failed — sorted set. Ishning o'zi — hash. Siz Hash, List, Set va Sorted Set darsida o'rgangan tuzilmalar.

4.1 O'quv modeli

Hodisalar ketma-ketligini Redis'siz modelda ko'raylik. Bu BullMQ emas — uning holat mashinasining soddalashtirilgan nusxasi, soat esa virtual:

// Ish hayot siklining o'quv modeli (Redis'siz, virtual soat bilan).
class MiniNavbat {
  ishlar = [];
  hodisalar = [];
  #n = 0;
  add(nom, data, { delay = 0 } = {}) {
    const ish = { id: ++this.#n, nom, data, delay, holat: "" };
    this.ishlar.push(ish);
    this.#oz(ish, "added");
    this.#oz(ish, delay > 0 ? "delayed" : "waiting");
    return ish;
  }
  #oz(ish, holat, qoshimcha = "") {
    if (holat !== "added") ish.holat = holat;
    this.hodisalar.push(`${holat}:${ish.id}${qoshimcha}`);
  }
  async ishla(ishlovchi, soat) {
    for (const ish of this.ishlar) {
      if (ish.holat === "delayed" && ish.delay <= soat) {
        this.#oz(ish, "waiting");
      }
    }
    const kutayotgan = this.ishlar
      .filter((x) => x.holat === "waiting");
    for (const ish of kutayotgan) {
      this.#oz(ish, "active");
      try {
        ish.natija = await ishlovchi(ish);
        this.#oz(ish, "completed");
      } catch (e) {
        this.#oz(ish, "failed", `(${e.message})`);
      }
    }
  }
}

const q = new MiniNavbat();
q.add("rasm", { x: 1 });
q.add("rasm", { x: "yiqil" });
q.add("rasm", { x: 2 }, { delay: 300 });

const ishlovchi = async (ish) => {
  if (ish.data.x === "yiqil") throw new Error("ataylab");
  return "ok";
};
await q.ishla(ishlovchi, 0);   // soat 0 ms
await q.ishla(ishlovchi, 300); // soat 300 ms
console.log(q.hodisalar.join(" "));
console.log(q.ishlar.map((i) => `${i.id}:${i.holat}`).join(", "));

Konsolda:

added:1 waiting:1 added:2 waiting:2 added:3 delayed:3 active:1 completed:1 active:2 failed:2(ataylab) waiting:3 active:3 completed:3
1:completed, 2:failed, 3:completed

4.2 Kanon: haqiqiy BullMQ va alohida worker jarayoni

Kanon xuddi shu uchta ishni haqiqiy BullMQ bilan yurgizdi. Worker esa alohida jarayonda edi: asosiy skript fork() (child_process) bilan ishchi.mjs faylini ishga tushirdi, ichida «Ishni bajarish» bo'limidagi Worker. Natija (worker pid'i <worker pid> bilan almashtirilgan). hodisalar qatorini modelimiz chiqishi bilan yonma-yon qo'yib, qaysi so'zlar farq qilishini qidiring:

a natija: ok-<worker pid>
b xato: ataylab
c holati (delay 300): active
hodisalar: added:2 waiting:2 added:3 delayed:3 waiting:3 active:1 progress:1(50) completed:1 active:2 progress:2(50) failed:2 active:3 progress:3(50) completed:3
sanoq: {"completed":2,"failed":1,"delayed":0,"waiting":0,"active":0}

Modeldan uchta farqni ko'ring:

1. ok-<pid> — natija worker jarayonida yasalgan. API jarayoni uni Redis orqali oldi. Ikki jarayon bir-birini to'g'ridan bilmaydi.
2. progress:N(50) — updateProgress(50) hodisasi. Modelimizda yo'q edi.
3. 1-ish uchun added:1 waiting:1 yo'q. QueueEvents oqimni sukutda $ dan — "hozirdan boshlab" o'qiydi (lastEventId sozlamasi, manba kodida ko'rinadi). waitUntilReady() faqat ulanishni kutadi, birinchi XREAD esa biroz keyin boshlanadi — birinchi add shu oraliqqa tushib qolgan. Xulosa: hodisalar kuzatuv uchun, hisob-kitob uchun emas. Ish holatining haqiqiy manbai — navbatning o'zi (getState(), getJobCounts()).

Yana bir narsa: c 300 ms ga kechiktirilgan edi, lekin tekshirilganda u allaqachon active. Alohida jarayon ishga tushib ulanguncha 300 ms o'tib ketgan. Kechiktirish — "shundan oldin emas", "aynan shu paytda" emas.

4.3 Nega worker alohida jarayonda

Kichik loyihada bitta jarayonda boshlash mumkin — kod bir xil, faqat qayerda new Worker chaqirilgani farq qiladi. Lekin kodni boshidan ikkiga bo'lib yozing: server.ts (faqat Queue) va ishchi.ts (faqat Worker).

Tekshirib ko'ring: API ishga tushdi, worker esa yo'q. Foydalanuvchi rasm yukladi. Nima bo'ladi?

5. Vazifa muddati eslatmasi: mantiq

Eslatma 30 daqiqa oldin ketishi kerak. Demak delay = muddat − 30 daqiqa − hozir. Ikki muammo bor:

1. Vazifa tahrirlansa, eslatma ikki marta qo'shilmasin.
2. Muddat o'zgarsa, eski eslatma o'chsin, yangisi qo'yilsin.

Kanon birinchisini jobId bilan hal qildi. BullMQ'da bir xil jobId li ish navbatda bor bo'lsa, ikkinchi add yangi ish yaratmaydi. jobId = vazifa + muddat. Muddatni getTime().toString(36) bilan qisqa satrga aylantiradi. Mantiqni brauzerda tekshiramiz:

// Kanon eslatmaRejala mantig'i: kechikish va idempotent jobId.
function eslatmaRejasi(x, oldinMs, hozir) {
  const muddat = new Date(x.muddat).getTime();
  const kechikish = muddat - oldinMs - hozir;
  if (kechikish <= 0) return "otgan";
  const jobId = `eslatma-${x.vazifaId}-${muddat.toString(36)}`;
  return { jobId, kechikishDaq: kechikish / 60_000 };
}

const hozir = Date.parse("2026-10-28T09:00:00Z");
const v = { vazifaId: 7, muddat: "2026-10-28T12:00:00Z" };
console.log(eslatmaRejasi(v, 30 * 60_000, hozir));
console.log(eslatmaRejasi(v, 30 * 60_000, hozir)); // yana o'sha
console.log(eslatmaRejasi(
  { ...v, muddat: "2026-10-28T15:00:00Z" }, 30 * 60_000, hozir));
console.log(eslatmaRejasi(
  { ...v, muddat: "2026-10-28T09:20:00Z" }, 30 * 60_000, hozir));

Konsolda:

{ jobId: 'eslatma-7-mvs2b9c0', kechikishDaq: 150 }
{ jobId: 'eslatma-7-mvs2b9c0', kechikishDaq: 150 }
{ jobId: 'eslatma-7-mvs8qqo0', kechikishDaq: 330 }
otgan

• Bir xil muddat — bir xil jobId. Ikkinchi add e'tiborsiz qoladi.
• Muddat o'zgardi — jobId ham o'zgardi. Lekin eski eslatma navbatda qolib ketadi! Shuning uchun kanon har rejalashdan oldin shu vazifaning kutib turgan eslatmalarini o'chiradi (eslatmaBekor).
• Muddatgacha 20 daqiqa qolgan — 30 daqiqa oldin eslatish vaqti o'tib ketgan, ish qo'shilmaydi.

Diqqat: jobId faqat ish navbatda turgan paytda takrorni to'xtatadi. Ish bajarilib, removeOnComplete bilan o'chirilgach, o'sha jobId bilan yangi ish yana qo'shiladi. jobId — "navbatda ikki nusxa bo'lmasin" kafolati, "umrbod bir marta" emas.

6. Loyiha qadami: navbat.ts (1-qism)

Kanon modul — modullar/src/navbat/navbat.ts. U to'rt darsda (shu dars, ishonchlilik, ilg'or imkoniyatlar, cron) bosqichma-bosqich o'qiladi; commit — Cron darsida, hozircha commit qilmang. Telefon uchun qatorlarni bo'ldik va eslatmaRejala da bitta yordamchi o'zgaruvchi (vaqt) qo'shdik — mazmun kanondagidek. Formatlangan nusxani vaqtinchalik papkadagi modul nusxasida tsc --noEmit va eslint (strictTypeChecked) bilan tekshirdik — 0 xato.

// navbat.ts — BullMQ (Redis) navbatlari: vazifa muddati eslatmasi
// (delayed, idempotent jobId), bildirishnoma yetkazish,
// rejali ishlar, worker sozlamalari.
// Redis talabi: `maxmemory-policy noeviction`.
import {
  Queue, Worker, UnrecoverableError,
  type ConnectionOptions, type Job, type Processor,
} from "bullmq";
import type {
  Navbat as BildirishnomaNavbati, YetkazishIshi,
} from "../bildirishnoma/markaz.ts";

export const NAVBAT = {
  eslatma: "eslatmalar", yetkazish: "yetkazishlar", tizim: "tizim",
} as const;

export type EslatmaIshi = {
  vazifaId: number; foydalanuvchiId: number; muddat: string;
};

/** Ulanish URL'dan: BullMQ ioredis ishlatadi; worker uchun
 * `maxRetriesPerRequest: null` shart (bloklovchi buyruqlar). */
export function ulanish(redisUrl: string): ConnectionOptions {
  const u = new URL(redisUrl);
  return {
    host: u.hostname,
    port: Number(u.port || 6379),
    ...(u.password
      ? { password: decodeURIComponent(u.password) }
      : {}),
    maxRetriesPerRequest: null,
  };
}

const SUKUT = {
  attempts: 5,
  backoff: { type: "exponential", delay: 2000 },
  removeOnComplete: { age: 24 * 3600, count: 1000 },
  removeOnFail: { age: 7 * 24 * 3600 },
} as const;

export class Navbatlar {
  readonly eslatma: Queue<EslatmaIshi>;
  readonly yetkazish: Queue<YetkazishIshi>;
  readonly tizim: Queue;
  constructor(ulanishOpt: ConnectionOptions, prefiks = "vz") {
    const o = {
      connection: ulanishOpt,
      prefix: prefiks,
      defaultJobOptions: SUKUT,
    };
    this.eslatma = new Queue<EslatmaIshi>(NAVBAT.eslatma, o);
    this.yetkazish = new Queue<YetkazishIshi>(NAVBAT.yetkazish, o);
    this.tizim = new Queue(NAVBAT.tizim, o);
  }

  /** Muddatdan `oldinMs` oldin eslatma. jobId = vazifa + muddat →
   * takror qo'shish e'tiborsiz; muddat o'zgarsa — eskisi o'chiriladi.
   * O'tib ketgan vaqt — eslatma yo'q. */
  async eslatmaRejala(
    x: EslatmaIshi, oldinMs = 30 * 60_000, hozir = Date.now(),
  ): Promise<"qoshildi" | "otgan"> {
    await this.eslatmaBekor(x.vazifaId);
    const kechikish = new Date(x.muddat).getTime() - oldinMs - hozir;
    if (kechikish <= 0) return "otgan";
    const vaqt = new Date(x.muddat).getTime().toString(36);
    await this.eslatma.add("muddat", x, {
      jobId: `eslatma-${String(x.vazifaId)}-${vaqt}`,
      delay: kechikish,
    });
    return "qoshildi";
  }

  /** Vazifa o'chirildi/bajarildi/muddati o'zgardi — kutib turgan
   * eslatmalarni olib tashlash. */
  async eslatmaBekor(vazifaId: number): Promise<number> {
    let n = 0;
    for (const j of await this.eslatma.getJobs(
      ["delayed", "waiting"], 0, -1,
    )) {
      if (j.data.vazifaId === vazifaId) { await j.remove(); n++; }
    }
    return n;
  }

  /** Bildirishnoma markazi uchun Navbat: jobId = bildirishnoma +
   * kanal (takror qo'shilmaydi). */
  bildirishnomaNavbati(): BildirishnomaNavbati {
    return {
      qosh: async (ishlar) => {
        await this.yetkazish.addBulk(ishlar.map((i) => ({
          name: i.kanal,
          data: i,
          opts: {
            jobId: `b-${String(i.bildirishnomaId)}-${i.kanal}`,
          },
        })));
      },
    };
  }

  // jadvallar() — Cron darsida; yop() — uchala navbatni yopadi
}

Nimalarga e'tibor bering:

• Uchta navbat: eslatmalar, yetkazishlar, tizim. Har tur o'z navbatida — o'z worker'i, o'z concurrency'si. Eslatmalar ko'p bo'lsa ham, tizim ishlari ularning orqasida qolib ketmaydi.
• defaultJobOptions: SUKUT — har ishga 5 urinish, eksponensial kutish va eski ishlarni tozalash. Bularni keyingi darsda batafsil ko'ramiz.
• bildirishnomaNavbati() — bildirishnoma markazi faqat Navbat interfeysini (qosh(ishlar)) biladi, BullMQ'ni emas. Markazni testda soxta navbat bilan sinash mumkin edi — interfeys shuning uchun.
• getJobs([...], 0, -1) — hamma kechiktirilgan ishlarni o'qiydi. Minglab eslatmada bu sekin bo'ladi. Katta hajmda jobId ni oldindan hisoblab, getJob(id) bilan to'g'ridan o'chirgan ma'qul — kanon kichik loyiha uchun soddaroq yo'lni tanladi.

6.1 Testlar

Kanon test/navbat.test.ts da 4 ta test yozdi. Shu darsga tegishli ikkitasi (qisqartirib, formatlab):

test("eslatma: idempotent jobId, muddat o'zgarsa almashtiriladi",
  async () => {
    const hozir = Date.now();
    const x: EslatmaIshi = { vazifaId: 42, foydalanuvchiId: 1,
      muddat: new Date(hozir + 3600_000).toISOString() };
    assert.equal(await n.eslatmaRejala(x, 30 * 60_000, hozir),
      "qoshildi");
    assert.equal(await n.eslatmaRejala(x, 30 * 60_000, hozir),
      "qoshildi");
    assert.equal((await n.eslatma.getDelayed()).length, 1);
    // ... muddat 2 soatga o'zgardi → baribir 1 ta, yangi muddat bilan
    // ... 1 daqiqa qolgan vazifa → "otgan"
    assert.equal(await n.eslatmaBekor(42), 1);
  });

test("bildirishnoma navbati: bir xil (bildirishnoma, kanal) — bitta",
  async () => {
    const q = n.bildirishnomaNavbati();
    await q.qosh([{ bildirishnomaId: 5, kanal: "email" },
      { bildirishnomaId: 5, kanal: "telegram" }]);
    await q.qosh([{ bildirishnomaId: 5, kanal: "email" }]);
    assert.equal(await n.yetkazish.getWaitingCount(), 2);
  });

Kanon to'liq paketni Redis bilan yurgizdi: tsc 0, eslint 0, 82/82 test. Biz Redis'siz yurgizdik — to'plam o'tkazib yuborildi:

﹣ BullMQ navbatlari (1.1534ms) # REDIS_URL yo'q

Testlar REDIS_URL bo'lmasa o'zini o'chiradi — testlarni avtomatik yurgizadigan serverda (CI, 34-qismda) Redis bo'lmasa ham boshqa testlar yuradi. Lekin "o'tkazib yuborilgan" va "o'tdi" bir xil emas: ﹣ belgisini ko'rsangiz, test ishlamagan.

vazifalar-api ga ulanishi 28-qism tugagach: vazifa yaratilganda va muddati o'zgarganda eslatmaRejala, o'chirilganda va bajarilganda eslatmaBekor chaqiriladi; worker alohida ishchi.ts jarayonida yuradi.

7. Ko'p uchraydigan xatolar

8. Mashqlar

Mashqlar kurs/mashqlar/29/45-bullmq/ da.

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

• Ish qo'shadigan BullMQ klassi —
• Ishni bajaradigan klass —
• Kechiktirilgan ish holati —
• Worker ulanishida majburiy sozlama — maxRetriesPerRequest:
• QueueEvents ichida ishlaydigan Redis tuzilmasi — Redis

2-mashq (o'rta): hodisalardan sanoq

Kanon chiqishidagi hodisalar satridan har ishning oxirgi holatini toping va holatlar bo'yicha sanang (getJobCounts() ga o'xshash). added va progress holat emas — ularni o'tkazib yuboring. Ishora: split(" "), keyin split(":"), Map (kalit — ish id, qiymat — oxirgi holat).

const hodisalar = "added:2 waiting:2 added:3 delayed:3 waiting:3 " +
  "active:1 progress:1 completed:1 active:2 progress:2 failed:2 " +
  "active:3 progress:3 completed:3";

function sanoq(matn) {
  const oxirgi = new Map();
  for (const h of matn.split(" ")) {
    const [holat, id] = h.split(":");
    if (holat === "added" || holat === "progress") continue;
    oxirgi.set(id, holat);
  }
  const natija = {};
  for (const holat of oxirgi.values()) {
    natija[holat] = (natija[holat] ?? 0) + 1;
  }
  return natija;
}

console.log(sanoq(hodisalar));

{ failed: 1, completed: 2 }

3-mashq (qiyin): jobId bilan takrorni rad etish

SoxtaNavbat klassini yozing: addBulk(royxat) har { name, data, opts } ni ishlar (Map) ga opts.jobId kaliti bilan qo'shsin, bunday kalit bor bo'lsa — e'tiborsiz qoldirsin. Keyin kanon bildirishnomaNavbati ni (BullMQ'siz, q parametri bilan) yozib, 5-bildirishnomani email va telegram ga, keyin yana email ga qo'shing. Navbatda nechta ish bor?

class SoxtaNavbat {
  ishlar = new Map();
  addBulk(royxat) {
    for (const { name, data, opts } of royxat) {
      if (this.ishlar.has(opts.jobId)) continue; // bor — e'tiborsiz
      this.ishlar.set(opts.jobId, { name, data });
    }
  }
}

function bildirishnomaNavbati(q) {
  return {
    qosh(ishlar) {
      q.addBulk(ishlar.map((i) => ({
        name: i.kanal,
        data: i,
        opts: { jobId: `b-${i.bildirishnomaId}-${i.kanal}` },
      })));
    },
  };
}

const q = new SoxtaNavbat();
const nb = bildirishnomaNavbati(q);
nb.qosh([
  { bildirishnomaId: 5, kanal: "email" },
  { bildirishnomaId: 5, kanal: "telegram" },
]);
nb.qosh([{ bildirishnomaId: 5, kanal: "email" }]); // takror
console.log(q.ishlar.size, [...q.ishlar.keys()]);

2 [ 'b-5-email', 'b-5-telegram' ]

9. Real ishda

• BullMQ — Node'da fon vazifalari uchun eng keng tarqalgan tanlov; NestJS @nestjs/bullmq orqali uni ishlatadi.
• Ikki jarayon — API va worker — production'dagi standart tuzilma. Docker Compose'da ular ikkita servis bo'ladi (33-qism).
• Redis managed xizmatlari (Upstash, ElastiCache va boshqalar) BullMQ bilan ishlaydi, lekin noeviction siyosati va bloklovchi buyruqlar qo'llab-quvvatlanishini tekshiring (hujjat bo'yicha, sinalmagan).
• Intervyuda: "BullMQ'da ish qanday holatlardan o'tadi?", "Nega worker alohida jarayonda?", "Redis xotirasi to'lsa navbat nima bo'ladi?"

Xulosa

• Queue.add(nom, data, opts) — ish qo'shadi; Worker(nom, processor) — bajaradi; QueueEvents — Redis Streams orqali hodisalar.
• Holatlar: waiting → active → completed/failed; delayed vaqti kelgach waiting ga.
• Redis ≥ 5.0 (biz 3.0.504 da xatoni ko'rdik), noeviction, maxRetriesPerRequest: null.
• BullMQ 6: ioredis ixtiyoriy; node-redis adapteri (createNodeRedisClient) ham ishladi (kanon: 42).
• Hodisalar kuzatuv uchun — tinglovchi kech ulansa tushib qoladi; holatni navbatdan so'rang.
• Kanon: eslatma jobId = vazifa + muddat, muddat o'zgarsa eskisi o'chiriladi.

Keyingi dars: BullMQ ishonchlilik — ish yiqilsa qayta urinish va eksponensial kutish, concurrency va rate limiter, worker o'lsa "stalled" ish va UnrecoverableError.

Manbalar

• BullMQ hujjati — Queues, Workers, Events: https://docs.bullmq.io/guide/queues, https://docs.bullmq.io/guide/workers, https://docs.bullmq.io/guide/events
• BullMQ — Going to production (maxRetriesPerRequest, noeviction): https://docs.bullmq.io/guide/going-to-production
• BullMQ README (node-redis adapteri uchun redis ≥ 5.0.0) va package.json (ioredis — ixtiyoriy peer) — bullmq 6.3.19 paketi ichidan, 2026-10-28 da o'qildi
• Redis — eviction siyosatlari: https://redis.io/docs/latest/develop/reference/eviction/
• Kanon tajribasi (IlmHamroh, 2026-10-27): bullmq 6.3.19, ioredis 6.0.0, redis 6.3.0, Redis 8.10.2 (Docker), Node 24.21.0, Windows 11. Redis 3.0.504 xatosi, o'quv modellari, testlarning Redis'siz yurishi va mashq yechimlari — biz ishga tushirdik (2026-10-28).