MongoDB Academy · Pelajaran

Menapis Peristiwa Dengan Saluran Paip Agregasi

Pelajar akan menghantar saluran paip kepada watch() untuk menerima hanya subset peristiwa yang diperlukan oleh aplikasi mereka.

Pelajaran 3 daripada 413 langkah

Menapis Peristiwa Dengan Saluran Paip Agregasi ialah pelajaran MongoDB Academy percuma di CoddyKit. Ini ialah pelajaran 3 daripada 4. Anda boleh membaca keseluruhan pelajaran di bawah secara percuma — kemudian berlatih secara praktikal dalam pelayar menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran MongoDB Academy, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus MongoDB Academy merangkumi sejumlah 4 pelajaran.

Mengapa Menapis Peristiwa Strim Perubahan?

Tanpa penapisan, strim perubahan menghantar setiap peristiwa perubahan pada sesebuah collection. Dalam collection pengeluaran yang sibuk, ini boleh bermakna ribuan peristiwa sesaat, yang kebanyakannya tidak diperlukan oleh aplikasi anda. Penapisan pada pelayan menggunakan saluran paip pengagregatan mengurangkan trafik rangkaian, mengurangkan penggunaan CPU dalam aplikasi anda dan memastikan pengendali peristiwa anda hanya memproses peristiwa yang berkaitan. Penapisan berlaku sebelum peristiwa meninggalkan MongoDB—hanya peristiwa yang sepadan dihantar kepada klien anda.

Menghantar Saluran Paip kepada watch()

Argumen pertama kepada watch() ialah tatasusunan saluran paip pengagregatan. MongoDB menggunakan saluran paip ini pada setiap dokumen peristiwa perubahan sebelum menentukan sama ada dokumen itu perlu dihantar kepada aplikasi anda. Tidak semua peringkat pengagregatan dibenarkan dalam saluran paip strim perubahan—hanya subset tertentu yang dibenarkan, terutamanya $match, $project, $addFields, $replaceRoot dan $redact. Peringkat $group dan $lookup tidak dibenarkan.

// Only receive insert events — filter everything else
const changeStream = db.collection('orders').watch([
  {
    $match: {
      operationType: 'insert'
    }
  }
]);

for await (const change of changeStream) {
  // Only insert events arrive here
  console.log('New order:', change.fullDocument._id);
}

Menapis Mengikut Jenis Operasi

Penapisan mengikut operationType ialah penapis saluran paip yang paling biasa. Anda boleh menggunakan $match dengan satu rentetan jenis operasi atau $in untuk memadankan beberapa jenis. Ini berguna apabila aplikasi anda memerlukan insert dan update tetapi tidak memerlukan delete, atau apabila perkhidmatan mikro yang berbeza melanggan jenis operasi yang berbeza pada collection yang sama.

// React only to new orders and status updates
const stream = db.collection('orders').watch([
  {
    $match: {
      operationType: { $in: ['insert', 'update'] }
    }
  }
]);

// Or match a single type:
const deletedStream = db.collection('orders').watch([
  { $match: { operationType: 'delete' } }
]);

Menapis Peristiwa Kemas Kini Mengikut Medan yang Berubah

Anda boleh menapis peristiwa kemas kini berdasarkan medan yang diubah suai dengan membuat pertanyaan terhadap objek updateDescription.updatedFields dalam peringkat $match. Ini membolehkan anda melanggan perubahan medan tertentu sahaja—contohnya, hanya apabila medan status dokumen berubah kepada nilai tertentu. Kaedah ini lebih cekap daripada menerima semua kemas kini dan menapisnya dalam kod aplikasi.

// Only receive updates where status changed to 'shipped'
const shippedStream = db.collection('orders').watch([
  {
    $match: {
      operationType: 'update',
      'updateDescription.updatedFields.status': 'shipped'
    }
  }
], { fullDocument: 'updateLookup' });

for await (const change of shippedStream) {
  const order = change.fullDocument;
  await sendShippingEmail(order.customerId, order.trackingNumber);
}

Menapis Mengikut Nilai Medan Dokumen

Untuk peristiwa insert, anda boleh menapis berdasarkan medan dalam subdokumen fullDocument. Contohnya, terima hanya insert yang fullDocument.priority ialah 'high' atau fullDocument.region sama dengan 'US-WEST'. Penapisan pada pelayan ini amat berkuasa dalam seni bina berbilang penyewa, apabila tika aplikasi yang berbeza memerlukan peristiwa untuk subset data yang berbeza.

