MySQL高级讲座篇之：探讨MySQL和`Apache Kafka`的`CDC`（变更数据捕获）实践：从`binlog`到消息流。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

咳咳，各位观众老爷们，大家好！我是今天的讲师，江湖人称“代码搬运工”，今天咱们就来聊聊MySQL和Apache Kafka的“爱情故事”，哦不，是CDC（变更数据捕获）实践。

开场白：数据江湖的那些事儿

话说在数据江湖里，MySQL就像一位兢兢业业的老掌柜，每天忙着记录着店铺的流水账。而Kafka呢，则像一位消息灵通的江湖百晓生，能把这些流水账快速传播给各个需要的人。

那么问题来了，老掌柜的流水账怎么才能实时同步给百晓生呢？这就是CDC要解决的问题。简单来说，CDC就像一个“情报员”，潜伏在MySQL身边，时刻监听着数据的变化，一旦发生变化，立马通知Kafka。

第一回合：什么是CDC？为何需要它？

CDC，全称Change Data Capture，即变更数据捕获。 顾名思义，它就是用来捕获数据库数据变更的技术。

为什么要用CDC呢？原因很简单，传统的同步方式太慢了！

假设你需要把MySQL的数据同步到Elasticsearch做搜索，或者同步到Hadoop做数据分析，如果采用定期全量同步的方式，数据延迟会非常高，实时性差。

而CDC可以做到近乎实时的同步，大大提升了数据处理的效率。

总结一下，CDC的优势：

实时性高： 近乎实时地捕获数据变更。
减少资源消耗： 只同步变更的数据，避免全量同步的资源消耗。
解耦： 将数据源和下游系统解耦，降低系统之间的依赖性。

第二回合：MySQL的binlog：CDC的命脉

要实现MySQL的CDC，最关键的就是binlog (binary log)。

什么是binlog？

binlog是MySQL的二进制日志，它记录了所有对数据库的变更操作，包括INSERT、UPDATE、DELETE等。

binlog有什么用？

数据恢复： 可以用binlog进行数据恢复，例如在误操作后恢复到某个时间点。
主从复制： MySQL的主从复制就是基于binlog实现的。
CDC： CDC系统正是通过解析binlog来捕获数据变更的。

如何开启binlog？

在MySQL的配置文件（通常是my.cnf或my.ini）中，添加以下配置：

[mysqld]
log-bin=mysql-bin  # 开启binlog，并指定binlog的文件名
binlog_format=ROW  # 设置binlog的格式为ROW
server-id=1        # 设置服务器ID，在主从复制中需要唯一

注意：

binlog_format有三种格式：STATEMENT、ROW、MIXED。CDC通常使用ROW格式，因为它记录了每一行数据的变化，更加可靠。
重启MySQL服务后，binlog才会生效。

binlog格式的区别：

格式	描述	优点	缺点
STATEMENT	记录SQL语句，例如`UPDATE table SET column = value WHERE id = 1;`	日志文件较小，占用磁盘空间少。	在某些情况下，SQL语句的执行结果可能不确定，例如使用了`NOW()`函数。
ROW	记录每一行数据的变化，例如`UPDATE table SET column = value WHERE id = 1;` 会记录`id = 1`的行在更新前后的值。	数据可靠性高，不会出现数据不一致的情况。	日志文件较大，占用磁盘空间多。
MIXED	混合模式，MySQL会根据SQL语句的类型自动选择使用`STATEMENT`或`ROW`格式。	兼顾了`STATEMENT`和`ROW`的优点，在大多数情况下都能工作良好。	在某些特殊情况下，可能会出现数据不一致的情况，需要仔细测试。

第三回合：CDC工具的选择：百花齐放

既然知道了binlog的重要性，接下来就需要选择合适的CDC工具来解析binlog，并把数据发送到Kafka。

市面上有很多CDC工具，例如：

