CDC với Debezium & cache invalidation

Làm hệ phân tán, microservices, kiểu gì cũng đụng hai bài toán khó nhằn này:

1. Làm sao giữ database chính (PostgreSQL/MySQL) và cache (Redis) luôn khớp nhau?
2. Làm sao bắn một domain event sang message broker (Kafka) chắc chắn 100%, ngay khi bản ghi trong database commit xong?

Cách đầu tiên ai cũng nghĩ tới là Dual-Write (ghi kép):

@Transactional
public void updateUser(User user) {
    userRepository.save(user);          // 1. Ghi vào Database
    redisTemplate.delete(user.getId()); // 2. Xóa Cache Redis
    kafkaTemplate.send("user-events", user.getId(), new UserUpdatedEvent(user)); // 3. Gửi Kafka
}

Nhìn thì rất hợp lý. Nhưng lên production, đoạn này là quả bom nổ chậm. @Transactional chỉ bao được database: lệnh xóa Redis và lệnh gửi Kafka chạy ngay lập tức, trước khi transaction commit, và không rollback được. Rộng hơn, Redis và Kafka không tham gia chung một transaction với PostgreSQL được; muốn ép cả ba nhất quán thì phải trả giá đắt về hiệu năng hoặc tính sẵn sàng.

Phần dưới nói về cách Change Data Capture (CDC) với Debezium hoạt động, cách PostgreSQL logical decoding chạy bên dưới, và cách dựng một đường ống đồng bộ dữ liệu gần real-time đủ chắc để chạy production.

# rủi ro của Dual-Write

Khi một luồng xử lý cố cập nhật cùng lúc nhiều hệ lưu trữ độc lập với nhau, sớm muộn gì nó cũng gãy ở 2 chỗ:

# lỗi 1: thứ tự ghi bị đảo (race condition do concurrency)

• Mạng và CPU không chạy đều nhau, nên Luồng B có thể commit vào DB sau Luồng A nhưng lại tới Redis trước.
• Hậu quả: giá trị cũ của Luồng A đè lên giá trị mới của Luồng B trong cache. Từ lúc đó, ai đọc cache cũng thấy dữ liệu cũ cho tới khi key hết hạn (TTL).
• Xóa cache thay vì ghi cache (như đoạn code ở trên) cũng không thoát: xóa xảy ra trước commit, một request đọc chen vào giữa sẽ cache miss, đọc giá trị cũ từ DB (transaction chưa commit) rồi nạp lại vào cache.

# lỗi 2: hỏng giữa chừng (partial failure & process crash)

• DB ghi xong, nhưng chưa kịp gọi Redis/Kafka thì JVM chết (OOM, K8s evict pod, đứt mạng):
  • Database đã commit dữ liệu mới.
  • Redis vẫn giữ dữ liệu cũ.
  • Kafka không nhận được event nào.
  • Giao dịch phân tán gãy làm đôi, mà chẳng có cơ chế rollback bù trừ nào.
• Chiều ngược lại cũng tệ không kém: event đã gửi sang Kafka, nhưng sau đó transaction DB bị rollback (vi phạm constraint, deadlock). Các service khác đã xử lý một sự kiện không hề xảy ra.

# kiến trúc CDC: database là single source of truth

Change Data Capture (CDC) làm ngược lại hoàn toàn:

• App chỉ ghi vào database, trong đúng một transaction ACID cục bộ.
• Transaction log của database (Write-Ahead Log, WAL của PostgreSQL hay binlog của MySQL) được coi là dòng dữ liệu ghi lại mọi thay đổi đã commit.
• Một công cụ CDC chuyên dụng (Debezium) đọc log này rồi bắn event ra Kafka. Độ trễ end-to-end phụ thuộc cấu hình và tải, thường cỡ vài chục tới vài trăm mili-giây; đủ nhanh cho cache và search, nhưng vẫn là eventual consistency.

# vì sao CDC dẹp được race condition?

• Mọi bản ghi trong WAL có một vị trí gọi là Log Sequence Number (LSN), và logical decoding phát các transaction ra theo đúng thứ tự commit.
• Debezium đọc theo thứ tự đó và lưu lại LSN đã xử lý (offset), sập rồi chạy lại thì đọc tiếp từ chỗ cũ.
• Lúc đẩy vào Kafka, mặc định key của message là primary key của bản ghi (ở đây là user_id). Các message cùng key vào cùng một partition, và Kafka giữ đúng thứ tự trong từng partition.
• Vậy là thay đổi của cùng một user không bị đảo thứ tự nữa.