// Only receive inserts for high-priority orders in the US-WEST region
const priorityStream = db.collection('orders').watch([
  {
    $match: {
      operationType: 'insert',
      'fullDocument.priority': 'high',
      'fullDocument.region': 'US-WEST'
    }
  }
]);

for await (const change of priorityStream) {
  await escalateOrder(change.fullDocument);
}

Menggunakan $project untuk Membentuk Semula Peristiwa

Peringkat $project dalam saluran paip strim perubahan membentuk semula dokumen peristiwa sebelum dokumen itu dihantar kepada aplikasi anda. Anda boleh menyertakan medan yang diperlukan oleh pengendali anda sahaja, menamakan semula medan atau mengira medan terbitan. Ini mengurangkan saiz muatan yang dihantar melalui rangkaian dan memudahkan kod pengendali peristiwa anda dengan hanya memaparkan data yang diperlukan.

// Project only the fields the handler needs
const stream = db.collection('users').watch([
  { $match: { operationType: { $in: ['insert', 'update'] } } },
  {
    $project: {
      operationType: 1,
      'documentKey._id': 1,
      'updateDescription.updatedFields.email': 1,
      'fullDocument.email': 1,
      'fullDocument.name': 1
    }
  }
]);

// Handler receives trimmed events with only email and name

Menggunakan $addFields untuk Memperkaya Peristiwa

Peringkat $addFields membolehkan anda menambahkan medan yang dikira pada dokumen peristiwa perubahan. Anda boleh menambahkan cap masa apabila peristiwa diproses, menerbitkan kategori daripada jenis operasi atau mengira kunci penghalaan. Medan yang diperkaya ini disertakan dalam peristiwa yang diterima oleh aplikasi anda, lalu membolehkan kod hiliran menggunakan nilai yang telah dikira tanpa mengiranya semula.

const stream = db.collection('payments').watch([
  {
    $addFields: {
      processedAt: '$$NOW', // current timestamp as event enrichment
      eventCategory: {
        $switch: {
          branches: [
            { case: { $eq: ['$operationType', 'insert'] }, then: 'NEW_PAYMENT' },
            { case: { $eq: ['$operationType', 'update'] }, then: 'PAYMENT_UPDATE' }
          ],
          default: 'OTHER'
        }
      }
    }
  }
]);

Merangkaikan Beberapa Peringkat

Anda boleh merangkaikan beberapa peringkat saluran paip dalam saluran paip strim perubahan untuk gubahan yang berkuasa. Corak biasa ialah: $match untuk menapis peristiwa → $addFields untuk memperkaya → $project untuk meringkaskan. Setiap peringkat memproses output daripada peringkat sebelumnya. Ingat bahawa susunan peringkat penting—gunakan $match yang paling selektif dahulu untuk meminimumkan dokumen yang diproses oleh peringkat seterusnya.

const stream = db.collection('inventory').watch([
  // Stage 1: filter to updates only
  { $match: { operationType: 'update' } },
  // Stage 2: add computed field
  {
    $addFields: {
      isLowStock: {
        $lt: ['$updateDescription.updatedFields.quantity', 10]
      }
    }
  },
  // Stage 3: only pass through low-stock events
  { $match: { isLowStock: true } },
  // Stage 4: trim to essential fields
  { $project: { 'documentKey._id': 1, operationType: 1 } }
]);

Kesan Prestasi Penapisan pada Pelayan

Penapisan saluran paip pada pelayan dalam strim perubahan adalah jauh lebih cekap daripada menerima semua peristiwa dan menapisnya dalam kod aplikasi. Tanpa penapisan pada pelayan, setiap peristiwa perlu disiri dan dihantar melalui rangkaian. Dengan peringkat $match, MongoDB menilai penapis secara dalaman dan hanya menghantar peristiwa yang sepadan. Untuk collection dengan trafik tinggi, ini boleh mengurangkan penggunaan rangkaian dan CPU aplikasi beberapa kali ganda.

// Inefficient: receive all events, filter in JS
for await (const change of db.collection('orders').watch()) {
  if (change.operationType === 'insert' && change.fullDocument.total > 1000) {
    // Most events are discarded here — wasted network I/O
  }
}

// Efficient: filter server-side
for await (const change of db.collection('orders').watch([
  { $match: { operationType: 'insert', 'fullDocument.total': { $gt: 1000 } } }
])) {
  // Only matching events arrive here
}

Peringkat yang Dibenarkan berbanding Dilarang

