Cái Giá Của Polling: Vì Sao Cách Đồng Bộ Truyền Thống Gặp Bế Tắc Khi Mở Rộng
Năm ngoái, danh mục sản phẩm của chúng tôi chạm mốc 1,8 triệu tài liệu, và đó cũng là lúc pipeline đồng bộ dữ liệu chạm ngưỡng chịu đựng. Cứ mỗi 10 giây, một worker Celery lại quét qua collection để tìm kiếm các bản ghi mới được ghi:
# Cách tiếp cận polling ngây thơ đã làm quá tải cơ sở dữ liệu của chúng tôi
last_sync = datetime.utcnow() - timedelta(seconds=10)
updates = db.products.find({"updated_at": {"$gte": last_sync}})Với tần suất 2.000 truy vấn mỗi phút, mô hình này nhanh chóng bộc lộ vấn đề:
- Lãng Phí CPU và Disk I/O: Ngay cả trong những khung giờ thấp điểm không hề có thao tác ghi nào, các lệnh quét chỉ mục vẫn chạy liên tục. Mức sử dụng CPU của node Primary luôn dao động quanh mức 45% chỉ riêng cho việc kiểm tra đồng bộ.
- Độ Trễ Dữ Liệu (Lag Gaps): Chỉ mục tìm kiếm bị trễ so với kho lưu trữ giao dịch lên đến 10 giây. Khách hàng thường xuyên thêm vào giỏ hàng những mặt hàng vốn đã hết hàng từ trước đó.
- Bỏ Sót Sự Kiện Trong Âm Thầm: Thao tác xóa cứng (hard delete) không để lại bất kỳ dấu vết nào. Một tài liệu bị xóa sẽ biến mất hoàn toàn khỏi các truy vấn
$gte, buộc chúng tôi phải chạy các tác vụ đối soát theo lô kéo dài 40 phút mỗi ngày để tìm kiếm những bản ghi mồ côi.
Các cơ sở dữ liệu quan hệ giải quyết bài toán này bằng cơ chế giải mã Write-Ahead Log như Debezium hoặc PostgreSQL replication slots. MongoDB chọn một hướng đi gọn gàng hơn: Change Streams, một cơ chế bắt sự kiện được tích hợp sẵn ngay trong replication engine.
Cách Change Streams Thu Thập Sự Kiện Từ Oplog
Change Streams cho phép các ứng dụng theo dõi các biến động dữ liệu trên phạm vi một collection, một cơ sở dữ liệu hoặc toàn bộ cluster mà không cần phải thực hiện các truy vấn đọc thủ công.
Cơ Chế Hoạt Động Bên Dưới
Mỗi replica set duy trì một capped collection nội bộ mang tên oplog.rs. Khi client thực hiện một thao tác ghi, node primary sẽ ghi lại thao tác đó vào log này và sao chép đến các node secondary. Change Streams sẽ gắn một tailable cursor vào oplog, giải mã các thao tác cơ sở dữ liệu thô thành các tài liệu JSON có cấu trúc rõ ràng.
Bạn mở một stream. Sau đó, MongoDB sẽ đẩy các thông báo qua kết nối mạng ngay khoảnh khắc thao tác ghi được xác nhận (commit) trên đa số các node.
Vì Sao Cách Này Tốt Hơn Dual-Writing
- Tính Nguyên Tử Đích Thực (True Atomicity): Việc ghi đồng thời vào cả MongoDB và một message broker bên ngoài như Kafka hay RabbitMQ tiềm ẩn rủi ro split-brain. Nếu mạng gặp sự cố giữa lúc commit cơ sở dữ liệu và lúc publish lên broker, dữ liệu của bạn sẽ bị mất đồng bộ. Change Streams đảm bảo rằng bạn chỉ xử lý những thay đổi thực sự đã được lưu trữ an toàn vào ổ đĩa.
- Khả Năng Phục Hồi Bền Bỉ: Mỗi sự kiện phát ra đều mang một resume token nhị phân dạng opaque trong trường
_idcủa nó. Nếu consumer khởi động lại sau một sự cố sập bất ngờ, chỉ cần truyền lại token đó khi kết nối lại. MongoDB sẽ tiếp tục xử lý chính xác tại điểm worker đã dừng lại. - Lọc Dữ Liệu Không Tốn Tài Nguyên Mạng: Bạn có thể chạy các stage aggregation—chẳng hạn như
$matchvà$project—ngay bên trong database engine. Cluster sẽ loại bỏ các cập nhật không liên quan trước khi gửi dữ liệu qua mạng.
Triển Khai Thực Tế: Xây Dựng Worker Đồng Bộ Bền Bỉ
Dưới đây là một worker Python đã được kiểm chứng qua thực chiến, làm nhiệm vụ theo dõi lượng hàng tồn kho và đồng bộ hóa với các dịch vụ hạ nguồn.
Yêu Cầu Tiên Quyết
Change Streams yêu cầu oplog phải đang hoạt động, đồng nghĩa với việc bạn cần có một replica set. Nếu bạn đang phát triển trên một instance độc lập (standalone) tại local, hãy chuyển đổi nó thành một replica set một node:
# Khởi động mongod với tính năng đặt tên replica set được bật
mongod --port 27017 --dbpath /var/lib/mongodb --replSet rs0
# Khởi tạo replica set bên trong mongosh
mongosh --eval 'rs.initiate()'Tiếp theo, hãy cài đặt PyMongo:
pip install pymongo1. Khởi Tạo Và Lọc Stream
Đừng stream toàn bộ hoạt động của collection vào bộ nhớ ứng dụng. Thay vào đó, hãy yêu cầu MongoDB thực thi một aggregation pipeline ở phía server. Bộ lọc dưới đây khớp với các thao tác insert, update và replace có trường stock thực sự thay đổi:
import time
import json
from pymongo import MongoClient
from pymongo.errors import PyMongoError
MONGO_URI = "mongodb://localhost:27017/?replicaSet=rs0"
client = MongoClient(MONGO_URI)
db = client["store"]
collection = db["inventory"]
# Lọc sự kiện ngay trên node cơ sở dữ liệu
pipeline = [
{
"$match": {
"operationType": {"$in": ["insert", "update", "replace"]},
"updateDescription.updatedFields.stock": {"$exists": True}
}
}
]2. Quản Lý Resume Token
Để duy trì trạng thái sau mỗi lần khởi động lại, hãy ghi resume token mới nhất vào ổ đĩa hoặc Redis sau khi xử lý thành công từng sự kiện:
TOKEN_FILE = "last_resume_token.json"
def save_resume_token(token):
if not token:
return
with open(TOKEN_FILE, "w") as f:
json.dump(token, f)
def load_resume_token():
try:
with open(TOKEN_FILE, "r") as f:
return json.load(f)
except (FileNotFoundError, json.JSONDecodeError):
return None3. Viết Vòng Lặp Reactive Consumer
Kết hợp việc quản lý token với một vòng lặp kết nối lại để giữ cho listener của bạn luôn bền bỉ trước các sự cố chập chờn mạng và hiện tượng chuyển giao vai trò cluster (step-down):
def process_change_event(event):
op_type = event["operationType"]
doc_id = event["documentKey"]["_id"]
if op_type == "insert":
product = event.get("fullDocument", {})
print(f"[SYNC] Sản phẩm đã tạo: {product.get('name')} (ID: {doc_id})")
elif op_type in ("update", "replace"):
updated = event["updateDescription"]["updatedFields"]
print(f"[SYNC] Đã cập nhật tồn kho cho {doc_id}: {updated}")
def run_stream():
resume_token = load_resume_token()
while True:
try:
print("Đang mở con trỏ change stream...")
with collection.watch(
pipeline=pipeline,
resume_after=resume_token,
full_document="updateLookup"
) as stream:
for change in stream:
process_change_event(change)
# Chỉ commit token sau khi xử lý thành công
resume_token = change["_id"]
save_resume_token(resume_token)
except PyMongoError as err:
print(f"Stream bị gián đoạn: {err}. Thử lại sau 5 giây...")
time.sleep(5)
if __name__ == "__main__":
run_stream()Kiểm Thử Pipeline
Khởi chạy script Python của bạn ở một terminal. Trong một terminal khác, kết nối qua mongosh và thêm một sản phẩm thử nghiệm mới:
use store
db.inventory.insertOne({ name: "Mechanical Keyboard", stock: 45 })Chỉ trong vòng 15 đến 25 mili-giây, worker của bạn sẽ in ra sự kiện đã được xử lý:
[SYNC] Sản phẩm đã tạo: Mechanical Keyboard (ID: 651fa4bc88a7c1e39...)Những Điểm Cần Lưu Ý Trong Môi Trường Production
Lookup Làm Tăng Tải Cho Server
Theo mặc định, sự kiện update chỉ trả về các trường bị thay đổi trong updateDescription. Việc thiết lập full_document="updateLookup" sẽ buộc MongoDB phải truy vấn lại post-image của tài liệu thực tế. Ở quy mô 5.000 bản cập nhật mỗi giây, điều này có thể làm tăng gấp đôi lưu lượng đọc. Bất cứ khi nào có thể, hãy ưu tiên xử lý dữ liệu thay đổi cục bộ (delta payloads) thay vì lookup toàn bộ tài liệu.
Oplog Rollover Gây Ra Lỗi Nghiêm Trọng
Oplog hoạt động như một bộ đệm vòng (ring buffer) có dung lượng cố định. Nếu consumer bị ngắt kết nối trong 6 giờ trong khi oplog chỉ được cấu hình cửa sổ lưu giữ 4 giờ, resume token của nó sẽ hết hạn. Khi điều đó xảy ra, server sẽ báo lỗi nghiêm trọng ChangeStreamHistoryLost. Hãy cấu hình dung lượng oplog đủ lưu lượng ghi trong ít nhất 24 đến 48 giờ, đồng thời thiết lập cảnh báo Prometheus cho độ trễ sao chép (replication lag) của consumer.
Lời Kết
Polling đánh đổi sự đơn giản lấy độ trễ cao, tải CPU lớn cùng vô số trường hợp ngoại lệ khó xử lý. MongoDB Change Streams loại bỏ hoàn toàn sự đánh đổi đó. Bằng cách truyền trực tiếp các sự kiện từ oplog qua một aggregation pipeline, bạn sẽ có được khả năng đồng bộ hóa dưới một giây và khả năng phục hồi sau sự cố mạnh mẽ mà không cần viết quá nhiều mã cồng kềnh.
