Bài 4: Cách sử dụng Backend (NodeJS hoặc Golang) để đọc và xử lý message từ Kafka do Debezium đẩy ra.

1. Bài toán thực tế

Giả sử bạn đang làm tính năng tìm kiếm user. Bạn muốn mỗi khi có user mới đăng ký (hoặc cập nhật thông tin) ở MySQL, thì ngay lập tức Backend của bạn phải nhận được thông tin đó để ghi vào Elasticsearch/Redis.

Thay vì query trực tiếp vào DB, Backend của chúng ta sẽ đóng vai trò là một Kafka Consumer, "ngồi hóng" trên topic dbserver1.inventory.users.

2. Thực hành Code Node.js

Bước 1: Khởi tạo Project Node.js

Tạo thư mục mới cho backend và cài đặt package kafkajs (thư viện cực kỳ phổ biến để làm việc với Kafka trong hệ sinh thái Node):

mkdir cdc-backend
cd cdc-backend
npm init -y
npm install kafkajs

Bước 2: Viết code Consumer

Tạo một file tên là consumer.js và dán đoạn code sau vào. File này sẽ kết nối đến cụm Kafka đang chạy ở cổng 9092 mà chúng ta đã dựng từ các bài trước:

const { Kafka } = require('kafkajs');

// 1. Cấu hình kết nối Kafka
const kafka = new Kafka({
  clientId: 'user-sync-service',
  brokers: ['localhost:9092'] 
});

const consumer = kafka.consumer({ groupId: 'sync-elasticsearch-group' });

const run = async () => {
  // 2. Kết nối và Đăng ký theo dõi Topic
  await consumer.connect();
  console.log('✅ Đã kết nối tới Kafka!');
  
  await consumer.subscribe({ 
    topic: 'dbserver1.inventory.users', 
    fromBeginning: true 
  });

  // 3. Xử lý từng message nhận được
  await consumer.run({
    eachMessage: async ({ topic, partition, message }) => {
      // Debezium gửi dữ liệu ở dạng chuỗi JSON
      if (!message.value) return;
      
      const event = JSON.parse(message.value.toString());
      
      // Lấy phần payload (bỏ qua phần schema đính kèm)
      const payload = event.payload; 
      if (!payload) return;

      const operation = payload.op;
      
      console.log(`\n--- Nhận được Event mới từ Kafka ---`);
      
      // Phân loại hành động (c: Create, u: Update, d: Delete)
      switch (operation) {
        case 'c': // CREATE
          console.log('🟢 [INSERT] User mới:', payload.after);
          // TODO: Code lưu payload.after vào Elasticsearch ở đây
          break;
          
        case 'u': // UPDATE
          console.log('🟠 [UPDATE] User thay đổi.');
          console.log('  - Dữ liệu cũ:', payload.before);
          console.log('  - Dữ liệu mới:', payload.after);
          // TODO: Code cập nhật vào Elasticsearch ở đây
          break;
          
        case 'd': // DELETE
          console.log('🔴 [DELETE] User bị xóa:', payload.before);
          // TODO: Code xóa user khỏi Elasticsearch ở đây
          break;
          
        case 'r': // READ (Lần đầu snapshot DB)
          console.log('🔵 [SNAPSHOT] Dữ liệu có sẵn:', payload.after);
          break;
          
        default:
          console.log('Trạng thái không xác định:', operation);
      }
    },
  });
};

run().catch(console.error);

Bước 3: Chạy thử hệ thống

Mở Terminal và chạy file vừa tạo:

node consumer.js

Lúc này, vì ta để fromBeginning: true, bạn sẽ thấy console in ra dữ liệu của các user đã có sẵn trong DB với cờ 🔵 [SNAPSHOT].

Bước 4: Mô phỏng Update dữ liệu dưới DB (Kiểm thử Real-time)

Hãy giữ nguyên Terminal đang chạy consumer.js. Mở một Terminal khác, nhảy vào MySQL và sửa tên của một User:

docker-compose exec mysql bash -c 'mysql -u root -pdebezium -e "
USE inventory;
UPDATE users SET name = ''Hoang VIP'' WHERE id = 1;
"'

Ngay lập tức, bạn sẽ thấy Terminal chạy Node.js nhảy log báo hiệu sự kiện Update 🟠 với dữ liệu cũ là Hoang và dữ liệu mới là Hoang VIP. Quá trình đồng bộ diễn ra gần như lập tức và hoàn toàn tách biệt với source code API chính của bạn!

3. Tổng kết Series

Tuyệt vời! Qua các bài học, bạn đã đi từ một khái niệm mơ hồ đến việc xây dựng được một luồng xử lý Change Data Capture (CDC) hoàn chỉnh:

1. Hiểu tại sao cần CDC thay vì Dual-write.
2. Dựng hạ tầng Zookeeper, Kafka, Kafka Connect bằng Docker.
3. Kích hoạt Debezium MySQL Connector bám sát file binlog.
4. Dùng Backend chắt lọc dữ liệu từ Kafka để phục vụ các bài toán Microservices.

Kiến trúc này chính là xương sống của rất nhiều hệ thống lớn hiện nay khi họ cần xử lý stream dữ liệu khối lượng khổng lồ.