Hai điều cần nhớ. Một là Debezium giao at-least-once: connector khởi động lại có thể bắn lại vài event, nên consumer phải idempotent (xóa cache hai lần thì vô hại, nhưng cộng điểm hai lần thì không). Hai là CDC không xóa hẳn được race giữa luồng đọc và luồng xóa cache: request đọc DB lúc dữ liệu chưa đổi, rồi nạp vào cache sau khi CDC đã xóa, thì cache vẫn giữ giá trị cũ. Khả năng xảy ra thấp hơn nhiều, nhưng vẫn nên giữ TTL làm lưới an toàn, hoặc ghi kèm version (vd LSN, updated_at) để không nạp đè bản mới bằng bản cũ.

# PostgreSQL logical decoding: pgoutput & lsn tracking

Với PostgreSQL, Debezium không chạy SELECT định kỳ để dò thay đổi (polling). Nó dùng logical decoding: PostgreSQL tự giải mã WAL thành các thay đổi theo hàng (INSERT/UPDATE/DELETE) qua một output plugin. pgoutput là plugin có sẵn từ PostgreSQL 10 (chính là plugin của logical replication), nên không phải cài thêm gì trên server. Debezium kết nối như một client replication, tạo một replication slot để PostgreSQL nhớ nó đã đọc tới đâu, và một publication để chọn bảng nào được phát ra.

# mấy cấu hình bắt buộc phải có trong postgresql.conf

wal_level = logical               # Bắt buộc để PostgreSQL ghi đủ metadata vào WAL
max_replication_slots = 10        # Số lượng replication slots tối đa
max_wal_senders = 10              # Số lượng tiến trình truyền WAL stream
max_slot_wal_keep_size = 10240MB  # RÀO CHẮN AN TOÀN: Giới hạn tối đa 10GB cho slot bị treo!

Đổi wal_level phải restart PostgreSQL. max_slot_wal_keep_size có từ PostgreSQL 13. Trên dịch vụ managed (RDS, Cloud SQL...) thì bật logical replication qua parameter group/flag của nhà cung cấp chứ không sửa file trực tiếp.

# transactional outbox pattern với Debezium smt (single message transformation)

Khi cần bắn các event nghiệp vụ phức tạp sang microservice khác (ví dụ OrderCreatedEvent chứa danh sách mặt hàng), đọc thẳng bảng orders sẽ dễ vỡ mỗi khi cấu trúc database thay đổi.

Transactional Outbox Pattern đi cùng Debezium Outbox Event Router (SMT) là cách làm phổ biến hiện nay: ứng dụng ghi event vào một bảng outbox trong cùng transaction với dữ liệu nghiệp vụ, Debezium đọc bảng outbox từ WAL rồi chuyển thành message Kafka. Event có định dạng do mình thiết kế, không lộ cấu trúc bảng nội bộ ra ngoài.

# cấu hình Debezium outbox event router (smt)

{
  "name": "order-outbox-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres-master",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/secrets/db.properties:password}",
    "database.dbname": "order_db",
    "topic.prefix": "orderdb",
    "table.include.list": "public.outbox_table",
 
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.route.topic.replacement": "outbox.event.${routedByValue}",
    "transforms.outbox.table.fields.additional.placement": "event_type:header:eventType"
  }
}

Mặc định Event Router đọc các cột tên id, aggregatetype, aggregateid, type, payload. Bảng ở trên đặt tên có gạch dưới nên phải khai báo lại route.by.field và table.field.event.key. Message ra Kafka có value là nội dung cột payload, key là aggregate_id, nên mọi event của cùng một đơn hàng giữ đúng thứ tự. topic.prefix là bắt buộc từ Debezium 2.0.

Bảng outbox sẽ phình mãi nếu không dọn. Cách hay dùng là INSERT rồi DELETE ngay trong cùng transaction: WAL vẫn có bản INSERT để Debezium đọc, còn bảng thì luôn trống. Event Router tự bỏ qua các event DELETE trên bảng outbox.

# chống cache stampede (dogpiling effect)

Khi Debezium thấy database thay đổi và bắn tín hiệu xóa cache (Cache Invalidation), lại nảy ra một rủi ro mới: Cache Stampede, request ùa xuống DB cùng lúc.

Giả sử key của một món hàng đang cực hot (Flash Sale) vừa bị xóa khỏi Redis:

• Trong 1 giây, 10,000 request đồng thời vào và đều bị cache miss.
• Cả 10,000 request cùng lao xuống database query lại.
• Database quá tải ngay (CPU 100%, connection pool cạn) và sập.

