Multi-core Node.js'da in-process event va cron nega buziladi — va to'g'ri yechim

Node.js ilovangizni cluster bilan ko'p yadroga tarqatganingiz zahoti, bitta jarayonda benuqson ishlagan ikki narsa jimgina buzila boshlaydi: in-process eventlar (EventEmitter) va dekorator-cron'lar (@Cron). Sabab bitta, yechim ham bitta. Quyida muammoning mexanizmi, nega "ba'zan ishlaydi, ba'zan yo'q" ko'rinishida namoyon bo'lishi va uni Redis/BullMQ bilan qanday to'g'ri hal qilish — amaliy tarzda.

Misollar mnazorat backend'idan (NestJS, cluster, 8 yadro; 40+ tashkilot, 400+ xodim kuzatiladi).

Asos: cluster har worker'ni alohida jarayon qiladi

Node.js bitta thread'da ishlaydi — bitta jarayon bitta CPU yadrosini band qiladi. 8 yadroni ishlatish uchun standart yo'l — cluster: primary jarayon har yadro uchun worker fork qiladi, hammasi bitta portni tinglaydi.

import * as cl from 'cluster';
import { cpus } from 'os';

const cluster = cl as unknown as cl.Cluster;

if (cluster.isPrimary) {
  const numCPUs = cpus().length; // 8
  for (let i = 0; i < numCPUs; i++) {
    cluster.fork();
  }
  cluster.on('exit', (worker) => {
    cluster.fork(); // o'lgan worker'ni qayta tiklash
  });
} else {
  bootstrap(); // har bir worker — to'liq alohida NestJS ilovasi
}

Bu yerda yodda tutiladigan bitta fakt bor, qolgan hamma narsa shundan kelib chiqadi:

Har bir worker — alohida jarayon: alohida xotira, alohida o'zgaruvchilar. Worker'lar xotiradagi hech narsani bo'lishmaydi.

Muammo 1: EventEmitter faqat o'z jarayoni ichida ishlaydi

@nestjs/event-emitter qulay: bir joyda emit, boshqa joyda @OnEvent — modullar bog'lanmaydi.

// chiqaruvchi
this.eventEmitter.emit('gps.location.received', payload);

// tinglovchi (xuddi shu jarayonda)
@OnEvent('gps.location.received')
async handleLocation(payload: GpsPayload) { /* qayta ishlash, saqlash */ }

EventEmitter — oddiy xotiradagi obyekt. emit() faqat o'z jarayoni ichidagi tinglovchini topadi. Worker A'da chiqarilgan event Worker B'gacha yetib bormaydi.

Bu amalda qanday ko'rinadi. Bizda GPS nuqtalari shu in-process oqim orqali ishlanardi. Kiruvchi har bir so'rovni OS tasodifiy worker'ga uzatadi: bitta nuqta Worker 3'ga, keyingisi Worker 7'ga. Zanjirning biror bo'g'ini "bitta jarayonda turibmiz" deb taxmin qilsa, boshqa worker'ga tushgan nuqta ulanolmay jimgina yo'qoladi — xatosiz, logsiz.

Natijada xato intermittent bo'ladi: bir foydalanuvchida ishlaydi, boshqasida yo'q; bugun ishlaydi, ertaga yo'q — chunki hammasi so'rov qaysi worker'ga tushganiga bog'liq. Bizda buni topish ~7 kun oldi, asosan e'tiborni boshida bazaga qaratganimiz uchun. Aslida ma'lumot bazaga umuman yetib bormas edi — u worker'lar orasidagi bo'shliqda yo'qolardi.

Diagnostik belgi: ma'lumot "yarim yo'lda" yo'qolsa, baza sog'lom bo'lsa va xato tasodifiy foydalanuvchi/vaqtda takrorlansa — birinchi gumon qiluvchi joy bu saqlash emas, jarayonlararo uzatish.

Muammo 2: @Cron har worker'da takrorlanadi

Xuddi shu "alohida jarayon" tabiati rejalashtirilgan vazifalarni ko'paytiradi.

@Cron(CronExpression.EVERY_10_MINUTES)
async cleanupOldData() { /* eski yozuvlarni o'chirish */ }

Har bir worker bu dekoratorni mustaqil ro'yxatdan o'tkazadi. 8 worker = vazifa har 10 daqiqada 8 marta parallel ishlaydi: bazaga 8 ta bir vaqtli o'chirish, race condition, va har qanday hisobot/bildirishnoma 8 nusxada.

Anti-pattern: "leader worker" gating

Birinchi xayolga keladigan tuzatish — bitta worker'ni "asosiy" deb belgilab, singleton ishlarni faqat o'shanga berish:

for (let i = 0; i < numCPUs; i++) {
  if (i === 0) process.env['PROCESS_WORKER_ID'] = '0';
  cluster.fork();
}
// keyin har joyda: if (process.env.PROCESS_WORKER_ID === '0') { ... }

