本文目录导读：
MySQL与Elasticsearch数据同步实战案例：从双写到CDC的全链路解析
目录导读
- 为什么要做数据同步？ – 业务场景与痛点
- 核心同步方案对比 – 双写、定时任务、CDC、日志订阅
- 基于Binlog+Canal的准实时同步
- Logstash JDBC+定时增量同步
- 常见问题与解决策略 – 数据不一致、延迟、全量重刷
- Q&A高频问题解答
- 总结与最佳实践建议
为什么要做数据同步？
在实际业务中，MySQL作为OLTP（在线事务处理）数据库擅长事务与关联查询，而Elasticsearch作为全文检索引擎能实现毫秒级模糊搜索、聚合分析与高并发读取，当业务需要同时支持“写库的完整性”和“读库的搜索性能”时,就需要将MySQL数据实时或准实时同步到Elasticsearch。
典型场景：电商商品搜索、日志分析平台、用户标签检索、论坛帖子全文搜索。
痛点：直接通过MySQL全表like查询性能极差，且无法支撑复杂分词；而ES本身不是事务型数据库,无法替代MySQL的ACID能力。
核心同步方案对比
| 方案 | 原理 | 优点 | 缺点 |
|---|---|---|---|
| 应用双写 | 业务代码里同时写入MySQL和ES | 简单直接 | 耦合高，容易遗漏，一致性难保证 |
| 定时任务 | 用Cron定时扫描MySQL增量更新ES | 开发成本低 | 延迟分钟级，无法做到秒级 |
| CDC (Change Data Capture) | 捕获Binlog变更事件实时推送到ES | 准实时，无侵入 | 需要运维Binlog与中间件（如Canal、Debezium） |
| Logstash JDBC Input | 定时执行SQL查询增量数据写入ES | 配置灵活，社区支持好 | 仍存在分钟级延迟，大表全量重刷较慢 |
选择建议：对实时性要求高的业务（如订单状态更新）首选CDC；对实时性要求不高（如每日报表）可用Logstash定时同步。
案例一：基于Binlog+Canal的准实时同步
架构说明
MySQL (Binlog) → Canal (客户端) → Kafka (可选) → Logstash/自定义程序 → Elasticsearch- Canal：阿里开源组件,伪装为MySQL从库读取Binlog。
- Kafka：可选缓冲层，用于削峰填谷,避免ES写入压力。
- ES写入优化：使用Bulk API（批量写入），设置refresh_interval=-1（手动刷新）。
核心步骤
- 开启MySQL Binlog：设置
binlog_format=ROW，binlog_row_image=FULL。 - 部署Canal：配置
destination指向需要同步的表。 - 写入ES：Canal客户端解析事件（Insert/Update/Delete），根据主键构造ES Document，然后调用Bulk API写入。
- 全量补充：首次同步时，通过JDBC读取全量数据,之后监控Binlog增量。
注意点
- 表结构变更（DDL）需要同步,否则ES映射会失效。
- 记录死锁或网络抖动时,通过序列号去重或幂等写入。
案例二：Logstash JDBC+定时增量同步
适用场景
业务允许分钟级延迟，数据量≤500万，无需实时感知，常见于BI报表、非核心搜索页面。
配置示例
input {
jdbc {
jdbc_driver_library => "/path/mysql-connector-java.jar"
jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb"
jdbc_user => "user"
jdbc_password => "pass"
statement => "SELECT * FROM articles WHERE updated_at > :sql_last_value"
schedule => "*/30 * * * * *" # 每30秒执行
use_column_value => true
tracking_column => "updated_at"
last_run_metadata_path => "/path/last_time"
}
}
output {
elasticsearch {
hosts => ["http://localhost:9200"]
index => "articles_index"
document_id => "%{id}"
action => "upsert"
}
}
核心逻辑
- 用
updated_at作为增量标记,Logstash每次记录上一次运行时间戳。 - 若记录有物理删除则需额外处理（如逻辑删除标记
is_deleted=1）。
常见问题与解决策略
问题1：数据不一致（MySQL有记录，但ES查不到）
- 原因：同步代码遗漏了某些更新事件,或者Binlog被过早清理。
- 解决：定期执行全量比对任务，通过MySQL的
checksum或count(*)验证；对于发现缺失的ID,重新推送。
问题2：写入ES时出现映射冲突
- 原因：MySQL字段类型变化（如String变Integer）导致ES mapping拒绝。
- 解决：使用dynamic mapping（谨慎）或预先在ES设置
"strict"模式；字段类型变更后重建索引。
问题3：延迟飙升
- 原因：ES写入QPS超过阈值,或者Canal消费能力不足。
- 解决：增加ES分片数；使用Kafka缓冲；优化写入策略（如将多个事件合并为一个Bulk请求）。
Q&A高频问题解答
Q1：MySQL主从同步与ES同步冲突吗？
A：不冲突，MySQL主从同步是数据库内部机制，ES同步是独立业务逻辑，你可以从MySQL主库或从库读取Binlog,但从库读取能减轻主库压力。
Q2：数据同步中如何保证幂等？
A：使用ES的_id字段设置为MySQL主键值，同一条记录反复发送时，ES通过_id覆盖更新,不会产生重复d文档。
Q3：ES中需要存储MySQL所有的字段吗？
A：不必要，通常只存储需要被搜索、排序、聚合的字段，例如文本字段需分词，数值字段需做范围过滤,其他冗余字段可省略。
Q4：如何处理MySQL中的软删除
A：推荐在MySQL表中增加deleted_at或is_deleted字段，同步时过滤deleted_at IS NOT NULL的记录，如果物理删除,则需在Binlog中捕获DELETE事件并及时从ES中删除对应文档。
Q5：全量同步时如何避免服务宕机？
A：使用滚动重建，先创建新索引（如articles_v2），同时保持双写，等同步完成后切换别名指向新索引,再删除旧索引。
总结与最佳实践建议
MySQL与Elasticsearch的数据同步不是一个“万能”配置，它需要根据业务实时性、数据量级、运维成本来权衡：
- 高频实时业务（如订单、库存）：采用Canal + Kafka + 自定义消费者，必须监控延迟告警。
- 低频分析业务（如用户行为统计）：Logstash JDBC定时同步足以,配置简单。
- 全量+增量双阶段：先通过JDBC全量导入,再无缝切换至Binlog监听。
- 监控不可少：监控ES写入延迟、Canal消费积压、MySQL Binlog过期时间。
最后提醒：无论哪种方案，均需在测试环境验证数据完整性和并发冲突,确保生产环境切换时不会影响现有业务。
如果公司域名是
example.com，部署监控仪表盘时请使用内网专有域名（如monitor.internal.example.com）,避免公网安全风险。