Có vài cách chống: chỉ cho một request được nạp lại (distributed lock, như bên dưới); gộp request ngay trong process (vd CompletableFuture dùng chung theo key, hoặc Caffeine AsyncLoadingCache); hoặc thay vì xóa thì cho consumer CDC ghi luôn giá trị mới vào cache / đánh dấu "cũ" để request vẫn đọc tạm bản cũ trong lúc một luồng nạp lại.

# xử lý cache miss an toàn bằng distributed lock

Vài chi tiết dễ sai trong đoạn dưới: lock phải mang token riêng và chỉ xóa khi còn là của mình (lock hết hạn trong lúc query DB chậm thì luồng khác đã lấy rồi); luồng chờ phải có giới hạn số lần thử, không đệ quy vô hạn.

@Service
public class ResilientProductCatalogService {
 
    // Chỉ xóa lock khi giá trị vẫn là token của mình (check-and-delete nguyên tử bằng Lua)
    private static final RedisScript<Long> RELEASE_LOCK = RedisScript.of(
        "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
        Long.class);
 
    @Autowired
    private StringRedisTemplate redisTemplate;
    @Autowired
    private ProductRepository productRepository;
 
    public ProductDto getProduct(String productId) {
        String cacheKey = "product:" + productId;
        String lockKey = "lock:" + cacheKey;
 
        for (int attempt = 0; attempt < 20; attempt++) {
            // 1. Thử lấy từ Cache
            String cachedJson = redisTemplate.opsForValue().get(cacheKey);
            if (cachedJson != null) {
                return deserialize(cachedJson);
            }
 
            // 2. Cache Miss: Cạnh tranh lấy Lock để truy cập DB (TTL 3 giây, giá trị là token riêng)
            String token = UUID.randomUUID().toString();
            Boolean acquired = redisTemplate.opsForValue()
                .setIfAbsent(lockKey, token, Duration.ofSeconds(3));
 
            if (Boolean.TRUE.equals(acquired)) {
                try {
                    // Chỉ duy nhất 1 luồng được phép load từ DB và nạp lại Cache!
                    ProductDto product = productRepository.findById(productId)
                        .map(this::toDto)
                        .orElse(ProductDto.empty());
 
                    // Không có sản phẩm thì chỉ cache ngắn (negative cache)
                    Duration ttl = product.isEmpty() ? Duration.ofSeconds(30) : Duration.ofMinutes(15);
                    redisTemplate.opsForValue().set(cacheKey, serialize(product), ttl);
                    return product;
                } finally {
                    // Lock có thể đã hết hạn và bị luồng khác lấy: không được xóa nhầm lock của họ
                    redisTemplate.execute(RELEASE_LOCK, List.of(lockKey), token);
                }
            }
 
            // 3. Các luồng khác chờ 50ms rồi thử đọc lại từ Cache
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
 
        // Chờ quá lâu (luồng giữ lock chết hoặc DB chậm): đọc thẳng DB thay vì đệ quy vô hạn.
        // Hệ thống nhạy cảm hơn có thể trả lỗi 503 ở đây để bảo vệ DB.
        return productRepository.findById(productId)
            .map(this::toDto)
            .orElse(ProductDto.empty());
    }
}

# so sánh các cách làm

# tổng kết

Dùng Change Data Capture (CDC) với Debezium là chuyển từ lối nghĩ "ứng dụng tự ghi khắp nơi" sang kiến trúc hướng sự kiện (event-driven), lấy database làm nguồn sự thật duy nhất:

• Tầng ứng dụng chỉ ghi vào database, không phải tự lo đồng bộ.
• WAL của database trở thành một kênh phát thay đổi gần real-time.
• Cache, search index và các microservice sớm muộn đều nhận đủ thay đổi theo đúng thứ tự, kể cả khi có service tạm sập.

Đổi lại, bạn phải vận hành thêm Kafka Connect, canh replication slot, và chấp nhận dữ liệu ở phía nhận trễ một chút. Hệ thống nhỏ chỉ có một cache thì TTL ngắn cộng xóa cache sau commit (@TransactionalEventListener(phase = AFTER_COMMIT)) nhiều khi đã đủ.

Chỉ là những ghi chép cá nhân với hy vọng mang lại chút giá trị. Nếu thấy hữu ích, đừng ngại chia sẻ cho bạn bè & đồng nghiệp nhé!

Happy coding 😎 👍🏻 🚀 🔥.