Ishlaydi, lekin bu yechim emas, plaster. Nega:

• Qo'lda o'rash kerak — bitta joyni unutsang, dublikat qaytadi.
• Failover yo'q — "leader" o'lsa, cron'lar butunlay to'xtaydi, qayta-saylov yo'q.
• Event muammosini hal qilmaydi — kiruvchi so'rovlar baribir tasodifiy worker'ga tushadi.

Asl savol boshqacha: worker'lar bir-biri bilan qanday muvofiqlashadi?

Yechim: muvofiqlashtirishni jarayondan tashqariga chiqarish

EventEmitter ham, @Cron ham bitta jarayon uchun mo'ljallangan. Demak ko'p jarayon umumiy tashqi nuqta orqali muvofiqlashishi kerak. Amaliy vosita — Redis ustidagi BullMQ.

1. Cron'ni BullMQ scheduler'iga ko'chirish

@Cron o'rniga upsertJobScheduler. Hamma worker uni chaqiradi, lekin BullMQ schedulerni Redis'da bitta id bo'yicha saqlaydi va dublikatlarni birlashtiradi (dedupe):

@Processor(GPS_CLEANUP_QUEUE)
export class GpsSchedule extends WorkerHost implements OnModuleInit {
  constructor(
    @InjectQueue(GPS_CLEANUP_QUEUE) private cleanupQueue: Queue,
  ) {
    super();
  }

  async onModuleInit() {
    // 8 worker ham chaqiradi — Redis'da bitta scheduler qoladi
    await this.cleanupQueue.upsertJobScheduler(
      'gps-cleanup-scheduler',
      { pattern: CronExpression.EVERY_DAY_AT_MIDNIGHT, tz: 'Asia/Tashkent' },
      { name: 'cleanup-old-data', data: { type: 'cleanup-old-data' } },
    );
  }

  // job global bo'yicha faqat BITTA worker'da bajariladi
  async process(job: Job) {
    return this.handleCleanup(job);
  }
}

Natija: cron global miqyosda bir marta ishlaydi (worker soni nechta bo'lishidan qat'i nazar), PROCESS_WORKER_ID gating'iga ehtiyoj qolmaydi, failover tekin keladi.

2. Worker chegarasini kesadigan eventlarni queue orqali yuborish

Bir jarayonda chiqarilib, ishonchli ishlanishi shart bo'lgan ma'lumot in-process emit() o'rniga queue'ga boradi — endi u xotirada emas, Redis'da:

// emit o'rniga
await this.gpsQueue.add('location.received', payload);

@Processor('gps')
export class GpsProcessor extends WorkerHost {
  async process(job: Job) {
    await this.processLocation(job.data); // istalgan worker oladi
  }
}

Endi nuqta qaysi worker'ga tushishi muhim emas: u Redis navbatida turadi, qabul qiluvchi worker o'lsa ham job qoladi va qayta uriniladi (retry). Bu in-process EventEmitter bera olmaydigan kafolat.

Qachon queue ishlatmaslik kerak

Hamma event'ni queue'ga ko'chirish — qarama-qarshi xato. Agar emit va listener doimo bitta worker ichida, bitta so'rov davomida ishlasa (masalan, request ichidagi modullararo signal), in-process EventEmitter ayni o'rnida va tezroq. Queue'ni faqat ikki holatda qo'sh:

• ma'lumot worker chegarasini kesib o'tishi kerak bo'lsa, yoki
• ish kafolatli bir marta (retry bilan) bajarilishi shart bo'lsa.

Aks holda ortiqcha infratuzilma va kechikish qo'shgan bo'lasan.

Amaliy xulosa (checklist)

Ilovani cluster yoki PM2 cluster rejimiga, yoki bir nechta replikaga o'tkazishdan oldin quyidagilarni tekshir:

• `@Cron` / `setInterval` / `@Interval` — har worker'da takrorlanadi. Scheduler'ni Redis-backed queue'ga (BullMQ upsertJobScheduler) ko'chir.
• `EventEmitter` / `@OnEvent` — agar chiqaruvchi va tinglovchi turli worker'da bo'lishi mumkin bo'lsa, queue yoki Redis Pub/Sub ishlat.
• Xotiradagi holat — Map/Set/o'zgaruvchidagi cache, sessiya, rate-limit hisoblagichlari worker'lar orasida bo'linmaydi. Umumiy holatni Redis'ga chiqar.
• "Bir martalik" ishlar — PROCESS_WORKER_ID === '0' kabi qo'lda gating'ga tayanma; dedupe'ni infratuzilmaga ber.

Umumiy qoida: multi-core'da "bu holat qayerda yashashi kerak — jarayon ichidami yoki undan tashqaridami?" degan savolni har bir global ish uchun oldindan ber. Ko'p hollarda javob "tashqarida" bo'ladi — va buni keyin emas, oldindan hal qilgan arzonroq.