Streaming Data Waktu Nyata dengan Aliran Perubahan MongoDB

Baru-baru ini, MongoDB merilis fitur baru mulai dari versi 3.6, Change Streams. Ini memberi Anda akses instan ke data Anda yang membantu Anda tetap up to date dengan perubahan data Anda. Di dunia sekarang ini, semua orang menginginkan pemberitahuan instan daripada mendapatkannya setelah beberapa jam atau menit. Untuk beberapa aplikasi, sangat penting untuk mendorong pemberitahuan waktu nyata ke semua pengguna yang berlangganan untuk setiap pembaruan. MongoDB membuat proses ini sangat mudah dengan memperkenalkan fitur ini. Dalam artikel ini, kita akan mempelajari tentang aliran perubahan MongoDB dan aplikasinya dengan beberapa contoh.

Mendefinisikan Aliran Perubahan

Aliran perubahan tidak lain adalah aliran waktu nyata dari setiap perubahan yang terjadi dalam database atau kumpulan atau bahkan dalam penerapan. Misalnya, setiap kali pembaruan (Sisipkan, Perbarui, atau Hapus) terjadi dalam koleksi tertentu, MongoDB memicu peristiwa perubahan dengan semua data yang telah dimodifikasi.

Anda dapat menentukan aliran perubahan pada koleksi apa pun seperti operator agregasi normal lainnya menggunakan operator $changeStream dan metode watch(). Anda juga dapat menentukan aliran perubahan menggunakan metode MongoCollection.watch().

Contoh

db.myCollection.watch()

Ubah Fitur Streaming

• Memfilter Perubahan

  Anda dapat memfilter perubahan untuk mendapatkan notifikasi acara hanya untuk beberapa data yang ditargetkan.

  Contoh:

  pipeline = [
     {
       $match: { name: "Bob" }
     } ];
  changeStream = collection.watch(pipeline);

  Kode ini akan memastikan bahwa Anda mendapatkan pembaruan hanya untuk catatan yang memiliki nama sama dengan Bob. Dengan cara ini Anda dapat menulis saluran pipa apa pun untuk memfilter aliran perubahan.

• Melanjutkan Aliran Perubahan

  Fitur ini memastikan bahwa tidak ada kehilangan data jika terjadi kegagalan. Setiap respons dalam aliran berisi token resume yang dapat digunakan untuk memulai kembali aliran dari titik tertentu. Untuk beberapa kegagalan jaringan yang sering terjadi, driver mongodb akan mencoba membangun kembali koneksi dengan pelanggan menggunakan token resume terbaru. Meskipun, jika aplikasi gagal total, token resume harus dipertahankan oleh klien untuk melanjutkan streaming.

• Aliran Perubahan yang Dipesan

  MongoDB menggunakan jam logis global untuk mengurutkan semua peristiwa aliran perubahan di semua replika dan pecahan cluster mana pun sehingga, penerima akan selalu menerima pemberitahuan dalam urutan yang sama dengan perintah yang diterapkan pada database.

• Acara dengan dokumen lengkap

  MongoDB mengembalikan bagian dari dokumen yang cocok secara default. Namun, Anda dapat mengubah konfigurasi aliran perubahan untuk menerima dokumen lengkap. Untuk melakukannya, berikan { fullDocument:“updateLookup”} ke metode tontonan.
  Contoh:

  collection = db.collection("myColl")
  changeStream = collection.watch({ fullDocument: “updateLookup”})

• Daya tahan

  Aliran perubahan hanya akan memberi tahu untuk data yang dikomit ke sebagian besar replika. Ini akan memastikan bahwa peristiwa dihasilkan oleh data persistensi mayoritas yang memastikan ketahanan pesan.

• Keamanan/Kontrol Akses

  Aliran perubahan sangat aman. Pengguna dapat membuat aliran perubahan hanya pada koleksi yang izin bacanya telah mereka miliki. Anda dapat membuat aliran perubahan berdasarkan peran pengguna.

