Event-driven pemrograman dengan Amazon DocumentDB dan Java

Event-driven pemrograman dalam konteks Amazon DocumentDB mewakili pola arsitektur yang kuat di mana perubahan database berfungsi sebagai generator peristiwa utama yang memicu logika dan proses bisnis berikutnya. Ketika catatan disisipkan, diperbarui, atau dihapus dalam koleksi DocumentDB, perubahan ini bertindak sebagai peristiwa yang secara otomatis memulai berbagai proses hilir, pemberitahuan, atau tugas sinkronisasi data. Pola ini sangat berharga dalam sistem terdistribusi modern di mana beberapa aplikasi atau layanan perlu bereaksi terhadap perubahan data secara real-time. Mekanisme utama penerapan pemrograman berbasis peristiwa di DocumentDB adalah dengan perubahan aliran.

Bekerja dengan perubahan aliran dari aplikasi Java

watch()Metode dalam driver Java MongoDB adalah mekanisme utama untuk memantau perubahan data real-time di Amazon DocumentDB. watch()Metode ini dapat dipanggil oleh MongoClient, MongoDatabase, dan MongoCollection objek.

watch()Metode ini mengembalikan instance ChangeStreamIterable yang mendukung berbagai opsi konfigurasi, termasuk pencarian dokumen lengkap untuk pembaruan, menyediakan token resume dan stempel waktu untuk keandalan, dan tahapan agregasi pipeline untuk memfilter perubahan.

ChangeStreamIterablemengimplementasikan antarmuka inti Java Iterable dan dapat digunakan denganforEach(). Untuk menangkap peristiwa menggunakanforEach(), berikan fungsi callback ke forEach() yang memproses peristiwa yang diubah. Cuplikan kode berikut menunjukkan cara membuka aliran perubahan pada koleksi untuk memulai pemantauan peristiwa perubahan:

ChangeStreamIterable < Document > iterator = collection.watch();
iterator.forEach(event - > {
    System.out.println("Received a change: " + event);
});

Cara lain untuk melintasi semua peristiwa perubahan adalah dengan membuka kursor yang mempertahankan koneksi ke cluster dan terus menerima peristiwa perubahan baru saat terjadi. Untuk mendapatkan kursor aliran perubahan, gunakan cursor() metode ChangeStreamIterable objek. Contoh kode berikut menunjukkan cara memantau peristiwa perubahan menggunakan kursor:

try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch().cursor()) {
    System.out.println(cursor.tryNext());
}

Sebagai praktik terbaik, buat pernyataan try-with-resource atau tutup kursor secara manual. MongoChangeStreamCursor Memanggil cursor() metode pada ChangeStreamIterable mengembalikan a MongoChangeStreamCursor yang dibuat di atas ChangeStreamDocument objek.

Kel ChangeStreamDocument as adalah komponen penting yang mewakili peristiwa perubahan individu dalam aliran. Ini berisi informasi terperinci tentang setiap modifikasi, termasuk jenis operasi (menyisipkan, memperbarui, menghapus, mengganti), kunci dokumen, informasi namespace, dan konten dokumen lengkap bila tersedia. Kelas menyediakan metode untuk mengakses berbagai aspek peristiwa perubahan, seperti getOperationType() untuk menentukan jenis perubahan, getFullDocument() untuk mengakses status dokumen lengkap, dan getDocumentKey() untuk mengidentifikasi dokumen yang dimodifikasi.

Ob ChangeStreamDocument jek menyediakan dua informasi penting, token resume dan waktu peristiwa perubahan.

Token resume dan operasi berbasis waktu di aliran perubahan DocumentDB menyediakan mekanisme penting untuk menjaga kontinuitas dan mengelola akses data historis. Token resume adalah pengenal unik yang dihasilkan untuk setiap peristiwa perubahan, berfungsi sebagai bookmark yang memungkinkan aplikasi memulai ulang pemrosesan aliran perubahan dari titik tertentu setelah pemutusan atau kegagalan. Ketika kursor aliran perubahan dibuat, kursor dapat menggunakan token resume yang disimpan sebelumnya melalui resumeAfter() opsi, memungkinkan aliran untuk melanjutkan dari tempat tinggalnya daripada memulai dari awal atau kehilangan acara.

Time-based operasi dalam aliran perubahan menawarkan pendekatan yang berbeda untuk mengelola titik awal pemantauan peristiwa perubahan. startAtOperationTime()Opsi ini memungkinkan Anda untuk mulai menonton perubahan yang terjadi pada atau setelah stempel waktu tertentu. Fitur berbasis waktu ini sangat berharga dalam skenario yang membutuhkan pemrosesan data historis, pemulihan point-in-time, atau sinkronisasi antar sistem.

Contoh kode berikut mengambil peristiwa yang terkait dengan dokumen insert, menangkap token resume, dan kemudian menyediakan token tersebut untuk mulai memantau peristiwa setelah peristiwa insert. Peristiwa dikaitkan dengan peristiwa pembaruan, kemudian mendapatkan waktu cluster ketika pembaruan terjadi dan menggunakan stempel waktu itu sebagai titik awal untuk pemrosesan lebih lanjut.

BsonDocument resumeToken;
BsonTimestamp resumeTime;

try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch().cursor()) {
    System.out.println("****************** Insert Document *******************");
    ChangeStreamDocument < Document > insertChange = cursor.tryNext();
    resumeToken = insertChange.getResumeToken();
    printJson(cursor.tryNext());
}
try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch()
    .resumeAfter(resumeToken)
    .cursor()) {
    System.out.println("****************** Update Document *******************");
    ChangeStreamDocument < Document > insertChange = cursor.tryNext();
    resumeTime = insertChange.getClusterTime();
    printJson(cursor.tryNext());
}
try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch()
    .startAtOperationTime(resumeTime)
    .cursor()) {
    System.out.println("****************** Delete Document *******************");
    printJson(cursor.tryNext());
  }

Secara default, peristiwa perubahan pembaruan tidak menyertakan dokumen lengkap dan hanya menyertakan perubahan yang dibuat. Jika Anda perlu mengakses dokumen lengkap yang diperbarui, Anda dapat memanggil fullDocument() metode pada ChangeStreamIterable objek. Perlu diingat bahwa ketika Anda meminta dokumen lengkap untuk dikembalikan untuk acara pembaruan, dokumen tersebut akan mengembalikan dokumen yang ada pada saat panggilan untuk mengubah aliran dilakukan.

Metode ini mengambil FullDocument enum sebagai parameter. Saat ini, Amazon DocumentDB hanya mendukung DEFAULT dan UPDATE_LOOKUP nilai. Cuplikan kode berikut menunjukkan cara meminta dokumen lengkap untuk acara pembaruan saat mulai menonton perubahan:

try (MongoChangeStreamCursor < ChangeStreamDocument < Document >> cursor = collection.watch().fullDocument(FullDocument.UPDATE_LOOKUP).cursor())