我要评论在现代电商系统中，mysql 作为关系型数据库负责数据的持久化存储，而 elasticsearch 则作为搜索引擎提供高效的全文检索能力。保证两者之间的数据一致性是系统设计的关键挑战。本文将详细介绍主流的同步方案、实现方式及其优缺点。
一、同步方案对比
| 同步方案 | 实时性 | 实现复杂度 | 一致性保证 | 性能影响 | 适用场景 |
|---|---|---|---|---|---|
| 同步双写 | 极高 | 简单 | 强一致性 | 高 | 金融交易、核心订单 |
| 异步双写 | 较高 | 中等 | 最终一致性 | 中 | 电商商品、用户信息 |
| canal + mq | 高 | 较高 | 最终一致性 | 低 | 大规模数据同步 |
| logstash 定时 | 低 | 简单 | 弱一致性 | 低 | 报表、分析数据 |
| debezium | 高 | 高 | 最终一致性 | 低 | 复杂数据同步 |
二、详细实现方案
1. 同步双写
原理：在业务代码中同时写入 mysql 和 elasticsearch，确保两者数据同步更新。
实现代码：
@service
@transactional
public class productserviceimpl implements productservice {
@autowired
private productmapper productmapper;
@autowired
private elasticsearchclient esclient;
@override
public void createproduct(product product) {
// 1. 写入 mysql
productmapper.insert(product);
// 2. 同步写入 es
try {
productindex productindex = converttoindex(product);
indexrequest<productindex> request = indexrequest.of(i -> i
.index("product_index")
.id(product.getid().tostring())
.document(productindex)
);
esclient.index(request);
} catch (exception e) {
// 处理 es 写入失败的情况
log.error("es同步失败: {}", e.getmessage());
// 可以选择抛出异常回滚事务，或者记录失败日志后续补偿
throw new runtimeexception("数据同步失败", e);
}
}
@override
public void updateproduct(product product) {
// 类似 createproduct，同时更新 mysql 和 es
productmapper.updatebyid(product);
// es 更新逻辑...
}
private productindex converttoindex(product product) {
// 实体转换逻辑
productindex index = new productindex();
index.setid(product.getid());
index.settitle(product.gettitle());
index.setprice(product.getprice());
// 其他字段转换...
return index;
}
}
优缺点：
- 优点：实现简单，数据一致性强，实时性最高
- 缺点：
- 代码耦合度高，业务逻辑与数据同步混合
- es 写入延迟影响主业务性能
- 故障处理复杂，需考虑回滚机制
2. 异步双写
原理：通过消息队列解耦，业务代码只负责写 mysql，然后发送消息到 mq，由消费者异步更新 es。
实现代码：
// 生产者端
@service
public class productserviceimpl implements productservice {
@autowired
private productmapper productmapper;
@autowired
private rabbittemplate rabbittemplate;
@override
@transactional
public void createproduct(product product) {
// 1. 写入 mysql
productmapper.insert(product);
// 2. 发送消息到队列
productevent event = new productevent();
event.settype("create");
event.setproductid(product.getid());
rabbittemplate.convertandsend("product-event-exchange", "product.create", event);
}
}
// 消费者端
@component
public class productsyncconsumer {
@autowired
private productmapper productmapper;
@autowired
private elasticsearchclient esclient;
@rabbitlistener(queues = "product-sync-queue")
public void handleproductevent(productevent event) {
try {
product product = productmapper.selectbyid(event.getproductid());
if (product == null) {
// 删除 es 中的数据
deletefromes(event.getproductid());
return;
}
// 同步到 es
productindex index = converttoindex(product);
indexrequest<productindex> request = indexrequest.of(i -> i
.index("product_index")
.id(product.getid().tostring())
.document(index)
);
esclient.index(request);
log.info("产品 {} 同步到 es 成功", event.getproductid());
} catch (exception e) {
log.error("同步 es 失败: {}", e.getmessage());
// 可以根据需要进行重试或记录到死信队列
}
}
}
优缺点：
- 优点：
- 解耦业务与同步逻辑
- 消息队列提供削峰填谷能力
- es 故障不影响主业务流程
- 缺点：
- 存在短暂的数据不一致
- 增加了系统复杂度
- 需要处理消息丢失、重复消费等问题
3. canal + 消息队列 方案
原理：利用 canal 监听 mysql 的 binlog，解析数据变更并发送到消息队列，再由消费者同步到 es。
3.1 环境准备
mysql 配置：
# 开启 binlog log-bin=mysql-bin # 选择 row 模式 binlog-format=row # 服务器唯一id server-id=1 # 开启 binlog 实时更新 sync_binlog=1
canal server 配置：
# canal-server/conf/example/instance.properties canal.instance.master.address=127.0.0.1:3306 canal.instance.dbusername=canal canal.instance.dbpassword=canal canal.instance.connectioncharset=utf-8 canal.instance.filter.regex=.*\..*
canal adapter 配置：
# canal-adapter/conf/application.yml
server:
port: 8081
spring:
jackson:
date-format: yyyy-mm-dd hh:mm:ss
time-zone: gmt+8
default-property-inclusion: non_null
canal.conf:
mode: kafka
zookeeperhosts:
syncbatchsize: 1000
retries: 0
timeout:
accesskey:
secretkey:
consumerproperties:
kafka.bootstrap.servers: 127.0.0.1:9092
kafka.enable.auto.commit: false
kafka.auto.commit.interval.ms: 1000
kafka.auto.offset.reset: latest
kafka.request.timeout.ms: 40000
kafka.session.timeout.ms: 30000
kafka.isolation.level: read_committed
kafka.max.poll.records: 1000
srcdatasources:
defaultds:
url: jdbc:mysql://127.0.0.1:3306/shop?useunicode=true
username: root
password: root
canaladapters:
- instance: example # canal instance name or mq topic name
groups:
- groupid: g1
outeradapters:
- name: es
hosts: 127.0.0.1:9200
properties:
mode: rest
cluster.name: elasticsearch
表映射配置：
# canal-adapter/conf/es/mytest_user.yml
datasourcekey: defaultds
destination: example
groupid:
topic: example
database: shop
table: tb_product
esmapping:
_index: product_index
_type: _doc
_id: _id
upsert: true
sql: |
select
p.id as _id,
p.title,
p.sub_title as subtitle,
p.price,
p.sales,
c.name as categoryname
from tb_product p
left join tb_category c on p.cid1 = c.id
commitbatch: 3000
3.2 自定义消费者实现
如果需要更灵活的处理，可以自定义 kafka 消费者：
@component
public class productsyncconsumer {
@autowired
private elasticsearchclient esclient;
@kafkalistener(topics = "example")
public void processmessage(string message) {
try {
// 解析 canal 消息
canalmessage canalmsg = json.parseobject(message, canalmessage.class);
for (canaldata data : canalmsg.getdata()) {
// 根据操作类型处理
switch (canalmsg.geteventtype()) {
case insert:
case update:
synctoes(data);
break;
case delete:
deletefromes(data);
break;
}
}
} catch (exception e) {
log.error("处理 canal 消息失败: {}", e.getmessage());
}
}
private void synctoes(canaldata data) throws ioexception {
// 构建 es 文档
productindex index = new productindex();
index.setid(long.valueof(data.get("id").tostring()));
index.settitle(data.get("title").tostring());
// 其他字段映射...
// 写入 es
esclient.index(i -> i
.index("product_index")
.id(index.getid().tostring())
.document(index)
);
}
}
优缺点：
- 优点：
- 完全解耦，对业务代码零侵入
- 高性能，只同步变更数据
- 支持全量和增量同步
- 可靠性高，基于 binlog 保证不丢失
- 缺点：
- 部署复杂度高
- 配置相对复杂
- 对 mysql binlog 有依赖
三、数据一致性保障策略
1. 幂等性设计
确保重复同步不会导致数据异常：
// es 操作幂等性实现
public void syncproduct(long productid) {
// 使用文档id作为唯一标识
indexrequest<productindex> request = indexrequest.of(i -> i
.index("product_index")
.id(productid.tostring())
.document(buildproductindex(productid))
// 设置乐观锁版本控制
.versiontype(versiontype.external)
.version(getcurrentversion(productid))
);
try {
esclient.index(request);
} catch (versionconflictexception e) {
// 版本冲突，需要重新获取最新数据
log.warn("版本冲突，重新同步: {}", productid);
// 重试逻辑...
}
}
2. 重试机制
@service
public class essyncservice {
@autowired
private elasticsearchclient esclient;
@autowired
private redistemplate redistemplate;
// 最大重试次数
private static final int max_retry_count = 3;
public void syncwithretry(productindex index) {
string key = "es:retry:" + index.getid();
for (int i = 0; i < max_retry_count; i++) {
try {
esclient.index(req -> req
.index("product_index")
.id(index.getid().tostring())
.document(index)
);
// 成功后删除重试标记
redistemplate.delete(key);
return;
} catch (exception e) {
log.error("第{}次同步失败: {}", i+1, e.getmessage());
if (i == max_retry_count - 1) {
// 达到最大重试次数，记录失败任务
redistemplate.opsforvalue().set(key, json.tojsonstring(index), 7, timeunit.days);
log.error("同步失败，已记录到失败队列: {}", index.getid());
} else {
// 指数退避重试
try {
thread.sleep((long) (math.pow(2, i) * 1000));
} catch (interruptedexception ie) {
thread.currentthread().interrupt();
}
}
}
}
}
// 定时任务处理失败的同步任务
@scheduled(cron = "0 0/5 * * * ?")
public void processfailedtasks() {
set<string> keys = redistemplate.keys("es:retry:*");
if (keys != null) {
for (string key : keys) {
string json = (string) redistemplate.opsforvalue().get(key);
productindex index = json.parseobject(json, productindex.class);
// 重新尝试同步
syncwithretry(index);
}
}
}
}
3. 全量校验与修复
定期全量对比 mysql 和 es 数据，修复不一致：
@service
public class dataconsistencyservice {
@autowired
private productmapper productmapper;
@autowired
private elasticsearchclient esclient;
@scheduled(cron = "0 0 2 * * ?") // 每天凌晨2点执行
public void checkandrepair() {
log.info("开始数据一致性校验");
// 分页查询 mysql 数据
int page = 1;
int pagesize = 1000;
while (true) {
page<product> productpage = productmapper.selectpage(
new page<>(page, pagesize), null);
for (product product : productpage.getrecords()) {
try {
// 查询 es 数据
getresponse<productindex> response = esclient.get(req -> req
.index("product_index")
.id(product.getid().tostring()),
productindex.class
);
if (!response.found()) {
// es 中不存在，需要插入
synctoes(product);
log.warn("修复缺失数据: {}", product.getid());
} else {
// 对比数据是否一致
productindex esdata = response.source();
if (!isconsistent(product, esdata)) {
// 数据不一致，更新 es
synctoes(product);
log.warn("修复不一致数据: {}", product.getid());
}
}
} catch (exception e) {
log.error("校验商品 {} 失败: {}", product.getid(), e.getmessage());
}
}
if (productpage.hasnext()) {
page++;
} else {
break;
}
}
log.info("数据一致性校验完成");
}
private boolean isconsistent(product mysql, productindex es) {
// 比较关键字段
return objects.equals(mysql.gettitle(), es.gettitle()) &&
objects.equals(mysql.getprice(), es.getprice()) &&
objects.equals(mysql.getsales(), es.getsales());
}
}
四、性能优化策略
1. es 批量写入
public void batchsynctoes(list<product> products) {
if (collectionutils.isempty(products)) {
return;
}
try {
list<bulkoperation> operations = new arraylist<>();
for (product product : products) {
productindex index = converttoindex(product);
operations.add(bulkoperation.of(op -> op
.index(idx -> idx
.index("product_index")
.id(product.getid().tostring())
.document(index)
)
));
}
bulkrequest request = bulkrequest.of(req -> req.operations(operations));
bulkresponse response = esclient.bulk(request);
if (response.errors()) {
// 处理错误
for (bulkresponseitem item : response.items()) {
if (item.error() != null) {
log.error("批量同步失败: {} - {}",
item.id(), item.error().reason());
}
}
}
} catch (exception e) {
log.error("批量同步异常: {}", e.getmessage());
}
}
2. 优化 canal 配置
# 增加批处理大小 syncbatchsize = 2000 # 优化网络参数 tcp.so.sndbuf = 1048576 tcp.so.rcvbuf = 1048576 # 调整消费线程数 canal.instance.parser.parallel = true canal.instance.parser.parallelthreadsize = 8
3. mysql binlog 优化
# 增加 binlog 大小限制 binlog-file-size = 1g # 优化 binlog 刷新策略 sync_binlog = 1 innodb_flush_log_at_trx_commit = 1 # 调整 binlog 保留时间 expire_logs_days = 7
五、最佳实践建议
1. 方案选型建议
- 小型系统/快速迭代：异步双写（mq）
- 大型系统/高可靠：canal + mq
- 实时性要求极高：同步双写（权衡性能影响）
- 历史数据迁移：logstash 或 canal 全量同步
2. 监控与告警
@service
public class syncmonitorservice {
@autowired
private redistemplate redistemplate;
// 记录同步时间戳
public void recordsynctimestamp(string tablename, long id) {
string key = "sync:timestamp:" + tablename + ":" + id;
redistemplate.opsforvalue().set(key, system.currenttimemillis(), 24, timeunit.hours);
}
// 检查同步延迟
@scheduled(fixedrate = 60000)
public void checksyncdelay() {
// 查询最近5分钟内更新的数据
list<product> recentproducts = productmapper.selectrecentupdated(5);
for (product product : recentproducts) {
string key = "sync:timestamp:tb_product:" + product.getid();
long synctime = (long) redistemplate.opsforvalue().get(key);
if (synctime == null) {
// 未同步
sendalarm("数据未同步", product.getid());
} else {
long delay = system.currenttimemillis() - synctime;
if (delay > 300000) { // 5分钟
// 同步延迟过大
sendalarm("数据同步延迟:" + (delay/1000) + "秒", product.getid());
}
}
}
}
private void sendalarm(string message, long productid) {
// 发送告警（邮件、短信、钉钉等）
log.error("告警: {} - 商品id: {}", message, productid);
// 实际告警逻辑...
}
}
3. 数据同步异常处理流程
- 重试机制：指数退避策略，避免立即重试造成雪崩
- 死信队列：记录无法通过重试解决的异常
- 手动干预：提供管理界面手动触发同步
- 数据校验：定期全量比对，发现并修复不一致
六、总结
mysql 和 elasticsearch 数据同步是电商系统中的关键技术挑战。选择合适的同步方案需要综合考虑实时性要求、系统复杂度、团队技术栈等因素。在实际项目中，推荐采用 canal + 消息队列 的方案，它提供了良好的实时性、可靠性和扩展性，同时对业务代码零侵入。
无论选择哪种方案，都需要特别关注数据一致性保障、异常处理、性能优化和监控告警等方面，确保系统在生产环境中的稳定运行。
到此这篇关于mysql和elasticsearch数据同步方案详解的文章就介绍到这了,更多相关mysql和elasticsearch数据同步内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论