Elasticsearch与MySQL的数据一致性保障：双写、CDC与最终对账方案对比
一、"搜索结果里有，但点进去页面404"——ES与MySQL不一致的100种方式
搜索"红色连衣裙"，ES 返回了商品 ID 12345，用户点击后跳转到商品详情页，MySQL 查询却返回空——因为商品已在 2 分钟前下架，但 ES 还没收到同步信号。这个场景每天在成千上万的电商系统中上演，根因就是 MySQL 和 ES 之间的数据不一致。
ES 和 MySQL 本质上是两个独立的数据库，它们之间不存在天然的分布式事务。保障两者数据一致性的核心挑战是：当 MySQL 写入成功后，如何确保 ES 也一定能收到并处理变更事件？本文对比三种主流方案——双写、CDC（Change 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[Canal<br/>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( index='orders', id=order_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(f"ES 写入失败: {es_result}") # 3. ES 写入成功后才提交 MySQL mysql_conn.commit() except Exception as e: mysql_conn.rollback() # 即使 MySQL 回滚成功，ES 可能已经写入（ES 不支持回滚） # 这种情况需要最终对账兜底 raise e return True方案2: CDC（Canal 监听 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( host=canal_config['host'], port=canal_config['port'] ) self.client.subscribe( client_id=canal_config['client_id'], destination=canal_config['destination'], filter=canal_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( index=entry['tableName'], id=after_data['id'], body=after_data ) elif event_type == 'UPDATE': # CALL: 更新 ES after_data = self._parse_row(row_change.get('afterColumns')) self.es.update( index=entry['tableName'], id=after_data['id'], body={'doc': after_data} ) elif event_type == 'DELETE': # CALL: 从 ES 删除 before_data = self._parse_row(row_change.get('beforeColumns')) self.es.delete( index=entry['tableName'], id=before_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( index='orders', 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(index='orders', id=row['id'], body=row) for item in inconsistent: es_client.update( index='orders', id=item['id'], body={'doc': item['mysql']} ) # 5. 更新检查点 new_checkpoint = mysql_rows[-1]['id'] save_checkpoint(new_checkpoint)四、方案对比与选择
| 维度 | 双写 | CDC (Canal) | 最终对账 |
|---|---|---|---|
| 实时性 | 实时（秒级） | 准实时（秒级） | 延迟（分钟级） |
| 一致性保障 | 弱（无原子性） | 强（按顺序回放） | 最终一致 |
| 入侵性 | 高（修改业务代码） | 无（监听Binlog） | 低（独立任务） |
| 架构复杂度 | 低 | 中（需Canal+MQR） | 低 |
| 并发送性能影响 | 有（ES写入在事务中） | 无 | 低（异步） |
| 异常恢复 | 需手动处理中间态 | 从Binlog位点恢复 | 自动补偿 |
推荐策略：
- 写入量 < 1000 TPS：双写 + 对账兜底
- 写入量 > 1000 TPS：CDC + 对账兜底
- 容忍 5 分钟延迟：仅对账即可
五、总结
MySQL 和 ES 的数据一致性没有完美方案，核心在于根据业务容忍度选择合适的策略：
- 双写适合简单场景但风险最高：ES 写入失败会导致 MySQL 事务回滚，但 ES 不支持回滚——双写需要最终对账兜底
- CDC 是最优雅的方案：零业务入侵、顺序回放、天然解决"回滚"问题——Binlog 中不会记录已回滚的事务
- 最终对账是任何方案的安全网：无论选择双写还是 CDC，都必须有定时对账任务作为兜底
在实际电商系统中，采用 CDC + 5 分钟对账的混合方案后，MySQL 与 ES 的数据不一致率从 0.3% 降低到 0.001%，<5 分钟内的不一致在用户可接受的范围内（用户不会在创建订单后立刻去搜索）——更重要的是，不会再有"搜索结果 404"的用户投诉。
)
)
 agent和组件)