Redis Pub/Sub va Streams: xabarlar, consumer group va ACK

Qisqacha: Redis'da ikki xil xabar tizimi bor. Pub/Sub — "yubor va unut": xabar faqat o'sha paytda ulangan obunachilarga boradi, saqlanmaydi (kanonda obunachisiz PUBLISH → 0, keyin ulangan obunachi eski xabarni olmadi). Streams — saqlanadigan jurnal: XADD bilan yoziladi, consumer group ichidagi ishchilar xabarlarni bo'lishib oladi (2 ishchi — 3 tadan, takror 0), har xabar XACK bilan tasdiqlanadi. Tasdiqlanmagan xabar pending ro'yxatida qoladi va yiqilgan ishchi o'rniga boshqasi XAUTOCLAIM bilan oladi.

Bu darsda

• Pub/Sub bilan kanalga xabar yuborasiz va obuna bo'lasiz.
• Pub/Sub'da xabar qachon yo'qolishini tushuntirasiz.
• Stream'ga yozib, XRANGE bilan o'qiysiz.
• Consumer group, XREADGROUP, XACK va pending ro'yxatini ishlatasiz.
• Yiqilgan ishchining xabarini XAUTOCLAIM bilan qayta tiklaysiz.

Oldin bilishingiz kerak: Hash, List, Set va Sorted Set, LISTEN/NOTIFY, SKIP LOCKED bilan DB-asosli navbat.

Oldingi darsda Redis'ni kesh sifatida ishlatdik va vazifalar-api ga uladik. Endi uning boshqa vazifasi — xizmatlar orasida xabar tashish. Hamma chiqishlar — kanon tajribasi (Redis 8.10.2, node-redis 6.3.0).

1. Nega bu kerak?

Aziz vazifani bajarildi deb belgiladi. Bundan keyin uchta ish kerak: (1) uning boshqa ochiq oynalariga "yangilandi" signali, (2) email bildirishnoma, (3) statistikani qayta hisoblash. Ularni HTTP so'rov ichida bajarsak — javob sekinlashadi, email serveri ishlamasa so'rov yiqiladi.

To'g'ri yo'l: API faqat "vazifa bajarildi" degan xabar (hodisa) chiqaradi. Kim qiziqsa — o'zi oladi va ishlaydi.

LISTEN/NOTIFY darsida Postgres'ning o'zidagi xabar tizimini ko'rganmiz va uning chegarasini aytganmiz: xabar saqlanmaydi, tasdiq yo'q. Redis'da ikkala variant bor — saqlanmaydigan (Pub/Sub) va saqlanadigan (Streams). Qaysi biri kerakligi bitta savolga bog'liq: xabar yo'qolsa nima bo'ladi?

• Oynadagi "yangilandi" signali yo'qolsa — foydalanuvchi sahifani yangilaydi. Zarar kichik → Pub/Sub.
• Email yoki to'lov xabari yo'qolsa — zarar katta → Streams (yoki navbat).

2. Pub/Sub — yubor va unut

2.1 Obuna va yuborish

const sub = r.duplicate(); // obuna uchun alohida ulanish
await sub.connect();
const kelgan = [];
await sub.subscribe("t59:vazifa", (xabar, kanal) =>
  kelgan.push(`${kanal}: ${xabar}`));
console.log("obunachilar:", await r.pubSubNumSub("t59:vazifa"));
console.log("PUBLISH qaytargani (nechta obunachi oldi):",
  await r.publish("t59:vazifa",
    JSON.stringify({ amal: "yaratdi", id: 7 })));

obunachilar: { 't59:vazifa': 1 }
PUBLISH qaytargani (nechta obunachi oldi): 1
keldi: [ 't59:vazifa: {"amal":"yaratdi","id":7}' ]

• subscribe(kanal, ishlovchi) — kanalga obuna; har xabar kelganda funksiya chaqiriladi.
• publish(kanal, matn) — xabar yuborish. Javob — nechta obunachi oldi (1).
• pubSubNumSub — kanalda nechta obunachi.
• Kanal oldindan yaratilmaydi — nom yetarli, xuddi LISTEN dagidek.

Naqsh (pattern) bilan obuna — bir nechta kanalga birdan:

