O'zgarishlar oqimi va gorizontal kengayish

Ma'lumot o'zgarganda dasturga xabar berish kerak bo'ladi: keshni yangilash, xabar yuborish, qidiruv indeksini sinxronlash. Buni so'rab turish (polling) bilan emas, o'zgarishlar oqimi bilan qilish to'g'ri.

db.buyurtma.insertMany([
  { _id: 1, mijoz: "Husanboy", holat: "yangi", summa: 250000 },
  { _id: 2, mijoz: "Malika",   holat: "yangi", summa: 180000 }
])

watch - oqimni ochish #

const oqim = db.buyurtma.watch()
oqim.disableBlockWarnings()
db.buyurtma.insertOne({ _id: 3, mijoz: "Nodira", summa: 95000 })
const h = oqim.next()
h.operationType
h.documentKey
h.fullDocument
oqim.close()

ChangeStreamCursor on buyurtma
{ acknowledged: true, insertedId: 3 }

insert
{ _id: 3 }
{ _id: 3, mijoz: 'Nodira', summa: 95000 }

Oqim ochilgandan keyin sodir bo'lgan har bir o'zgarishni beradi.

Hodisa turlari #

const oqim = db.buyurtma.watch()
oqim.disableBlockWarnings()
db.buyurtma.updateOne({ _id: 1 }, { $set: { holat: "yolda" } })
db.buyurtma.deleteOne({ _id: 2 })
const y = oqim.next()
y.operationType
y.updateDescription.updatedFields
const o = oqim.next()
o.operationType
o.documentKey
oqim.close()

ChangeStreamCursor on buyurtma
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}
{ acknowledged: true, deletedCount: 1 }

update
{ holat: 'yolda' }

delete
{ _id: 2 }

db.buyurtma.watch([], { fullDocument: "updateLookup" })

const oqim = db.buyurtma.watch([], { fullDocument: "updateLookup" })
oqim.disableBlockWarnings()
db.buyurtma.updateOne({ _id: 1 }, { $set: { holat: "yakunlandi" } })
const h = oqim.next()
h.operationType
h.fullDocument
oqim.close()

ChangeStreamCursor on buyurtma
{
  acknowledged: true,
  insertedId: null,
  matchedCount: 1,
  modifiedCount: 1,
  upsertedCount: 0
}

update
{ _id: 1, mijoz: 'Husanboy', holat: 'yakunlandi', summa: 250000 }

Oqimni filtrlash #

watch agregatsiya quvurini qabul qiladi:

const oqim = db.buyurtma.watch([
  { $match: {
      operationType: "insert",
      "fullDocument.summa": { $gte: 200000 }
  } }
])
oqim.disableBlockWarnings()
db.buyurtma.insertOne({ _id: 4, mijoz: "Aziza", summa: 50000 })
db.buyurtma.insertOne({ _id: 5, mijoz: "Kamola", summa: 400000 })
const h = oqim.next()
h.fullDocument.mijoz
h.fullDocument.summa
oqim.close()

ChangeStreamCursor on buyurtma
{ acknowledged: true, insertedId: 4 }
{ acknowledged: true, insertedId: 5 }

Kamola
400000

Birinchi hujjat filtrga tushmadi - oqim uni umuman bermadi. Filtrlash serverda bajariladi, shuning uchun keraksiz ma'lumot tarmoqqa chiqmaydi.

Resume token - uzilishdan keyin davom etish #

Har bir hodisada _id bor - bu davom etish belgisi:

const oqim = db.buyurtma.watch()
oqim.disableBlockWarnings()
db.buyurtma.insertOne({ _id: 6, mijoz: "Dilnoza", summa: 77000 })
const h = oqim.next()
typeof h._id._data
h._id._data.length > 0
oqim.close()

ChangeStreamCursor on buyurtma
{ acknowledged: true, insertedId: 6 }

string
true

Uni saqlab qo'yib, keyinroq shu joydan davom etish mumkin:

db.buyurtma.watch([], { resumeAfter: saqlangan_token })

Butun baza va klaster bo'yicha kuzatish #

const oqim = db.watch()
oqim.disableBlockWarnings()
db.buyurtma.insertOne({ _id: 7, mijoz: "Husanboy", summa: 30000 })
db.mijoz.insertOne({ _id: 1, ism: "Malika" })
const a = oqim.next()
a.ns.coll
const b = oqim.next()
b.ns.coll
oqim.close()

ChangeStreamCursor on dokon
{ acknowledged: true, insertedId: 7 }
{ acknowledged: true, insertedId: 1 }

buyurtma

mijoz

ns.coll - hodisa qaysi to'plamda sodir bo'lganini ko'rsatadi.

Gorizontal kengayish - bo'laklash #

Bitta server yetmay qolganda ikki yo'l bor:

MongoDB da gorizontal kengayish sharding deb ataladi. To'plam bo'lak kaliti (shard key) bo'yicha bir nechta serverga taqsimlanadi:

sh.enableSharding("dokon")
sh.shardCollection("dokon.buyurtma", { mijoz_id: "hashed" })

Dastur mongos ga ulanadi va bo'laklar haqida hech narsa bilmaydi - so'rovlar o'zgarishsiz qolaveradi.

Xulosa #

• O'zgarishlar oqimi so'rab turishning o'rnini bosadi: server o'zi xabar beradi.
• watch() oqim ochilgandan keyingi o'zgarishlarni beradi.
• operationType hodisa turini bildiradi: insert, update, delete va boshqalar.
• update hodisasi faqat o'zgargan maydonlarni beradi.
• To'liq hujjat uchun fullDocument: "updateLookup" kerak - lekin u hozirgi holatni o'qiydi.
• watch agregatsiya quvurini qabul qiladi - filtrlash serverda bajariladi.
• Har hodisadagi _id - resume token; uni saqlamasangiz uzilish paytidagi o'zgarishlar yo'qoladi.
• Oqim oplog ustida ishlaydi, shuning uchun replika to'plami talab qilinadi.
• Gorizontal kengayish - sharding; dastur mongos ga ulanadi va bo'laklarni ko'rmaydi.
• Bo'lak kaliti eng muhim qaror: monoton o'suvchi kalit butun yukni bitta shard ga to'playdi.

Keyingi bo'limda xavfsizlik va zaxira nusxani ko'ramiz.