How to Build a Real‑Time CDC Service with Tencent Cloud DTS: From MySQL Binlog to Business Events

CDC Basics

MySQL binlog records every INSERT, UPDATE, and DELETE as chronological events. Enabling binlog is a prerequisite for MySQL replication, so it is usually on in production. A CDC service pretends to be a MySQL replica, reads the binlog, parses change events into structured data, and pushes them downstream.

Compared with polling, CDC offers millisecond‑level latency, negligible database load (only reads binlog), full change history (including before‑and‑after values), but requires adding a CDC component and learning its operation.

Overall Architecture

The data flow consists of four steps:

MySQL generates binlog, captured by Tencent Cloud DTS.

DTS SDK’s ClusterListener receives change messages.

Business code filters fields and separates old and new values.

Based on the change content, messages are pushed to different RabbitMQ queues for downstream consumption.

The FILTER stage is where the business code discards irrelevant fields and extracts the values needed for downstream logic.

DTS Subscription Integration

After creating a subscription channel in DTS, a channel ID (dtsId) is assigned. The SDK connects using this ID and starts receiving binlog events.

SubscribeContext context = new SubscribeContext();
context.setSecretId(properties.getProperty("secretId"));
context.setSecretKey(properties.getProperty("secretKey"));
context.setRegion(properties.getProperty("region"));
context.setServiceIp(properties.getProperty("ip"));
context.setServicePort(Integer.parseInt(properties.getProperty("port")));

DefaultSubscribeClient client = new DefaultSubscribeClient(context);
client.addClusterListener(getListener());
client.askForGUID(properties.getProperty("dtsId"));
client.start();

The service runs as a standard Spring Boot microservice, providing health checks, graceful shutdown, and configuration via a central config center.

Field‑Level Filtering

DTS delivers all columns of a changed row, but downstream only needs a subset (e.g., price, stock, on‑sale status). The list of fields to retain is stored in Nacos so it can be updated without redeploying the service.

@ConfigurationProperties(prefix = "dts.subscribe")
@RefreshScope
public class DtsSubscribeProperties {
    private List<String> remainFields;
    // getter/setter
}

Example Nacos entry:

dts.subscribe.remainFields=skuPrice,onsale,skuStock,uId,skuId,orderId,status,itemInsale

Remember to add @RefreshScope so changes in Nacos take effect immediately.

At runtime the configuration is bound to a HashSet for fast lookup:

Set<String> remainSet = new HashSet<>(dtsSubscribeProperties.getRemainFields());

During field iteration, only fields present in remainSet are kept:

if (!remainSet.contains(propertyName)) {
    continue;
}

Comparing Old and New Values for UPDATE

For UPDATE events, the SDK returns each changed column twice: first the old value, then the new value, alternating in the FieldList. A simple counter modulo 2 can separate them, but this is opaque.

if (changeRecord.getType().equals("UPDATE")) {
    i = i % 2;
    if (i > 0) {
        newHashMap.put(propertyName, value);
    } else {
        oldHashMap.put(propertyName, value);
    }
    i++;
}

A clearer approach iterates with a step of 2, explicitly pairing old and new fields:

List<DataMessage.Record.Field> fields = m.getRecord().getFieldList();
for (int idx = 0; idx < fields.size(); idx += 2) {
    Field oldField = fields.get(idx);
    Field newField = fields.get(idx + 1);
    String propertyName = fieldToProperty(oldField.getFieldname());
    if (!remainSet.contains(propertyName)) {
        continue;
    }
    oldHashMap.put(propertyName, oldField.getValue());
    newHashMap.put(propertyName, newField.getValue());
}

This makes the intent obvious and guards against SDK changes that might reorder fields.

From Data Change to Business Event: Inventory Alert

With old and new values available, business logic can detect meaningful transitions. The example triggers an alert when stock drops from ≥30 to <30:

Integer newSkuStock = 0;
Integer oldSkuStock = 0;
if (newHashMap.get("skuStock") != null && oldHashMap.get("skuStock") != null) {
    newSkuStock = Integer.parseInt((String) newHashMap.get("skuStock"));
    oldSkuStock = Integer.parseInt((String) oldHashMap.get("skuStock"));
}
if (newSkuStock < 30 && oldSkuStock >= 30) {
    changeRecord.setChangeType(1); // inventory alert
} else {
    changeRecord.setChangeType(0); // normal change
}

This condition ensures the alert fires only once at the moment the threshold is crossed, avoiding repeated notifications while stock remains low.

Pushing Changes to RabbitMQ

After filtering and classification, the change is sent to RabbitMQ. Directly discarding failed pushes can cause data loss, especially for critical alerts. To avoid this, the service first writes the change into a local message table, then acknowledges the DTS message.

CREATE TABLE cdc_change_record (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    message_id VARCHAR(64) NOT NULL,
    table_name VARCHAR(64) NOT NULL,
    change_type VARCHAR(16) NOT NULL,
    old_data JSON,
    new_data JSON,
    business_type TINYINT DEFAULT 0,
    status TINYINT DEFAULT 0,
    retry_count INT DEFAULT 0,
    create_time DATETIME DEFAULT CURRENT_TIMESTAMP,
    update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    UNIQUE KEY uk_message_id (message_id)
);

The status field tracks processing state (0 = pending, 1 = sent, 2 = failed). A background task retries records with status 2.

executor.submit(() -> {
    try {
        if (record.getBusinessType() == 1) {
            rabbitTemplate.convertAndSend(stockLess30Queue, json);
        } else {
            rabbitTemplate.convertAndSend(dataChangeQueue, json);
        }
        record.setStatus(1);
        changeRecordMapper.updateById(record);
    } catch (Exception e) {
        record.setStatus(2);
        record.setRetryCount(record.getRetryCount() + 1);
        changeRecordMapper.updateById(record);
    }
});

Two queues are used:

stock_less30_notify_queue – carries inventory‑alert events to the operations notification service.

queue-datasubscribe – carries generic data‑change events for search index updates, cache refreshes, etc.

Downstream consumers must be idempotent because both DTS and RabbitMQ may redeliver messages.

Conclusion

CDC shifts data synchronization from application‑level push to database‑level capture, guaranteeing that any change—regardless of how it was made—is observed downstream. In the presented e‑commerce project, key design choices include field filtering to reduce noise, explicit old‑new value comparison for precise business rules, and queue‑based decoupling to isolate the CDC service from downstream failures. The same principles apply when using other CDC tools such as Canal or Debezium.