Debezium: 一个开源的分布式CDC平台，支持多种数据库。
Canal: 阿里巴巴开源的CDC工具，专门针对MySQL。
Maxwell: 一个简单的CDC工具，使用Java编写。
Flink CDC: Apache Flink提供的CDC Connector，可以直接读取binlog。

Debezium：

Debezium是一个非常强大的CDC平台，它支持多种数据库，包括MySQL、PostgreSQL、MongoDB等。Debezium的架构比较复杂，但是功能非常强大，可以满足各种复杂的CDC需求。

优点：

支持多种数据库。
功能强大，支持各种复杂的CDC需求。
社区活跃，文档完善。

缺点：

配置复杂，学习曲线陡峭。
资源消耗较大。

Canal：

Canal是阿里巴巴开源的CDC工具，它专门针对MySQL，性能非常优秀。Canal的架构比较简单，易于部署和维护。

优点：

性能优秀，专门针对MySQL优化。
配置简单，易于部署和维护。
支持多种消息队列，包括Kafka、RocketMQ等。

缺点：

只支持MySQL。
功能相对Debezium较少。

Maxwell：

Maxwell是一个简单的CDC工具，使用Java编写，易于使用和扩展。Maxwell的架构非常简单，适合小型项目。

优点：

易于使用和扩展。
资源消耗较小。

缺点：

功能相对Debezium和Canal较少。
性能不如Canal。

Flink CDC：

Flink CDC是Apache Flink提供的CDC Connector，可以直接读取binlog，并进行实时处理。Flink CDC的优势在于可以与Flink无缝集成，方便进行复杂的数据处理。

优点：

与Flink无缝集成。
可以进行复杂的数据处理。

缺点：

需要熟悉Flink。

选择哪个工具？

选择哪个工具取决于你的具体需求。

如果需要支持多种数据库，可以选择Debezium。
如果只使用MySQL，并且需要高性能，可以选择Canal。
如果需要与Flink集成，可以选择Flink CDC。
如果项目规模较小，可以选择Maxwell。

第四回合：实战演练：以Canal为例

咱们以Canal为例，演示一下如何实现MySQL的CDC。

1. 安装Canal Server：

从Canal的官网下载Canal Server的安装包，解压到服务器上。

2. 配置Canal Server：

修改conf/canal.properties文件，配置MySQL的连接信息：

canal.instance.master.address=127.0.0.1:3306
canal.instance.master.username=canal
canal.instance.master.password=canal
canal.instance.master.defaultDatabaseName=testdb

注意：

需要创建一个专门用于Canal的MySQL用户，并授予REPLICATION SLAVE权限。

CREATE USER 'canal'@'%' IDENTIFIED BY 'canal';
GRANT REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
FLUSH PRIVILEGES;

3. 配置Canal Instance：

在conf/example目录下，创建一个名为example.properties的文件，配置Canal Instance的信息：

canal.instance.mysql.slaveId=1234
canal.instance.filter.regex=testdb\..*  # 过滤testdb数据库下的所有表
canal.instance.canal.id=1001

4. 启动Canal Server：

./bin/startup.sh

5. 创建Kafka Topic：

kafka-topics.sh --create --topic canal_topic --partitions 1 --replication-factor 1 --zookeeper localhost:2181

6. 编写Canal Client：

编写一个Canal Client，连接Canal Server，并把数据发送到Kafka。

以下是一个简单的Java示例：

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Properties;

public class CanalClientExample {

    public static void main(String[] args) {
        // 配置Canal连接信息
        String address = AddressUtils.getHostAddress();
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(address, 11111), "example", "canal", "canal");

        // 配置Kafka Producer
        Properties kafkaProps = new Properties();
        kafkaProps.put("bootstrap.servers", "localhost:9092");
        kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(kafkaProps);