MongoDB mengehadkan peringkat pengagregatan yang boleh digunakan dalam saluran paip strim perubahan. Dibenarkan: $match, $project, $addFields, $replaceRoot, $replaceWith, $redact. Dilarang: $group, $lookup, $unwind, $geoNear, $out, $merge dan beberapa yang lain. Percubaan menggunakan peringkat yang dilarang menyebabkan ralat apabila strim dibuka. Jika anda memerlukan transformasi yang kompleks, lakukan transformasi tersebut dalam kod aplikasi selepas menerima peristiwa yang telah ditapis.

// WRONG — $group is not allowed in change stream pipelines
db.collection('orders').watch([
  { $group: { _id: '$fullDocument.region', count: { $sum: 1 } } } // Error!
]);

// RIGHT — use only permitted stages in the pipeline
db.collection('orders').watch([
  { $match: { operationType: 'insert' } },
  { $project: { 'fullDocument.region': 1, 'fullDocument.total': 1 } }
]);

Corak Penapisan Berbilang Penyewa

Dalam aplikasi berbilang penyewa, beberapa penyewa berkongsi satu collection dengan medan tenantId. Daripada menjalankan satu strim perubahan bagi setiap penyewa (yang mahal), jalankan satu strim bagi setiap tika perkhidmatan dengan penapis $match pada fullDocument.tenantId yang dihadkan kepada penyewa yang dilayan oleh tika tersebut. Kaedah ini boleh diskalakan kepada ratusan penyewa dengan jauh lebih sedikit kursor terbuka pada pelayan MongoDB.

// Service instance handles tenants T1 and T2 only
const myTenants = ['T1', 'T2'];

const stream = db.collection('events').watch([
  {
    $match: {
      $or: [
        { 'fullDocument.tenantId': { $in: myTenants } },  // for inserts
        { 'updateDescription.updatedFields.tenantId': { $in: myTenants } }  // for updates
      ]
    }
  }
], { fullDocument: 'updateLookup' });

Semakan Ringkas

Uji pemahaman anda tentang konsep MongoDB & pangkalan data NoSQL daripada pelajaran ini.

Jawab dalam pelajaran interaktif

Ringkasan Pelajaran

Dalam pelajaran ini, anda telah mempelajari bahawa: hantar saluran paip pengagregatan sebagai argumen pertama kepada watch() untuk menapis peristiwa pada pelayan, peringkat yang dibenarkan termasuk $match, $project, $addFields, $replaceRoot dan $redact — tetapi bukan $group atau $lookup, dan penapisan pada pelayan mengurangkan trafik rangkaian dan CPU aplikasi dengan ketara berbanding penapisan pada aplikasi. Seterusnya, kita akan meneroka cara menyambung semula strim perubahan selepas gangguan menggunakan token resume.

Percuma untuk bermula

Pelajari JavaScript dengan tutor kecerdasan buatan — percuma

Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.

Kursus: 30
Pelajaran: 120

Mula belajar secara percuma Lihat kursus penuh

Soalan Lazim

Adakah pelajaran “Menapis Peristiwa Dengan Saluran Paip Agregasi” percuma?

Ya — teks penuh “Menapis Peristiwa Dengan Saluran Paip Agregasi” boleh dibaca secara percuma di web ini. Untuk berlatih secara interaktif menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7, serta membuka kunci baki kursus MongoDB Academy, tingkat taraf kepada CoddyKit PRO. Kursus MongoDB Academy merangkumi sejumlah 4 pelajaran.

Apakah yang akan saya pelajari dalam “Menapis Peristiwa Dengan Saluran Paip Agregasi”?

Pelajar akan menghantar saluran paip kepada watch() untuk menerima hanya subset peristiwa yang diperlukan oleh aplikasi mereka. Anda berlatih MongoDB Academy menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.

Adakah saya memerlukan pengalaman untuk memulakan MongoDB Academy?

Tiada pengalaman terdahulu diperlukan. Pembelajaran MongoDB Academy di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 3 daripada 4.

Berapa lamakah pelajaran “Menapis Peristiwa Dengan Saluran Paip Agregasi” diambil?

Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.

Bolehkah saya menulis dan menjalankan kod dalam pelajaran MongoDB Academy ini?

Ya. Setiap pelajaran MongoDB Academy menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.

Semua pelajaran dalam kursus ini

Membuka Strim Perubahan pada Koleksi
Struktur Dokumen Peristiwa Perubahan
Menapis Peristiwa Dengan Saluran Paip Agregasi
Menyambung Semula Strim Perubahan Selepas Gangguan

← Kembali ke MongoDB Academy