1. 数据同步架构概述MySQL - Canal - Kafka - ES这套数据同步方案本质上构建了一个高可靠、低延迟的数据库变更捕获与分发系统。我在多个电商和物联网项目中实际应用过这套架构它能完美解决传统ETL工具无法满足的实时性需求。核心工作原理分为三层数据采集层Canal伪装成MySQL从库通过binlog机制获取增量数据变更消息缓冲层Kafka作为高吞吐消息队列解耦生产者和消费者数据消费层通过Kafka消费者将数据写入Elasticsearch这种架构的优势在于实时性从数据库变更到ES索引更新延迟可控制在秒级可靠性Kafka的持久化机制确保数据不丢失扩展性各组件均可水平扩展应对数据量增长重要提示生产环境部署前务必评估MySQL的binlog格式必须为ROW模式这是Canal正常工作的前提条件。2. 环境准备与配置2.1 MySQL配置要点首先需要在MySQL服务器启用binlog并配置复制权限# 查看当前binlog格式 SHOW VARIABLES LIKE binlog_format; # 修改my.cnf配置 [mysqld] log-binmysql-bin binlog-formatROW server_id1 binlog_row_imageFULL expire_logs_days3 # 创建Canal专用账号 CREATE USER canal% IDENTIFIED BY Canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;2.2 Canal部署详解Canal 1.1.5版本的部署流程# 下载解压 wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz mkdir -p /opt/canal tar -zxvf canal.deployer-1.1.5.tar.gz -C /opt/canal # 关键配置修改 vi conf/example/instance.properties # 数据库连接配置 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordCanal123 canal.instance.filter.regex.*\\..* # 监控所有库表 # Kafka相关配置 canal.mq.topiccanal_mysql canal.mq.partition02.3 Kafka集群搭建建议使用3节点Kafka集群保证高可用# docker-compose.yml示例 version: 3 services: zookeeper: image: zookeeper:3.6 ports: - 2181:2181 kafka1: image: confluentinc/cp-kafka:6.2.0 ports: - 9092:9092 environment: KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092 KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_BROKER_ID: 1 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 depends_on: - zookeeper # 类似配置kafka2、kafka3...3. 数据同步核心实现3.1 Canal到Kafka的数据流转Canal会将binlog事件转换为Protocol Buffer格式消息发送到Kafka。消息结构示例{ data: [{ id: 1, name: 测试商品, price: 99.00, update_time: 2023-07-20 12:00:00 }], database: mall, table: products, type: UPDATE, es: 1689840000000 }可以通过以下命令监控Kafka消息# 查看topic列表 kafka-topics.sh --list --bootstrap-server localhost:9092 # 消费canal_mysql topic消息 kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic canal_mysql --from-beginning3.2 ES数据写入策略推荐使用Bulk API进行批量写入以下是用Logstash实现的配置示例input { kafka { bootstrap_servers kafka:9092 topics [canal_mysql] codec json } } filter { # 根据操作类型路由 if [type] INSERT or [type] UPDATE { mutate { add_field { [metadata][action] index } } } if [type] DELETE { mutate { add_field { [metadata][action] delete } } } } output { elasticsearch { hosts [http://elasticsearch:9200] index %{[database]}_%{[table]} document_id %{[data][0][id]} action %{[metadata][action]} } }4. 性能优化实践4.1 关键参数调优Canal参数调整# conf/canal.properties canal.instance.memory.buffer.size 32MB # 内存缓冲区大小 canal.instance.memory.buffer.memunit 1024KB canal.instance.transaction.size 1024 # 事务批处理大小Kafka生产者参数kafka.batch.size 16384 kafka.linger.ms 50 kafka.compression.type snappyES写入优化PUT _template/canal_template { index_patterns: [*_*], settings: { refresh_interval: 30s, number_of_shards: 3, number_of_replicas: 1, index.translog.durability: async } }4.2 监控方案设计建议部署以下监控指标Canal延迟监控通过canal.instance.detecting.interval检测主从延迟Kafka堆积监控使用kafka-consumer-groups.sh查看消费延迟ES写入性能监控bulk操作的响应时间和错误率Prometheus配置示例- job_name: canal static_configs: - targets: [canal:11112] - job_name: es metrics_path: /_prometheus/metrics static_configs: - targets: [elasticsearch:9200]5. 常见问题排查5.1 数据不一致问题现象ES中数据与MySQL不一致排查步骤检查Canal的meta.log确认位点是否正常推进使用kafka-console-consumer验证Kafka消息内容查看Logstash日志是否有处理错误解决方案# 重置Canal位点 curl http://canal:8081/destination/example/1001/reset -d {timestamp:1672502400000} # 重建ES索引 POST _reindex { source: {index: mall_products}, dest: {index: mall_products_new} }5.2 性能瓶颈分析CPU过高Canal检查parseThread线程是否阻塞Kafka监控NetworkProcessor线程使用率ES关注merge线程状态内存不足# 调整JVM参数 CANAL_OPTS-Xms4g -Xmx4g -Xmn1g KAFKA_HEAP_OPTS-Xmx8g -Xms8g ES_JAVA_OPTS-Xms16g -Xmx16g6. 生产环境部署建议经过多个项目的实践验证我总结出以下部署规范资源隔离原则MySQL主从分离Canal连接专用从库Kafka集群独立部署不与业务系统混用ES集群数据节点与主节点分离高可用配置# canal部署双实例 services: canal_primary: image: canal/canal-server:v1.1.5 ports: [11111:11111] canal_secondary: image: canal/canal-server:v1.1.5 ports: [11112:11111]灾备方案定期备份Kafka的__consumer_offsets主题为ES建立跨机房副本实现Canal配置的版本化管理这套架构在日处理10亿级数据量的电商系统中表现稳定从数据库变更到ES可搜索的平均延迟控制在3秒内。最关键的是要确保Kafka的磁盘IO性能和ES的JVM配置合理这是保证整体吞吐量的关键因素。
1. 数据同步架构概述MySQL - Canal - Kafka - ES这套数据同步方案本质上构建了一个高可靠、低延迟的数据库变更捕获与分发系统。我在多个电商和物联网项目中实际应用过这套架构它能完美解决传统ETL工具无法满足的实时性需求。核心工作原理分为三层数据采集层Canal伪装成MySQL从库通过binlog机制获取增量数据变更消息缓冲层Kafka作为高吞吐消息队列解耦生产者和消费者数据消费层通过Kafka消费者将数据写入Elasticsearch这种架构的优势在于实时性从数据库变更到ES索引更新延迟可控制在秒级可靠性Kafka的持久化机制确保数据不丢失扩展性各组件均可水平扩展应对数据量增长重要提示生产环境部署前务必评估MySQL的binlog格式必须为ROW模式这是Canal正常工作的前提条件。2. 环境准备与配置2.1 MySQL配置要点首先需要在MySQL服务器启用binlog并配置复制权限# 查看当前binlog格式 SHOW VARIABLES LIKE binlog_format; # 修改my.cnf配置 [mysqld] log-binmysql-bin binlog-formatROW server_id1 binlog_row_imageFULL expire_logs_days3 # 创建Canal专用账号 CREATE USER canal% IDENTIFIED BY Canal123; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal%; FLUSH PRIVILEGES;2.2 Canal部署详解Canal 1.1.5版本的部署流程# 下载解压 wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz mkdir -p /opt/canal tar -zxvf canal.deployer-1.1.5.tar.gz -C /opt/canal # 关键配置修改 vi conf/example/instance.properties # 数据库连接配置 canal.instance.master.address127.0.0.1:3306 canal.instance.dbUsernamecanal canal.instance.dbPasswordCanal123 canal.instance.filter.regex.*\\..* # 监控所有库表 # Kafka相关配置 canal.mq.topiccanal_mysql canal.mq.partition02.3 Kafka集群搭建建议使用3节点Kafka集群保证高可用# docker-compose.yml示例 version: 3 services: zookeeper: image: zookeeper:3.6 ports: - 2181:2181 kafka1: image: confluentinc/cp-kafka:6.2.0 ports: - 9092:9092 environment: KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:9092 KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092 KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 KAFKA_BROKER_ID: 1 KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 depends_on: - zookeeper # 类似配置kafka2、kafka3...3. 数据同步核心实现3.1 Canal到Kafka的数据流转Canal会将binlog事件转换为Protocol Buffer格式消息发送到Kafka。消息结构示例{ data: [{ id: 1, name: 测试商品, price: 99.00, update_time: 2023-07-20 12:00:00 }], database: mall, table: products, type: UPDATE, es: 1689840000000 }可以通过以下命令监控Kafka消息# 查看topic列表 kafka-topics.sh --list --bootstrap-server localhost:9092 # 消费canal_mysql topic消息 kafka-console-consumer.sh --bootstrap-server localhost:9092 \ --topic canal_mysql --from-beginning3.2 ES数据写入策略推荐使用Bulk API进行批量写入以下是用Logstash实现的配置示例input { kafka { bootstrap_servers kafka:9092 topics [canal_mysql] codec json } } filter { # 根据操作类型路由 if [type] INSERT or [type] UPDATE { mutate { add_field { [metadata][action] index } } } if [type] DELETE { mutate { add_field { [metadata][action] delete } } } } output { elasticsearch { hosts [http://elasticsearch:9200] index %{[database]}_%{[table]} document_id %{[data][0][id]} action %{[metadata][action]} } }4. 性能优化实践4.1 关键参数调优Canal参数调整# conf/canal.properties canal.instance.memory.buffer.size 32MB # 内存缓冲区大小 canal.instance.memory.buffer.memunit 1024KB canal.instance.transaction.size 1024 # 事务批处理大小Kafka生产者参数kafka.batch.size 16384 kafka.linger.ms 50 kafka.compression.type snappyES写入优化PUT _template/canal_template { index_patterns: [*_*], settings: { refresh_interval: 30s, number_of_shards: 3, number_of_replicas: 1, index.translog.durability: async } }4.2 监控方案设计建议部署以下监控指标Canal延迟监控通过canal.instance.detecting.interval检测主从延迟Kafka堆积监控使用kafka-consumer-groups.sh查看消费延迟ES写入性能监控bulk操作的响应时间和错误率Prometheus配置示例- job_name: canal static_configs: - targets: [canal:11112] - job_name: es metrics_path: /_prometheus/metrics static_configs: - targets: [elasticsearch:9200]5. 常见问题排查5.1 数据不一致问题现象ES中数据与MySQL不一致排查步骤检查Canal的meta.log确认位点是否正常推进使用kafka-console-consumer验证Kafka消息内容查看Logstash日志是否有处理错误解决方案# 重置Canal位点 curl http://canal:8081/destination/example/1001/reset -d {timestamp:1672502400000} # 重建ES索引 POST _reindex { source: {index: mall_products}, dest: {index: mall_products_new} }5.2 性能瓶颈分析CPU过高Canal检查parseThread线程是否阻塞Kafka监控NetworkProcessor线程使用率ES关注merge线程状态内存不足# 调整JVM参数 CANAL_OPTS-Xms4g -Xmx4g -Xmn1g KAFKA_HEAP_OPTS-Xmx8g -Xms8g ES_JAVA_OPTS-Xms16g -Xmx16g6. 生产环境部署建议经过多个项目的实践验证我总结出以下部署规范资源隔离原则MySQL主从分离Canal连接专用从库Kafka集群独立部署不与业务系统混用ES集群数据节点与主节点分离高可用配置# canal部署双实例 services: canal_primary: image: canal/canal-server:v1.1.5 ports: [11111:11111] canal_secondary: image: canal/canal-server:v1.1.5 ports: [11112:11111]灾备方案定期备份Kafka的__consumer_offsets主题为ES建立跨机房副本实现Canal配置的版本化管理这套架构在日处理10亿级数据量的电商系统中表现稳定从数据库变更到ES可搜索的平均延迟控制在3秒内。最关键的是要确保Kafka的磁盘IO性能和ES的JVM配置合理这是保证整体吞吐量的关键因素。