await sub.pSubscribe("t59:*", (xabar, kanal) =>
  kelgan.push(`pattern ${kanal}: ${xabar}`));
await r.publish("t59:izoh", "yangi izoh");

pattern bilan: pattern t59:izoh: yangi izoh

2.2 Alohida ulanish

Klassik Redis protokolida (RESP2) obuna bo'lgan ulanish "obuna rejimi"ga o'tadi va boshqa buyruq bera olmaydi. Kanon node-redis 6.3 da tekshirdi:

obuna ulanishida GET: o'tdi

Xato chiqmadi: node-redis 6 Redis'ning yangi protokoli (RESP3) bilan ishlaydi va unda bu cheklov yo'q. Lekin baribir obuna uchun alohida ulanish (duplicate()) qiling: eski klientlar va protokolda bu talab, va obuna ulanishi uzoq "osilib" turadi — asosiy ulanishni band qilmasin.

2.3 Xabar yo'qolishi

obunachisiz PUBLISH: 0
qayta obunadan keyin eski xabar keldimi: false

Bu ikki qatorda Pub/Sub'ning asosiy chegarasi ko'rinadi. Kanon obunani bekor qildi, xabar yubordi (0 — hech kim olmadi), keyin qayta obuna bo'ldi. Eski xabar kelmadi. Pub/Sub hech narsani saqlamaydi: obunachi ulanmagan, qayta ulanayotgan yoki sekin bo'lsa — xabar yo'qoldi. Bu at-most-once ("ko'pi bilan bir marta") yetkazish.

Pub/Sub'ning yaxshi joyi: tezlik va oddiylik. Bir nechta API serverining xotiradagi keshini tozalash ("17-vazifa o'zgardi"), WebSocket orqali brauzerlarga jonli yangilik tarqatish (29-qismda Socket.IO Redis adapteri aynan shunday ishlaydi).

Tekshirib ko'ring: PUBLISH 0 qaytardi. Bu xatomi?

3. Streams — saqlanadigan jurnal

3.1 XADD va XRANGE

Stream — faqat oxiriga qo'shiladigan jurnal (append-only log). Har yozuv — kichik hash (maydon-qiymat) va noyob ID.

for (let i = 1; i <= 6; i++) {
  await r.xAdd("t59:hodisalar", "*", {
    amal: i % 2 ? "yaratdi" : "bajardi",
    vazifa: String(i),
  });
}

XLEN: 6 | birinchi ID: 1792909510387-0
[{"amal":"yaratdi","vazifa":"1"},{"amal":"bajardi","vazifa":"2"}]

• * — "ID'ni Redis bersin". ID shakli <millisekund>-<tartib>: 1792909510387-0 — yozilgan vaqt (Unix ms) va shu millisekunddagi tartib raqami. ID har ishga tushirishda boshqacha — vaqtga bog'liq. U doim o'sib boradi.
• XLEN — yozuvlar soni, XRANGE kalit - + — boshidan oxirigacha (- — eng kichik ID, + — eng katta). Kanon COUNT: 2 bilan birinchi ikkitasini oldi.

Pub/Sub'dan asosiy farq: yozuvlar qoladi. Kim istasa, qachon istasa, istalgan joydan o'qiy oladi.

3.2 Consumer group

Bitta stream'ni bir nechta ishchi (consumer) qayta ishlashi kerak — har xabar faqat bittasiga borsin. Buning uchun consumer group (iste'molchilar guruhi):

await r.xGroupCreate("t59:hodisalar", "bildirishnoma", "0");
const ishchi1 = await r.xReadGroup("bildirishnoma", "ishchi-1",
  { key: "t59:hodisalar", id: ">" }, { COUNT: 3 });
const ishchi2 = await r.xReadGroup("bildirishnoma", "ishchi-2",
  { key: "t59:hodisalar", id: ">" }, { COUNT: 3 });

ishchi-1 oldi: 3 | ishchi-2 oldi: 3 | takror: 0

• xGroupCreate(stream, guruh, "0") — guruh yaratildi; "0" — boshidan o'qiydi ("$" — faqat yangilarni).
• xReadGroup(guruh, ishchi, { id: ">" }) — > — "guruhga hali hech kimga berilmagan xabarlar". Guruh qaysi xabar kimga berilganini eslab qoladi.
• 6 ta xabar ikki ishchiga 3 tadan bo'lindi, takror 0.