Contoh Aliran Perubahan

Dalam contoh ini, kami akan membuat aliran perubahan pada koleksi Saham untuk mendapatkan pemberitahuan bila ada harga saham yang melampaui ambang batas.

• Siapkan kluster

  Untuk menggunakan aliran perubahan, kita harus membuat kumpulan replika terlebih dahulu. Jalankan perintah berikut untuk membuat kumpulan replika simpul tunggal.

  mongod --dbpath ./data --replSet “rs”

• Memasukkan beberapa catatan dalam koleksi Saham

  var docs = [
   { ticker: "AAPL", price: 210 },
   { ticker: "AAPL", price: 260 },
   { ticker: "AAPL", price: 245 },
   { ticker: "AAPL", price: 255 },
   { ticker: "AAPL", price: 270 }
  ];
  db.Stocks.insert(docs)

• Siapkan lingkungan simpul dan pasang dependensi

  mkdir mongo-proj && cd mongo-proj
  npm init -y
  npm install mongodb --save

• Berlangganan untuk perubahan

  Buat satu file index.js dan masukkan kode berikut ke dalamnya.

  const mongo = require("mongodb").MongoClient;
  mongo.connect("mongodb://localhost:27017/?replicaSet=rs0").then(client => {
   console.log("Connected to MongoDB server");
   // Select DB and Collection
   const db = client.db("mydb");
   const collection = db.collection("Stocks");
   pipeline = [
     {
       $match: { "fullDocument.price": { $gte: 250 } }
     }
   ];
   // Define change stream
   const changeStream = collection.watch(pipeline);
   // start listen to changes
   changeStream.on("change", function(event) {
     console.log(JSON.stringify(event));
   });
  });

  Sekarang jalankan file ini:

  node index.js

• Sisipkan catatan baru di db untuk menerima pembaruan

  db.Stocks.insert({ ticker: “AAPL”, price: 280 })

  Sekarang periksa konsol Anda, Anda akan menerima pembaruan dari MongoDB.
  Contoh tanggapan:

  {
  "_id":{
  "_data":"825C5D51F70000000129295A1004E83608EE8F1B4FBABDCEE73D5BF31FC946645F696400645C5D51F73ACA83479B48DE6E0004"},
  "operationType":"insert",
  "clusterTime":"6655565945622233089",
  "fullDocument":{
  "_id":"5c5d51f73aca83479b48de6e",
  "ticker":"AAPL",
  "Price":300
  },
  "ns":{"db":"mydb","coll":"Stocks"},
  "documentKey":{"_id":"5c5d51f73aca83479b48de6e"}
  }

Di sini Anda dapat mengubah nilai parameter operationType dengan operasi berikut untuk mendengarkan berbagai jenis perubahan dalam koleksi:

• Sisipkan
• Ganti (Kecuali ID unik)
• Perbarui
• Hapus
• Batalkan (Kapan pun Mongo mengembalikan kursor yang tidak valid)

Aliran Perubahan Mode Lainnya

Anda dapat mulai mengubah aliran terhadap database dan penerapan dengan cara yang sama seperti terhadap koleksi. Fitur ini telah dirilis dari MongoDB versi 4.0. Berikut adalah perintah untuk membuka aliran perubahan terhadap database dan penerapan.

Against DB: db.watch()
Against deployment: Mongo.watch()

Kesimpulan

MongoDB Change Streams menyederhanakan integrasi antara frontend dan backend secara realtime dan mulus. Fitur ini dapat membantu Anda menggunakan MongoDB untuk model pubsub sehingga Anda tidak perlu lagi mengelola penerapan Kafka atau RabbitMQ. Jika aplikasi Anda memerlukan informasi waktu nyata maka Anda harus memeriksa fitur MongoDB ini. Saya harap postingan ini akan membantu Anda memulai aliran perubahan MongoDB.