        try {
            connector.connect();
            connector.subscribe(".*\..*"); // 订阅所有数据库的所有表
            connector.rollback();

            while (true) {
                Message message = connector.getWithoutAck(100); // 获取指定数量的数据
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        // ignore
                    }
                } else {
                    printEntry(message.getEntries(), producer);
                }

                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }

        } finally {
            connector.disconnect();
            producer.close();
        }
    }

    private static void printEntry(List<Entry> entrys, KafkaProducer<String, String> producer) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChange = null;
            try {
                rowChange = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChange.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList(), producer, entry);
                } else if (eventType == EventType.INSERT) {
                    printColumn(rowData.getAfterColumnsList(), producer, entry);
                } else {
                    System.out.println("-------> before");
                    printColumn(rowData.getBeforeColumnsList(), producer, entry);
                    System.out.println("-------> after");
                    printColumn(rowData.getAfterColumnsList(), producer, entry);
                }
            }
        }
    }

    private static void printColumn(List<Column> columns, KafkaProducer<String, String> producer, Entry entry) {
        StringBuilder sb = new StringBuilder();
        for (Column column : columns) {
            sb.append(column.getName()).append(":").append(column.getValue()).append("   ");
        }
        System.out.println(sb.toString());
        String message = String.format("binlog[%s:%s] , name[%s,%s] , data: %s",
                entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                sb.toString());
        ProducerRecord<String, String> record = new ProducerRecord<>("canal_topic", message);
        producer.send(record);
    }
}

注意：

需要引入Canal Client的依赖，以及Kafka Client的依赖。
需要根据实际情况修改代码中的连接信息和Topic名称。

7. 测试：

在MySQL中执行一些INSERT、UPDATE、DELETE操作，然后查看Kafka Topic，应该可以看到Canal Client发送的数据。

第五回合：高级技巧：容错、监控、优化

CDC不仅仅是把数据同步到Kafka就完事了，还需要考虑容错、监控和优化。

1. 容错：

Canal HA： Canal Server可以配置HA（High Availability），保证Canal Server的高可用性。
消息重试： Kafka Producer可以配置重试机制，保证消息的可靠性。
死信队列： 可以配置死信队列，用于存储处理失败的消息，方便后续处理。

2. 监控：

Canal监控： 可以使用Canal提供的监控接口，监控Canal Server的运行状态。
Kafka监控： 可以使用Kafka Manager或Kafka Eagle等工具，监控Kafka的运行状态。
自定义监控： 可以自定义监控指标，例如数据延迟、数据量等。

3. 优化：

binlog格式： 选择合适的binlog格式，通常使用ROW格式。
网络带宽： 优化网络带宽，避免网络瓶颈。
Kafka配置： 调整Kafka的配置，例如batch.size、linger.ms等，提高吞吐量。
Canal Instance配置: 调整Canal Instance的配置，例如canal.instance.get.batchSize等，提高读取效率。

第六回合：江湖总结：数据流转的未来

通过今天的讲座，我们了解了MySQL和Kafka的CDC实践，从binlog到消息流，实现了数据的实时同步。

CDC技术在现代数据架构中扮演着越来越重要的角色，它可以应用于各种场景，例如：

实时数据仓库： 构建实时数据仓库，提供实时的数据分析能力。
实时搜索： 将数据同步到Elasticsearch，提供实时的搜索功能。
实时推荐： 基于实时数据，进行个性化推荐。
微服务架构： 在微服务之间进行数据同步。

希望今天的讲座能对大家有所帮助，谢谢大家！

最后，给大家留个思考题：

在实际应用中，如何保证数据的一致性？例如，在UPDATE操作中，如果Canal Client在读取到before和after数据之间宕机了，会导致数据不一致。该如何解决？

欢迎大家在评论区留言讨论！

开场白：数据江湖的那些事儿

第一回合：什么是CDC？为何需要它？

第二回合：MySQL的binlog：CDC的命脉

第三回合：CDC工具的选择：百花齐放

第四回合：实战演练：以Canal为例

第五回合：高级技巧：容错、监控、优化

第六回合：江湖总结：数据流转的未来

发表回复 取消回复

发表回复取消回复