flowchart LR
  API -->|XADD| S[(Stream: hodisalar)]
  S --> G{{Guruh: bildirishnoma}}
  G -->|1, 2, 3| W1[ishchi-1]
  G -->|4, 5, 6| W2[ishchi-2]
  W1 -->|XACK| G
  W2 -.->|ACK yo'q| P[pending ro'yxati]

Diagrammaga qarang: guruh — "dispetcher". U xabarlarni bo'lib beradi va har birini kimga berganini yozib qo'yadi. Ishchi XACK qilmaguncha xabar pending (kutilmoqda) ro'yxatida turadi. Bir stream'da bir nechta guruh bo'lishi mumkin: "bildirishnoma" guruhi va "statistika" guruhi har biri barcha xabarlarni oladi.

3.3 XACK va pending

ishchi-1 hammasini ACK: 3
ishchi-2 'yiqildi' (ACK yo'q) — XPENDING: {"pending":3,"firstId":"1792909510396-0","lastId":"1792909510402-0","consumers":[{"name":"ishchi-2","deliveriesCounter":3}]}
yangi xabar yo'q paytda XREADGROUP (BLOCK 200): null

• XACK — "qayta ishladim". ishchi-1 o'z 3 ta xabarini tasdiqladi.
• ishchi-2 xabarlarni oldi, lekin tasdiqlamay "yiqildi". XPENDING ko'rsatadi: 3 ta xabar kutilmoqda, hammasi ishchi-2 da. Diqqat: consumers ichidagi deliveriesCounter: 3 — node-redis bergan nom, bu qisqa shaklda ma'nosi "shu ishchidagi kutilayotgan xabarlar soni" (Redis hujjati shunday ta'riflaydi). Har xabar necha marta berilganini batafsil shakl (xPendingRange) ko'rsatadi — pastda shu raqamni ko'ramiz.
• BLOCK 200 — yangi xabar bo'lmasa 200 ms kutish, keyin null. Ishchilar odatda BLOCK bilan sikl ichida o'qiydi — so'rab turish (polling) shart emas.

Muhim tartib: avval ishni bajaring, keyin XACK. Teskarisi — yiqilganda xabar yo'qoladi. Bu at-least-once ("kamida bir marta") yetkazish: xabar yo'qolmaydi, lekin ikki marta qayta ishlanishi mumkin (ishchi ishni tugatib, XACK dan oldin yiqilsa). Shuning uchun ishlovchi idempotent bo'lishi kerak (Idempotency-Key darsidagi g'oya).

4. Yiqilgan ishchining xabarlari

4.1 XAUTOCLAIM

ishchi-2 qaytmadi. Uning 3 ta xabari abadiy pending'da qolmasligi kerak. XAUTOCLAIM — "berilganiga ma'lum vaqtdan ko'p bo'lgan, tasdiqlanmagan xabarlarni menga o'tkaz":

const olindi = await r.xAutoClaim("t59:hodisalar", "bildirishnoma",
  "ishchi-3", 300, "0-0", { COUNT: 10 });

ishchi-3 oldi: 3 | deliveries (necha marta berildi): [2,2,2]
pending qoldi: 0

• 300 — 300 ms dan ko'p kutilgan xabarlar. Productionda bu — ishning odatiy davomiyligidan katta qiymat (masalan 60 s).
• ishchi-3 uchta xabarni oldi. Kanon keyin xPendingRange (batafsil XPENDING) bilan har xabarning berilish sonini so'radi — hammasida 2: ishchi-2 ga bir marta, ishchi-3 ga bir marta.
• ishchi-3 XACK qildi — pending 0.

deliveriesCounter muhim: xabar 5–10 marta berilib, hammasida ishchi yiqilsa — xabarning o'zida muammo bor ("zaharli xabar"). Uni alohida stream'ga (dead letter) ko'chirish kerak — SKIP LOCKED bilan DB-asosli navbat darsidagi dead letter queue g'oyasi.

4.2 Ishchi yo'q paytdagi xabarlar

[{"amal":"ishchi yo'q paytda"}]
XTRIM MAXLEN 3: 4 o'chirildi; XLEN 3

Hech qaysi ishchi o'qimayotgan paytda yozilgan xabar saqlandi — keyin ishchi-1 uni oldi. Pub/Sub'dagi "qayta obunada eski xabar kelmadi" bilan solishtiring.

Stream cheksiz o'smasligi uchun qirqiladi: XTRIM MAXLEN 3 — oxirgi 3 ta qoldi, 4 tasi o'chdi. Odatda XADD … MAXLEN ~ 100000 — har yozuvda taxminiy chegara (~ — Redis aniq son emas, samarali o'chiradi). Yoki MINID — ma'lum vaqtdan eski yozuvlarni.

5. BullMQ va boshqalar

Streams — "qurilish bloki". Kechiktirilgan ishlar, qayta urinish jadvali, ustuvorlik, cron — bularning hammasini o'zingiz yozishingiz kerak bo'ladi. Shuning uchun Node'da fon ishlari uchun ko'pincha BullMQ ishlatiladi — u Redis ustida (List, Sorted Set, Lua) shularning hammasini tayyor beradi. BullMQ — 29-qismda; bu yerda sinalmagan. Juda katta oqimlar va bir necha kun saqlash uchun — Kafka, RabbitMQ (alohida xabar tizimlari; hozir bilish shart emas).

6. Ko'p uchraydigan xatolar

7. Mashqlar

Mashqlar kurs/mashqlar/27/59-xabarlar/ da.

1-mashq (oson): Pub/Sub yoki Streams?

• Barcha API serverlarida xotiradagi keshni tozalash signali —
• To'lov tasdiqlangandan keyin chek yuborish —
• Chatda "yozmoqda…" belgisi —
• Xabarni tasdiqlash buyrug'i —
• Yiqilgan ishchining xabarlarini olish —

2-mashq (o'rta): stream ID

Stream ID — <ms>-<tartib>. idOqi(id) funksiyasini yozing: { vaqt: Date, tartib: number } qaytarsin. idSolishtir(a, b) — Redis tartibida solishtirsin (avval ms, keyin tartib; natija manfiy/0/musbat). Kanon ID'lari bilan sinang: 1792909510387-0 sanasini ISO ko'rinishda chiqaring va 1792909510396-0, 1792909510387-1, 1792909510387-0 ni o'sish tartibida saralang. Nega oddiy satr solishtirish (<) xavfli ekanini "999-0" va "1000-0" misolida ko'rsating.

function idOqi(id) {
  const [ms, tartib] = id.split("-").map(Number);
  return { vaqt: new Date(ms), tartib };
}

function idSolishtir(a, b) {
  const [am, at] = a.split("-").map(Number);
  const [bm, bt] = b.split("-").map(Number);
  return am - bm || at - bt;
}

console.log(idOqi("1792909510387-0").vaqt.toISOString());
const idlar = [
  "1792909510396-0", "1792909510387-1", "1792909510387-0",
];
console.log(idlar.toSorted(idSolishtir));
console.log("999-0" < "1000-0", idSolishtir("999-0", "1000-0") < 0);

2026-10-25T06:25:10.387Z
[ '1792909510387-0', '1792909510387-1', '1792909510396-0' ]
false true

3-mashq (qiyin): consumer group taqlidi

Redis'siz Guruh klassini yozing: xadd(malumot) (ID — 1, 2, 3…), oqi(ishchi, soni) (hali berilmaganlardan beradi va pending'ga yozadi: { ishchi, berildi }), ack(id), pending() (ishchi bo'yicha son), autoclaim(yangiIshchi) (hamma pending'ni yangi ishchiga o'tkazib, har birining berildi sonini 1 ga oshiradi). Kanon ssenariysini takrorlang: 6 xabar, ishchi-1 va ishchi-2 3 tadan oladi, ishchi-1 ACK qiladi, keyin pending(), autoclaim("ishchi-3"), ACK, yana pending().

class Guruh {
  #xabarlar = [];
  #keyingi = 0; // hali hech kimga berilmagan birinchi indeks
  #pending = new Map(); // id -> { ishchi, berildi }

  xadd(malumot) {
    this.#xabarlar.push({ id: this.#xabarlar.length + 1, malumot });
  }
  oqi(ishchi, soni) {
    const olindi = this.#xabarlar.slice(
      this.#keyingi, this.#keyingi + soni);
    this.#keyingi += olindi.length;
    for (const x of olindi) {
      this.#pending.set(x.id, { ishchi, berildi: 1 });
    }
    return olindi.map((x) => x.id);
  }
  ack(id) {
    return this.#pending.delete(id) ? 1 : 0;
  }
  pending() {
    const s = {};
    for (const { ishchi } of this.#pending.values()) {
      s[ishchi] = (s[ishchi] ?? 0) + 1;
    }
    return s;
  }
  autoclaim(yangiIshchi) {
    const idlar = [];
    for (const [id, p] of this.#pending) {
      p.ishchi = yangiIshchi;
      p.berildi++;
      idlar.push([id, p.berildi]);
    }
    return idlar;
  }
}

const g = new Guruh();
for (let i = 1; i <= 6; i++) g.xadd({ vazifa: i });
const a = g.oqi("ishchi-1", 3);
const b = g.oqi("ishchi-2", 3);
console.log(a, b);
console.log("ACK:", a.reduce((s, id) => s + g.ack(id), 0));
console.log(g.pending());
const olindi = g.autoclaim("ishchi-3");
console.log(olindi);
for (const [id] of olindi) g.ack(id);
console.log(g.pending(), g.oqi("ishchi-1", 3));

[ 1, 2, 3 ] [ 4, 5, 6 ]
ACK: 3
{ 'ishchi-2': 3 }
[ [ 4, 2 ], [ 5, 2 ], [ 6, 2 ] ]
{} []

8. Loyiha qadami: kanon tajribasi, loyihaga kirmaydi

vazifalar-api ga bu darsda kod qo'shilmaydi. Hodisalarni oladigan tomon (WebSocket, email, fon ishchilari) hali yo'q — u 29-qismda (BullMQ, realtime). Bitta API serveri bo'lgani uchun Pub/Sub orqali kesh tozalash ham kerak emas: KeshliOmbor avlod kaliti Redis'da umumiy. O'zingiz sinab ko'ring: ikki terminalda redis-cli oching — birida SUBSCRIBE vz:vazifa, ikkinchisida PUBLISH vz:vazifa salom. Keyin XADD, XGROUP CREATE, XREADGROUP GROUP g ishchi COUNT 1 STREAMS s > va XPENDING ni qo'lda bering.

9. Real ishda

• Mikroxizmatlar — "buyurtma yaratildi" hodisasi Streams orqali ombor, to'lov va bildirishnoma xizmatlariga.
• Realtime — Socket.IO Redis adapteri (Pub/Sub yoki Streams) bir nechta server orasida xabar tarqatadi (29-qism).
• Audit va faollik lentasi — Stream + MAXLEN.
• Intervyuda: "Pub/Sub va Streams farqi?", "At-least-once nima va nega idempotentlik kerak?", "Consumer group'da ishchi yiqilsa nima bo'ladi?"

Xulosa

• Pub/Sub — tez, saqlamaydi: obunachisiz PUBLISH → 0, qayta ulanganda eski xabar yo'q (at-most-once).
• Obuna uchun alohida ulanish (duplicate()).
• Streams — saqlanadigan jurnal; ID <ms>-<tartib>, XADD, XRANGE, XTRIM/MAXLEN.
• Consumer group xabarlarni ishchilarga bo'lib beradi; XACK siz xabar pending'da qoladi.
• XAUTOCLAIM yiqilgan ishchining xabarini boshqasiga beradi; at-least-once — ishlovchi idempotent bo'lsin.

Keyingi dars: Distributed lock va rate limit Redis'da — bir nechta server bitta ishni ikki marta qilmasligi va so'rovlarni cheklash: SET NX PX, token bilan ochish, fixed va sliding window, token bucket.

Manbalar

• Redis hujjati — Pub/Sub (yetkazish semantikasi, RESP3): https://redis.io/docs/latest/develop/interact/pubsub/
• Streams (consumer group, XACK, XAUTOCLAIM, MAXLEN): https://redis.io/docs/latest/develop/data-types/streams/
• node-redis — Pub/Sub va Streams: https://github.com/redis/node-redis/blob/master/docs/pub-sub.md
• Chiqishlar — kanon o'lchovi (IlmHamroh, 2026-10): Redis 8.10.2, Node 24.21.0, node-redis 6.3.0.