PostgreSQL LISTEN/NOTIFY: broker'siz realtime xabarlar Node'da

Qisqacha: LISTEN kanal — "shu kanalga xabar kelsa, menga ayt"; NOTIFY kanal, 'matn' yoki pg_notify('kanal', matn) — xabar yuborish. Postgres xabarni hamma tinglovchi ulanishlarga o'zi yetkazadi — Redis yoki boshqa broker kerak emas. Kanon: xabar 7 ms da keldi; tranzaksiya ichidagi xabar faqat COMMIT da yetkaziladi, ROLLBACK'da — umuman yo'q; payload 8 000 baytdan kam bo'lishi shart (8 000 → payload string too long). Tinglovchi doimiy alohida ulanish talab qiladi va o'chiq paytdagi xabarlar yo'qoladi — shuning uchun xabarda faqat "nima o'zgardi" (id), ma'lumotning o'zi esa jadvalda.

Bu darsda

• LISTEN, NOTIFY va pg_notify ni psql'da va Node'da ishlatasiz.
• Trigger orqali har o'zgarishda avtomatik JSON xabar yuborasiz.
• Xabar tranzaksiya bilan qanday bog'liqligini (COMMIT, ROLLBACK, takror) tushuntirasiz.
• Tinglovchi nega hovuzdan emas, alohida Client bo'lishini va uzilganda qayta ulanishni bilasiz.
• Cheklovlarni (8 000 bayt, yetkazish kafolati yo'qligi, pooler) va qachon boshqa vosita kerakligini bilasiz.

Oldin bilishingiz kerak: Connection pooling, EventEmitter asoslari, Denormalizatsiya va hisoblangan maydonlar (trigger'lar).

1. Nega bu kerak?

Aziz vazifalar ilovasini ikki joyda ochgan: kompyuterda va telefonda. Telefonda "Onamga dori olib borish" ni bajarildi deb belgiladi. Kompyuterdagi ro'yxat buni qanday biladi?

Birinchi yo'l — polling (so'rab turish): kompyuter har 5 soniyada serverdan "yangilik bormi?" deb so'raydi. 1 000 ta ochiq oyna — sekundiga 200 so'rov, ularning deyarli hammasi "yo'q" javobi. Va baribir 5 soniya kechikish.

Ikkinchi yo'l — push (itarish): o'zgarish bo'lgan zahoti server xabar beradi. Buning uchun server o'zi ham o'zgarishdan xabar topishi kerak. API serverlari bir nechta bo'lsa (telefon so'rovi 1-serverga, kompyuter oynasi 2-serverga ulangan), 1-server 2-serverga qanday aytadi?

Odatda bunga alohida xabar brokeri qo'yiladi (Redis pub/sub — 27-qism, RabbitMQ, Kafka). Lekin bizda allaqachon hamma serverlar ulangan umumiy joy bor — Postgres. Uning o'zida oddiy xabar tizimi bor: LISTEN / NOTIFY.

O'xshatish: radiostansiya. Tinglovchi radiosini kerakli to'lqinga sozlaydi (LISTEN kanal). Diktor efirga chiqadi (NOTIFY) — o'sha paytda radiosi yoqiq hamma eshitadi. Radio o'chiq bo'lsa — eshittirish o'tib ketdi, yozib qolinmaydi.

sequenceDiagram
  participant T as Telefon → API 1
  participant P as Postgres
  participant K as API 2 → kompyuter
  K->>P: LISTEN vazifa_ozgardi
  T->>P: UPDATE vazifalar … COMMIT
  P-->>K: NOTIFY {"id": 9, "amal": "UPDATE"}
  K->>P: SELECT … WHERE id = 9

Diagrammaga qarang: 2-server xabarni olgach, vazifani bazadan o'zi o'qiydi. Xabar — faqat qo'ng'iroq, ma'lumotning o'zi emas. Nega shunday — «Cheklovlar» bo'limida.

2. psql'da birinchi tajriba

Bitta psql sessiyasida: tinglaymiz, keyin o'zimizga xabar yuboramiz. Biz ishga tushirdik (PostgreSQL 17.6, psql 17.6):

LISTEN vazifa_ozgardi;
NOTIFY vazifa_ozgardi, 'salom';
SELECT pg_notify('vazifa_ozgardi', 'ikkinchi');
UNLISTEN *;
NOTIFY vazifa_ozgardi, 'eshitilmaydi';

LISTEN
NOTIFY
Asynchronous notification "vazifa_ozgardi" with payload "salom" received from server process with PID 25256.
 pg_notify 
-----------
 
(1 row)

Asynchronous notification "vazifa_ozgardi" with payload "ikkinchi" received from server process with PID 25256.
UNLISTEN
NOTIFY

• LISTEN vazifa_ozgardi — shu ulanish kanalga obuna bo'ldi. Kanal oldindan yaratilmaydi — nom yetarli.
• NOTIFY kanal, 'matn' — SQL buyrug'i; payload — oddiy satr literal.
• pg_notify('kanal', matn) — xuddi shu, lekin funksiya: kanal nomi va matnni ifoda yoki parametr bilan berish mumkin. Node'dan va trigger'dan aynan shu ishlatiladi.
• "Asynchronous notification … received from server process with PID 25256" — "asinxron xabar … 25256-jarayondan keldi". PID — yuborgan ulanish (bu yerda — o'zimiz).
• UNLISTEN * — hamma obunani bekor qilish; keyingi NOTIFY hech kimga bormaydi va xato ham bermaydi.

Maslahat: psql xabarlarni faqat navbatdagi buyruqdan keyin ko'rsatadi. Ikki terminalda sinab ko'ring: birida LISTEN, ikkinchisida NOTIFY — birinchisida xabar keyingi ; yoki bo'sh Enter'dan keyin chiqadi. Node'dagi tinglovchi esa xabarni darhol oladi.

3. Node'da tinglovchi

3.1 Alohida Client

import pg from "pg";

const tinglovchi = new pg.Client({
  connectionString: process.env.DATABASE_URL,
});
await tinglovchi.connect();
tinglovchi.on("notification", (n) => {
  console.log(`[${n.channel}] ${n.payload}`);
});
await tinglovchi.query("LISTEN vazifa_ozgardi");

"notification" — Client ning hodisasi (EventEmitter asoslari): n.channel — kanal, n.payload — matn, n.processId — yuborgan jarayon. Kanon ikkinchi ulanishdan NOTIFY vazifa_ozgardi, 'salom' yubordi:

7 ms: [vazifa_ozgardi] salom

7 millisekund — Docker ichidagi Postgres orqali bir ulanishdan boshqasiga.

Nega Pool emas? Kanon izohi aniq: pool.query() har safar boshqa ulanish olishi mumkin. pool.query("LISTEN …") bir ulanishni obuna qiladi va uni hovuzga qaytaradi — keyin u ulanish boshqa so'rovlar uchun ishlatiladi, "notification" tinglovchisi esa Pool da yo'q. Tinglovchi — doimiy band ulanish: ilova ishlab turguncha yopilmaydi. Connection pooling dagi hisobda uni alohida sanang.

3.2 Xabarlar va tranzaksiya

Kanon uch holatni tekshirdi (to'liq chiqish):

7 ms: [vazifa_ozgardi] salom
317 ms: (COMMIT oldidan — yuqoridagidan boshqa xabar kelmagan)
319 ms: [vazifa_ozgardi] tranzaksiya ichida
325 ms: [vazifa_ozgardi] takror
362 ms: [vazifa_ozgardi] {"id" : 1, "egasi" : 1, "amal" : "UPDATE"}

1. Tranzaksiya ichida pg_notify chaqirildi, keyin 300 ms kutildi — hech narsa kelmadi. Xabar faqat COMMIT dan keyin (319 ms) keldi. Bu juda to'g'ri xulq: tinglovchi xabarni olib bazadan o'qiganda, o'zgarish allaqachon ko'rinadi.
2. ROLLBACK bilan tugagan tranzaksiyadagi 'rollback qilingan' xabari — ro'yxatda yo'q. Bekor qilingan o'zgarish haqida hech kim bilmaydi.
3. Bitta tranzaksiyada 'takror' ikki marta yuborildi — bir marta keldi. Postgres bir tranzaksiyadagi bir xil (kanal + payload) xabarlarni birlashtiradi.

Tekshirib ko'ring: Ilova vazifani UPDATE qildi va xuddi o'sha tranzaksiyada pg_notify yubordi. Tinglovchi xabarni olib, darhol SELECT qildi. U eski qiymatni ko'rishi mumkinmi?

4. Trigger'dan avtomatik xabar

Har joyda pg_notify ni qo'lda chaqirish — unutish oson. Ishonchliroq yo'l — trigger (Denormalizatsiya va hisoblangan maydonlar darsida trigger'lar bilan ishlaganmiz): vazifa o'zgarsa, baza o'zi xabar beradi. Biz ishga tushirdik (PostgreSQL 17.6, holat 40):

CREATE FUNCTION vazifa_xabari() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
  PERFORM pg_notify('vazifa_ozgardi', json_build_object(
    'id', NEW.id, 'egasi', NEW.egasi_id, 'amal', TG_OP)::text);
  RETURN NULL;
END $$;
CREATE TRIGGER vazifa_xabari_trg
  AFTER INSERT OR UPDATE ON vazifalar
  FOR EACH ROW EXECUTE FUNCTION vazifa_xabari();
LISTEN vazifa_ozgardi;
BEGIN;
UPDATE vazifalar SET ustuvorlik = 'yuqori' WHERE id = 17;
SELECT 'COMMIT oldidan' AS holat;
COMMIT;

CREATE FUNCTION
CREATE TRIGGER
LISTEN
BEGIN
UPDATE 1
     holat      
----------------
 COMMIT oldidan
(1 row)

COMMIT
Asynchronous notification "vazifa_ozgardi" with payload "{"id" : 17, "egasi" : 1, "amal" : "UPDATE"}" received from server process with PID 26400.

Uchta detal:

• PERFORM — PL/pgSQL'da natijasi kerak bo'lmagan SELECT. json_build_object(…)::text — JSON satr: {"id" : 17, …} (Postgres json turi ikki nuqta atrofida bo'shliq qo'yadi; JSON.parse uchun farqi yo'q).
• TG_OP — trigger'ni nima ishga tushirdi: INSERT, UPDATE yoki DELETE.
• AFTER … RETURN NULL — AFTER trigger qaytargan qiymat e'tiborga olinmaydi.

Xabar UPDATE da emas, COMMIT dan keyin chiqdi — psql chiqishida "COMMIT oldidan" qatoridan keyin. Kanon xuddi shu trigger'ni Node tinglovchisi bilan sinadi: {"id" : 1, "egasi" : 1, "amal" : "UPDATE"}.

Diqqat: Bu trigger vazifalar dagi har qator o'zgarishida ishlaydi. 100 000 qatorli UPDATE (COPY, batch va cursor) — 100 000 ta xabar. Ommaviy amallarda bunday trigger yuklamani keskin oshiradi. Bu misol — o'quv tajribasi, kanon sxemasiga (sql/) qo'shilmagan.

5. Cheklovlar

5.1 Payload: 8 000 baytdan kam

payload 7999 bayt: yuborildi
payload 8000 bayt: XATO 22023 payload string too long

Kanon aniq chegarani topdi: 7 999 bayt — o'tdi, 8 000 — 22023 ("payload satri juda uzun"). Hujjat: payload sukut sozlamada "8 000 baytdan qisqa" bo'lishi kerak. Bayt, belgi emas: lotin harfi — 1 bayt, kirill — 2, emoji — 4.

Shuning uchun xabarga butun vazifani (matn 100 belgigacha, izohlar, teglar) solmang. Faqat identifikator va amal: {"id": 17, "amal": "UPDATE"}. Tinglovchi kerakli ma'lumotni o'zi o'qiydi.

5.2 Yetkazish kafolati yo'q

• Xabar faqat o'sha paytda ulangan tinglovchilarga boradi. Tinglovchi qayta ishga tushayotgan 2 soniyada kelgan xabarlar — yo'qoldi, hech qayerda saqlanmaydi.
• Tasdiq (acknowledgement) yo'q: tinglovchi xabarni oldi-yu, qayta ishlashdan oldin yiqildi — Postgres buni bilmaydi.
• Xabarlar navbati serverda cheklangan (hujjat bo'yicha sukut 8 GB): tinglovchi uzoq tranzaksiyada "qotib" qolsa, navbat to'ladi. Uni kuzatish — pg_notification_queue_usage() (kanonda: 0).

Xulosa: LISTEN/NOTIFY — "nimadir o'zgardi, qarab qo'y" signali uchun. "Har xabar albatta qayta ishlanishi kerak" (to'lov, email yuborish) — bu navbat vazifasi: SKIP LOCKED bilan DB-asosli navbat yoki Redis/BullMQ (27-qism).

5.3 Pooler bilan

PgBouncer transaction rejimida LISTEN ishlamaydi (hujjat jadvali; kanonda sinalmagan), NOTIFY esa ishlaydi. Amaliy naqsh: yozuvchilar pooler orqali, tinglovchi — to'g'ridan Postgres'ga bitta alohida ulanish.

6. Qayta ulanish

Tinglovchi ulanishi kunlab ochiq turadi — va bir kun albatta uziladi: Postgres qayta ishga tushadi, tarmoq uziladi, administrator ulanishni yopadi. Uzilgan Client qayta tiklanmaydi — yangisini yaratish, qayta LISTEN qilish kerak. Kutish vaqti — har urinishda ikki baravar (timers va timers/promises chuqur dagi eksponensial kutish):

import pg from "pg";

const url = process.env.DATABASE_URL;
let urinish = 0;
let joriy;

async function tinglovchiniUla(kanal, ishlov) {
  const client = new pg.Client({ connectionString: url });
  client.on("notification", (n) => ishlov(n.payload));
  client.on("error", (e) =>
    console.log("xato:", e.code ?? e.message),
  );
  client.on("end", async () => {
    console.log("ulanish uzildi, qayta ulanamiz");
    const kutish = Math.min(5000, 100 * 2 ** urinish++);
    await new Promise((r) => setTimeout(r, kutish));
    joriy = await tinglovchiniUla(kanal, ishlov);
  });
  await client.connect();
  await client.query(`LISTEN ${client.escapeIdentifier(kanal)}`);
  urinish = 0;
  console.log("tinglayapman");
  return client;
}

joriy = await tinglovchiniUla("vazifa_ozgardi", (p) =>
  console.log("keldi:", p),
);
// sinov: server tomonidan ulanishni uzamiz
const admin = new pg.Client({ connectionString: url });
await admin.connect();
await admin.query("SELECT pg_terminate_backend($1)", [
  joriy.processID,
]);
await new Promise((r) => setTimeout(r, 500));
await admin.query("NOTIFY vazifa_ozgardi, 'qayta ulangandan keyin'");

Biz ishga tushirdik (PostgreSQL 17.6, Node 24.21.0, pg 8.23.1; oxiridagi tozalash qatorlari ko'rsatilmadi):

tinglayapman
xato: 57P01
xato: ECONNRESET
xato: Connection terminated unexpectedly
ulanish uzildi, qayta ulanamiz
tinglayapman
keldi: qayta ulangandan keyin

pg_terminate_backend — administrator ulanishni majburan yopishi (Postgres qayta ishga tushganda ham shunday bo'ladi). Bitta uzilish — uchta "error": 57P01 (admin_shutdown — "administrator buyrug'i bilan uzildi"), keyin tarmoq darajasidagi ECONNRESET va drayverning xabari. "error" tinglovchisi bo'lmasa, birinchisi jarayonni yiqitardi. Keyin "end" — va bizning kod 100 ms kutib, yangi ulanish ochdi. Yangi xabar yetib keldi.

Ikki muhim nuqta:

• Kanal nomi — identifikator, parametr bo'la olmaydi, shuning uchun escapeIdentifier (Parametrli so'rovlar). LISTEN uchun pg_notify kabi funksiya yo'q — shuning uchun kanal nomlari doimiy bo'lgani ma'qul.
• Uzilish paytidagi xabarlar yo'qoldi. Qayta ulangach, ilova "o'tkazib yuborilganlarni" tiklashi kerak: masalan ro'yxatni to'liq qayta o'qish yoki yangilangan_vaqt > oxirgi_ko'rilgan so'rovi.

7. Ko'p uchraydigan xatolar

8. Mashqlar

Mashqlar kurs/mashqlar/26/47-listen/ da.

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

• Kanalga obuna bo'lish buyrug'i —
• Parametr bilan xabar yuborish funksiyasi —
• Payload chegarasi — baytdan kam
• ROLLBACK qilingan tranzaksiyadagi xabar —
• Client ning xabar hodisasi — ""

2-mashq (o'rta): xavfsiz payload

xabarTayyorla(voqea) funksiyasini yozing: obyektni JSON qiladi va baytlar sonini tekshiradi (new TextEncoder().encode(s).length — brauzer va Node'da bor). 7 999 baytdan oshsa — RangeError tashlasin. Sinang: { id: 17, amal: "UPDATE" }, matni 4 000 ta "a" bo'lgan obyekt va matni 4 000 ta "ш" (kirill) bo'lgan obyekt. Har birida baytlar sonini yoki xato matnini chiqaring.

const MAKS_BAYT = 7999;

function xabarTayyorla(voqea) {
  const s = JSON.stringify(voqea);
  const bayt = new TextEncoder().encode(s).length;
  if (bayt > MAKS_BAYT) {
    throw new RangeError(`payload ${bayt} bayt — juda katta`);
  }
  return { s, bayt };
}

const sinovlar = [
  { id: 17, amal: "UPDATE" },
  { id: 18, matn: "a".repeat(4000) },
  { id: 19, matn: "ш".repeat(4000) },
];
for (const v of sinovlar) {
  try {
    console.log("bayt:", xabarTayyorla(v).bayt);
  } catch (e) {
    console.log(e.message);
  }
}

bayt: 25
bayt: 4019
payload 8019 bayt — juda katta

3-mashq (qiyin): faqat COMMIT bo'lganlar

Ikki Client oching: tinglovchi va yozuvchi. xabarYubor(client, voqea, yakun) funksiyasini yozing: BEGIN, pg_notify (payload — JSON.stringify(voqea), parametr bilan), keyin yakun ("COMMIT" yoki "ROLLBACK"). Uchta voqea yuboring: 17-tahrir (COMMIT), 23-o'chirish (ROLLBACK), 38-bajarildi (COMMIT). 300 ms kutib, tinglovchi olgan obyektlar massivini chiqaring.

import pg from "pg";

const url = process.env.DATABASE_URL;
const tinglovchi = new pg.Client({ connectionString: url });
const yozuvchi = new pg.Client({ connectionString: url });
await tinglovchi.connect();
await yozuvchi.connect();

const kelgan = [];
tinglovchi.on("notification", (n) => {
  kelgan.push(JSON.parse(n.payload));
});
await tinglovchi.query("LISTEN vazifa_ozgardi");

async function xabarYubor(client, voqea, yakun) {
  await client.query("BEGIN");
  await client.query("SELECT pg_notify('vazifa_ozgardi', $1)", [
    JSON.stringify(voqea),
  ]);
  await client.query(yakun);
}

await xabarYubor(yozuvchi, { id: 17, amal: "tahrir" }, "COMMIT");
await xabarYubor(yozuvchi, { id: 23, amal: "o'chirish" }, "ROLLBACK");
await xabarYubor(yozuvchi, { id: 38, amal: "bajarildi" }, "COMMIT");
await new Promise((r) => setTimeout(r, 300));

console.log(kelgan);
await tinglovchi.end();
await yozuvchi.end();

[ { id: 17, amal: 'tahrir' }, { id: 38, amal: 'bajarildi' } ]

9. Node qadami: kanon tajribasi, loyihaga kirmaydi

vazifalar-api ga bu darsda kod qo'shilmaydi. Sabab: xabarni oladigan tomon hali yo'q. Signal faqat brauzerga jonli yetkazilganda foyda beradi (WebSocket yoki SSE) — bu realtime mavzusi, 29-qismda. Hozir bitta API serveri bor va u xotirada kesh saqlamaydi — "o'zgardi, keshni yangila" signaliga ham ehtiyoj yo'q.

O'zingiz sinab ko'ring (pg-tajriba papkasida, vazifalar loyihasida emas): tinglovchi skriptini bitta terminalda ishga tushiring, ikkinchi terminalda psql bilan UPDATE qiling (trigger'li variant — «Trigger'dan avtomatik xabar» bo'limidan). Keyin docker restart bilan Postgres'ni qayta ishga tushiring — qayta ulanish kodingiz ishladimi? Tajribadan keyin bazani tozalang: DROP TRIGGER vazifa_xabari_trg ON vazifalar; DROP FUNCTION vazifa_xabari();.

10. Real ishda

• Kesh tozalash: bir nechta API serverida xotiradagi kesh — "17-vazifa o'zgardi" signali kelganda hammasi keshni yangilaydi.
• Jonli panel: admin panelda yangi buyurtma paydo bo'lishi; Supabase Realtime kabi xizmatlar dastlab shu g'oyadan boshlangan (hozir ko'pi WAL asosidagi mantiqiy replikatsiyaga o'tgan).
• Kichik loyihalarda broker o'rnida: Redis'ni qo'shmasdan oddiy realtime. Yuklama o'sganda — Redis pub/sub yoki maxsus broker.
• Intervyuda: "Postgres'da pub/sub bormi?", "NOTIFY tranzaksiya bilan qanday ishlaydi?", "Nega xabarda butun obyektni yubormaslik kerak?"

Xulosa

• LISTEN kanal — obuna, NOTIFY / pg_notify(kanal, matn) — yuborish; Postgres hamma tinglovchilarga yetkazadi (kanon: 7 ms).
• Xabar faqat COMMIT'da yuboriladi; ROLLBACK — umuman yo'q; tranzaksiyadagi takror xabar — bitta.
• Trigger + pg_notify + json_build_object — har o'zgarishda avtomatik signal; ommaviy amallarda ehtiyot bo'ling.
• Payload < 8 000 bayt (bayt, belgi emas) — xabarda faqat id va amal.
• Tinglovchi — alohida doimiy Client, "error" va "end" da qayta ulanish; uzilishdagi xabarlar yo'qoladi — kafolat kerak bo'lsa, navbat.

Keyingi dars: Tranzaksiya asoslari — BEGIN/COMMIT/ROLLBACK Node'da: nega pool.query("BEGIN") tranzaksiyani boshqa ulanishda qoldiradi va to'g'ri tranzaksiyada(fn) yordamchisi.

Manbalar

• PostgreSQL 18 hujjati — NOTIFY (payload chegarasi, tranzaksiya bilan bog'liqlik, takror xabarlar): https://www.postgresql.org/docs/18/sql-notify.html
• PostgreSQL 18 hujjati — LISTEN, pg_notification_queue_usage: https://www.postgresql.org/docs/18/sql-listen.html, https://www.postgresql.org/docs/18/functions-info.html
• PostgreSQL 18 hujjati — Appendix A (57P01, 22023): https://www.postgresql.org/docs/18/errcodes-appendix.html
• node-postgres — Client hodisalari (notification, error, end): https://node-postgres.com/apis/client
• Kanon: tajriba/t47-listen.mjs, natijalar/tajriba/t47-listen.txt (PostgreSQL 18.6, Node 24.21.0, pg 8.23.1, 2026-10-24). psql misollari, qayta ulanish va 3-mashq — biz ishga tushirdik (PostgreSQL 17.6, Node 24.21.0).