1. 数据同步架构概述MySQL到Elasticsearch的数据同步是企业级应用中常见的需求特别是在需要实时搜索和分析的场景下。这套基于CanalKafka的同步方案完美解决了传统ETL工具无法满足的实时性要求。整个流程的核心原理是Canal伪装成MySQL的从库通过解析binlog获取数据变更然后将变更事件发布到Kafka消息队列最后由消费者将数据写入Elasticsearch。这种架构有三大优势对源库零侵入不需要修改业务代码毫秒级延迟满足绝大多数实时场景高可靠性Kafka的持久化机制确保数据不丢失2. 环境准备与配置2.1 MySQL配置要求要让Canal正常工作MySQL必须开启binlog并设置为ROW模式[mysqld] log-binmysql-bin binlog-formatROW server_id1注意确保Canal使用的数据库账号具有REPLICATION SLAVE和REPLICATION CLIENT权限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 tar -zxvf canal.deployer-1.1.5.tar.gz -C /opt/canal关键配置文件说明conf/canal.properties全局配置conf/example/instance.properties实例配置3. Kafka集成配置3.1 Canal到Kafka的配置在canal.properties中配置Kafka相关参数canal.serverMode kafka kafka.bootstrap.servers127.0.0.1:9092 kafka.acksall kafka.compression.typenone kafka.batch.size163843.2 消息分区策略通过partitionHash可以实现按主键自动分区canal.mq.partitionsNum3 canal.mq.partitionHashtest.user:id4. Elasticsearch写入实现4.1 消费Kafka消息使用Kafka Consumer API消费消息的代码片段Properties props new Properties(); props.put(bootstrap.servers, localhost:9092); props.put(group.id, es_loader); KafkaConsumerString, String consumer new KafkaConsumer(props); consumer.subscribe(Collections.singletonList(canal_topic));4.2 ES文档操作使用RestHighLevelClient写入ES的示例IndexRequest request new IndexRequest(user_index) .id(user.getId()) .source(JSON.toJSONString(user), XContentType.JSON); client.index(request, RequestOptions.DEFAULT);5. 监控与问题排查5.1 关键监控指标需要监控的核心指标包括Canal解析延迟Kafka堆积量ES写入QPS5.2 常见问题解决方案问题1Canal CPU占用高可能原因大事务解析 解决方案调整canal.instance.transaction.size参数问题2ES写入性能差优化建议开启批量写入适当增加refresh_interval6. 生产环境优化建议Kafka配置优化合理设置retries和linger.ms根据数据量调整batch.sizeES索引设计预创建索引模板设置合理的分片数容灾方案部署Canal集群配置Kafka镜像集群这套方案我们已经在大规模生产环境验证单集群每天可稳定同步数十亿条数据变更平均延迟控制在500ms以内。