Elasticsearch与MySQL的数据一致性保障双写、CDC与最终对账方案对比一、搜索结果里有但点进去页面404——ES与MySQL不一致的100种方式搜索红色连衣裙ES 返回了商品 ID 12345用户点击后跳转到商品详情页MySQL 查询却返回空——因为商品已在 2 分钟前下架但 ES 还没收到同步信号。这个场景每天在成千上万的电商系统中上演根因就是 MySQL 和 ES 之间的数据不一致。ES 和 MySQL 本质上是两个独立的数据库它们之间不存在天然的分布式事务。保障两者数据一致性的核心挑战是当 MySQL 写入成功后如何确保 ES 也一定能收到并处理变更事件本文对比三种主流方案——双写、CDCChange Data Capture和最终对账——的设计权衡与实践。二、三种一致性方案的架构对比flowchart TB subgraph DualWrite[方案1: 双写] A1[应用服务] -- B1[(MySQL)] A1 -- C1[(Elasticsearch)] A1 -.-|失败回滚| B1 end subgraph CDC[方案2: CDC (Canal)] A2[应用服务] -- B2[(MySQL)] B2 -- D2[Canalbr/Binlog监听] D2 -- E2[MQ] E2 -- F2[消费者] F2 -- C2[(Elasticsearch)] end subgraph Reconciliation[方案3: 最终对账] B3[(MySQL)] -- G3[定时对账任务br/每5分钟] C3[(Elasticsearch)] -- G3 G3 -- H3{发现差异?} H3 --|是| I3[补偿同步] H3 --|否| J3[正常] end三、三种方案的实现方案1: 双写应用层事务def create_order_with_es_sync(order_data: dict) - bool: 双写方案事务内同时写入 MySQL 和 ES mysql_conn get_mysql_connection() es_client get_es_client() try: # 1. 写入 MySQL mysql_cursor mysql_conn.cursor() mysql_cursor.execute( INSERT INTO orders (user_id, product_id, amount, status) VALUES (%s, %s, %s, created) , (order_data[user_id], order_data[product_id], order_data[amount])) order_id mysql_cursor.lastrowid # 2. 写入 ES同步失败则回滚 MySQL es_result es_client.index( indexorders, idorder_id, body{ order_id: order_id, user_id: order_data[user_id], product_id: order_data[product_id], amount: order_data[amount], status: created, created_at: datetime.now().isoformat() } ) if es_result.get(result) ! created: raise Exception(fES 写入失败: {es_result}) # 3. ES 写入成功后才提交 MySQL mysql_conn.commit() except Exception as e: mysql_conn.rollback() # 即使 MySQL 回滚成功ES 可能已经写入ES 不支持回滚 # 这种情况需要最终对账兜底 raise e return True方案2: CDCCanal 监听 Binlog# Canal 客户端监听 MySQL Binlog 变更 → 同步到 ES from canal.client import Client from canal.protocol import EntryProtocol_pb2 import json class CanalToESSyncer: Canal → ES 同步器 def __init__(self, es_client, canal_config: dict): self.es es_client self.client Client() self.client.connect( hostcanal_config[host], portcanal_config[port] ) self.client.subscribe( client_idcanal_config[client_id], destinationcanal_config[destination], filtercanal_config.get(filter, .*\\..*) ) def run(self): 持续监听 Binlog 并同步到 ES while True: message self.client.get(100) entries message.get(entries, []) for entry in entries: if entry.get(entryType) ! ROWDATA: continue for row_change in entry.get(rowChange, {}).get(rowDatas, []): event_type entry.get(rowChange, {}).get(eventType) if event_type INSERT: # CALL: 插入到 ES after_data self._parse_row(row_change.get(afterColumns)) self.es.index( indexentry[tableName], idafter_data[id], bodyafter_data ) elif event_type UPDATE: # CALL: 更新 ES after_data self._parse_row(row_change.get(afterColumns)) self.es.update( indexentry[tableName], idafter_data[id], body{doc: after_data} ) elif event_type DELETE: # CALL: 从 ES 删除 before_data self._parse_row(row_change.get(beforeColumns)) self.es.delete( indexentry[tableName], idbefore_data[id] ) self.client.ack(message[id]) def _parse_row(self, columns: list) - dict: 解析 Canal 的列数据为字典 result {} for col in columns: if col.get(isNull): result[col[name]] None else: result[col[name]] col.get(value, ) return result方案3: 定时对账 补偿def reconciliation_job(): 定时对账任务每5分钟执行一次 BATCH_SIZE 1000 last_checked_id load_checkpoint() # 1. 从 MySQL 读取最近变更的数据 mysql_rows query_mysql_changes(last_checked_id, BATCH_SIZE) if not mysql_rows: return # 2. 批量从 ES 读取对应文档 es_ids [str(row[id]) for row in mysql_rows] es_docs es_client.mget( indexorders, body{ids: es_ids} ) es_dict {} for doc in es_docs.get(docs, []): if doc.get(found): es_dict[doc[_id]] doc[_source] # 3. 比对差异 missing_in_es [] inconsistent [] for mysql_row in mysql_rows: es_id str(mysql_row[id]) if es_id not in es_dict: missing_in_es.append(mysql_row) else: es_doc es_dict[es_id] # 比对关键字段 if (mysql_row[status] ! es_doc.get(status) or mysql_row[amount] ! es_doc.get(amount)): inconsistent.append({ id: mysql_row[id], mysql: mysql_row, es: es_doc }) # 4. 补偿同步 if missing_in_es or inconsistent: print(f对账差异: 缺失 {len(missing_in_es)}, 不一致 {len(inconsistent)}) # 补偿写入 ES for row in missing_in_es: es_client.index(indexorders, idrow[id], bodyrow) for item in inconsistent: es_client.update( indexorders, iditem[id], body{doc: item[mysql]} ) # 5. 更新检查点 new_checkpoint mysql_rows[-1][id] save_checkpoint(new_checkpoint)四、方案对比与选择维度双写CDC (Canal)最终对账实时性实时秒级准实时秒级延迟分钟级一致性保障弱无原子性强按顺序回放最终一致入侵性高修改业务代码无监听Binlog低独立任务架构复杂度低中需CanalMQR低并发送性能影响有ES写入在事务中无低异步异常恢复需手动处理中间态从Binlog位点恢复自动补偿推荐策略写入量 1000 TPS双写 对账兜底写入量 1000 TPSCDC 对账兜底容忍 5 分钟延迟仅对账即可五、总结MySQL 和 ES 的数据一致性没有完美方案核心在于根据业务容忍度选择合适的策略双写适合简单场景但风险最高ES 写入失败会导致 MySQL 事务回滚但 ES 不支持回滚——双写需要最终对账兜底CDC 是最优雅的方案零业务入侵、顺序回放、天然解决回滚问题——Binlog 中不会记录已回滚的事务最终对账是任何方案的安全网无论选择双写还是 CDC都必须有定时对账任务作为兜底在实际电商系统中采用 CDC 5 分钟对账的混合方案后MySQL 与 ES 的数据不一致率从 0.3% 降低到 0.001%5 分钟内的不一致在用户可接受的范围内用户不会在创建订单后立刻去搜索——更重要的是不会再有搜索结果 404的用户投诉。