《Spring ES实战》第10招 DB如何同步到ES
# MySQL 到 Elasticsearch 数据一致性实战：从双写陷阱到可重建索引 ```text 旧方案：应用双写 MySQL 与 ES → 冲突：事务边界不一致、部分失败 → 后果：搜索结果与数据库长期偏差且难以发现 ``` 很多团队把 Elasticsearch 当成第二个数据库，应用层同时写 MySQL 和 ES。这个选择在流量小、数据简单时看不出问题，一旦出现部分失败、消息乱序或重复消费，搜索结果就开始“漂移”。 我是老李，过去半年在商品搜索项目里反复处理“数据库已改、搜索页还是旧数据”的工单。每次手工刷索引都像救火，直到我们把问题重新定义为：ES 不是真相源，而是 MySQL 的搜索投影。 为什么一个 @Transactional 无法同时保证 MySQL 与 Elasticsearch 原子提交？ 在双写、事务后事件、MQ、CDC 等方案之间，应该选择哪一条路？ 如果选错方案，会不会让索引偏差从“偶发”变成“常态化”，最后连对账都无从下手？ # 01、故事 ```text 周二 14:30 → 商品中心 → 任务：修复搜索页价格与详情页不一致 → 冲突：运营催、日志空、索引搜不到 ``` 那天下午，运营发来截图：App 搜索列表显示某商品 299 元，点进详情页却是 259 元。任务是定位为什么搜索索引里的价格没有更新。 我打开应用日志，没有异常堆栈。查 MySQL，商品价格已经是 259。查 Elasticsearch，文档里的 price 还是 299。中间发生了什么，日志里一片空白。 运营每五分钟催一次，因为促销页正在推这个商品。我们只能先手动执行一段脚本，把该商品重新写入 ES。问题暂时消失，但没人知道它什么时候会再出现。 > 价格已改库，索引仍如故。 > 日志无踪迹，催单声声促。 > 手工刷一条，心知非长路。 > 若问根何在，且向边界顾。 # 02、问题 ```text 需求变化：商品价格频繁变更 → 冲突：应用双写 MySQL 与 ES，ES 写失败被吞掉 → 后果：搜索结果与数据库不一致且无告警 ``` 商品价格变更本来只写 MySQL。后来搜索服务需要实时反映新价格，开发在更新商品的方法里加了一行 ES 更新调用。 问题在于：MySQL 提交成功，ES 调用超时或抛错时，异常被 catch 后只打了 debug 日志。业务认为更新完成，搜索索引却停留在旧值。 可验证的验收标准是：任意一次商品价格更新后，MySQL 与 ES 中的价格最终一致；即使 ES 短时间不可用，恢复后也能自动追上。 > 一入双写深似海，事务从此难自洽。 > 左库已定右索引，右路失手无人察。 > 若把投影当真相，偏差累积如雪下。 > 待到对账无凭据，方知边界是根芽。 # 03、原理 ```text 核心机制：MySQL 是 Source of Truth，ES 是 Search Projection；一致性靠“可发现、可重建”，不靠“强原子” ``` MySQL 的事务保证的是单库 ACID。Elasticsearch 的写入是近实时，refresh 前不可见，且没有跨库事务语义。 一个 @Transactional 只能包住本地数据源。ES 的 HTTP 调用发生在事务提交前或提交后，都无法与 MySQL 的提交点形成原子关系。 反直觉判断：搜索索引最重要的生产能力不是“永远不丢”，而是“出现偏差后可以发现，并且可以重建”。 > 事务如锁链，只锁自家门。 > 跨库无盟约，提交两处分。 > 若求强一致，架构必自困。 > 不如留凭证，偏差可追根。 # 04、架构 ```text 输入：MySQL 商品变更 → 本模块：Outbox 事件 + MQ + 索引写入器 → 输出：ES 商品文档；边界：不承担库存扣减等事务逻辑 ``` 我们把商品写路径拆成三层：业务层只写 MySQL 和 Outbox 表；投递层读取 Outbox 并发送 MQ；索引层消费消息并更新 ES。 这样做的收益是：MySQL 事务内只产生本地事件，不依赖外部系统；ES 写入失败可以重试，不影响主流程。 代价是链路变长，需要处理重复消费、乱序和延迟。适用条件是：业务能接受秒级最终一致，且商品变更频率在 MQ 吞吐范围内。 > 库内留凭据，消息作舟桥。 > 索引成投影，失败可重挑。 > 链路虽多曲，边界却明了。 > 若问何时用，最终一致好。 # 05、第一版实战 ```text 环境：MySQL 8.0 + RabbitMQ 3.12 + Elasticsearch 8.11 → 配置：Outbox 表 + 交换器 → 代码：商品服务 + 索引消费者 → 启动 → 验证 → V1 ``` 先建 Outbox 表： ```sql CREATE TABLE product_outbox ( event_id BIGINT AUTO_INCREMENT PRIMARY KEY, product_id BIGINT NOT NULL, event_type VARCHAR(32) NOT NULL, payload JSON NOT NULL, status VARCHAR(16) NOT NULL DEFAULT 'PENDING', retry_count INT NOT NULL DEFAULT 0, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_status_created (status, created_at) ) ENGINE=InnoDB; ``` 商品服务在同一个本地事务里写商品表和 Outbox 表： ```java @Transactional public void updateProductPrice(Long productId, BigDecimal newPrice) { productMapper.updatePrice(productId, newPrice); ProductPriceChanged event = new ProductPriceChanged(productId, newPrice); outboxMapper.insert(event.toOutboxRow()); } ``` 投递器每 2 秒扫描 PENDING 事件，发送到 RabbitMQ，发送成功后标记 SENT： ```java List rows = outboxMapper.findPending(100); for (OutboxRow row : rows) { rabbitTemplate.convertAndSend("product.exchange", "price.changed", row.getPayload()); outboxMapper.markSent(row.getEventId()); } ``` 索引消费者监听队列，写入 ES： ```java @RabbitListener(queues = "product.index.queue") public void onMessage(ProductPriceChanged event) { IndexRequest request = new IndexRequest("product") .id(String.valueOf(event.getProductId())) .source(Map.of("price", event.getNewPrice())); esClient.index(request, RequestOptions.DEFAULT); } ``` 启动商品服务、投递器和消费者。验证请求： ```bash curl -X PUT "http://localhost:9200/product/_doc/1001" -H 'Content-Type: application/json' -d '{"price":299}' curl -X POST "http://localhost:8080/product/1001/price?value=259" curl -X GET "http://localhost:9200/product/_doc/1001" ``` 示例输出： ```text {"_index":"product","_id":"1001","_source":{"price":259}} ``` 成功标准：MySQL 价格变更后，ES 文档在 5 秒内更新为 259。未在当前环境实测，以下为预期结果。 > 库内同事务，凭据落表中。 > 投递如信使，消息过桥东。 > 索引随波至，价格换新容。 > 初版虽可跑，暗礁待浪冲。 # 06、排查 ```text 诊断树：ES 文档未更新 → 是消息未发出？是消费失败？是重复覆盖？→ 查 Outbox 状态、MQ 队列、消费者日志 ``` 第一个问题：商品更新后 ES 没变。怀疑投递器没扫到事件。检查 Outbox 表： ```sql SELECT event_id, status, retry_count, updated_at FROM product_outbox ORDER BY event_id DESC LIMIT 5; ``` 示例输出： ```text event_id=88, status=PENDING, retry_count=0, updated_at=14:31:02 ``` 事件一直 PENDING，说明投递器扫描条件有误。 检查代码，发现 findPending 的 SQL 多了 `AND created_at > NOW() - INTERVAL 1 MINUTE`，导致积压事件被跳过。 修复后事件正常发送。 第二个问题：消息重复消费导致旧值覆盖新值。消费者日志显示同一 productId 被处理两次，第二次是旧消息。原因是 MQ 重投时没有幂等控制。 **错误尝试：** 我先在消费者里加 `if (exists) skip`，但 exists 判断和 index 不是原子操作，并发下仍会覆盖。 > 事件久未发，扫描漏旧单。 > 消息重投至，旧值覆新颜。 > 先查表与队，再验消费端。 > 若凭猜测定，徒劳绕圈圈。 # 07、优化 ```text V1：无幂等、无重试、无对账 → 证据：重复消费覆盖新值、ES 宕机丢消息 → V2：eventId 幂等 + 重试 + DLQ + 对账任务 ``` 给每条事件生成全局唯一 eventId，写入 Outbox 和消息头。消费者用 eventId 做幂等： ```java String eventId = message.getMessageProperties().getHeader("eventId"); UpdateRequest request = new UpdateRequest("product", String.valueOf(productId)) .doc(Map.of("price", newPrice, "last_event_id", eventId)) .upsert(Map.of("price", newPrice, "last_event_id", eventId)); esClient.update(request, RequestOptions.DEFAULT); ``` 消费失败时重试 3 次，仍失败则进入 DLQ： ```yaml spring: rabbitmq: listener: simple: retry: enabled: true max-attempts: 3 default-requeue-rejected: false ``` DLQ 消费者记录失败事件，等待对账任务处理。对账任务每小时扫描 MySQL 近一小时变更的商品，与 ES 文档比对，不一致则重新索引。 > 事件带唯一，幂等有凭依。 > 重试限次数，死信不丢弃。 > 对账如巡夜，偏差无所匿。 > 优化非终点，演进见真义。 # 08、演进 ```text V2：幂等 + 重试 + DLQ + 对账 → 能力：ES 短时不可用可恢复 → 遗留：对账延迟、DLQ 积压需人工介入 ``` 用 V1 同一组请求验证：先停掉 ES，更新商品价格，再启动 ES。 ```bash systemctl stop elasticsearch curl -X POST "http://localhost:8080/product/1001/price?value=199" systemctl start elasticsearch ``` 消费者重试 3 次失败后进入 DLQ。对账任务在下一轮扫描时发现 MySQL 价格 199 与 ES 价格 259 不一致，重新索引。 示例输出： ```text reconcile: product_id=1001, mysql_price=199, es_price=259, action=reindex ``` 收益是：ES 宕机不再导致数据永久丢失。代价是：对账任务有小时级延迟，DLQ 积压需要监控和人工处理。 > 停索引如断桥，消息暂入巢。 > 重试三番后，死信待晨朝。 > 对账巡夜至，偏差尽可消。 > 遗留非终局，演进路迢迢。 # 09、洞见 ```text 从现象到判断：双写之痛不在代码，而在边界；一致性之难不在技术，而在发现 ``` ## 反直觉判断：强一致不是搜索索引的目标 ```text 强一致 → 高耦合 → 低可用 ``` 搜索场景下，用户对秒级延迟几乎无感，但对“搜不到”或“价格错”非常敏感。把 ES 当强一致系统，只会让商品主流程被索引拖垮。 ## 反直觉判断：Outbox 的价值不在“发出去”，而在“留得下” ```text Outbox → 本地事务凭证 → 可重放 ``` 很多人以为 Outbox 是为了可靠投递。其实它最大的价值是：MySQL 里永远有一份“应该发生什么”的凭证，索引可以随时从这份凭证重建。 ## 幂等不是“重复不处理”，而是“重复处理结果不变” ```text 重复消息 → 幂等键 → 最终状态一致 ``` 用 eventId 做幂等，不是简单地丢弃重复消息，而是保证无论消息来几次，ES 文档的最终状态都与最新事件一致。 ## 对账是最后一道防线，不是补丁 ```text 对账 → 发现偏差 → 触发重建 ``` 对账任务不是“出了问题再跑一下”的工具，而是常态化的巡检机制。它让偏差从“用户投诉才发现”变成“系统自己发现”。 > 边界若不明，双写必成灾。 > 凭证若不留，偏差无处猜。 > 幂等若不做，旧值覆新来。 > 对账若常在，索引可重栽。 # 10、系统落地 ```text 入口：商品管理后台 → 服务：商品服务写 MySQL + Outbox → 本模块：投递器 + MQ + 索引消费者 + 对账任务 → 存储：MySQL 商品表、Outbox 表、ES 商品索引 → 未实现：CDC 实时监听 Binlog ``` 原来商品服务只写 MySQL，搜索服务直接查库。本篇增加了 Outbox 事件、MQ 投递、ES 索引写入和对账任务。 现在商品价格变更后，搜索索引能在秒级最终一致；ES 宕机恢复后，对账任务能自动修复偏差。 下一篇缺的是基于 Debezium 的 CDC 方案，直接从 Binlog 捕获变更，减少应用层 Outbox 的维护成本。官方文档可参考 MySQL 与 Elasticsearch 官网主页。 > 入口在后台，服务写库忙。 > 本篇添凭证，消息过桥梁。 > 索引终可建，对账护周详。 > 若问下一程，Binlog 待启航。 # 11、小结 ```text 三问三答：事务为何不原子？方案如何选？选错有何果？ ``` 为什么一个 @Transactional 无法同时保证 MySQL 与 Elasticsearch 原子提交？因为 ES 是外部 HTTP 系统，不在本地事务边界内，提交点无法对齐。 在双写、事务后事件、MQ、CDC 之间如何选？业务能接受秒级延迟时，Outbox + MQ 是性价比最高的方案；追求低侵入可选 CDC。 选错方案会怎样？双写吞异常会让偏差常态化；无对账的 MQ 方案会让消息丢失不可见。当前状态：V2 已上线，对账任务每半小时运行一次。 > 一问事务界，跨库无盟约。 > 二问方案选，Outbox 最可托。 > 三问选错果，偏差积成疴。 > 当前已落地，对账护城河。 # 12、作业 ```text 理解 → 实战 → 排障 → 架构判断 ``` ## 理解题：为什么 Outbox 表必须和业务表在同一个本地事务中写入？ 因为只有同一个本地事务才能保证“业务变更”和“事件凭证”同时成功或同时回滚。如果分开写，业务成功但 Outbox 失败，事件就丢了，索引永远追不上。 ## 实战题：实现一个对账任务，扫描 MySQL 中近 10 分钟变更的商品，与 ES 文档比对价格，不一致则重新索引。 先查 MySQL 变更记录，再批量查 ES 文档，逐条比对 price 字段。不一致的商品构造 IndexRequest 重新写入。注意控制批量大小，避免 ES 压力过大。 ## 排障题：消费者日志显示消息被消费，但 ES 文档始终不更新，如何定位？ 先确认消费者是否真正执行了 index 或 update 请求，检查 ES 客户端是否抛出异常。 再查 ES 索引是否存在、mapping 是否冲突、refresh 是否未触发。 最后用 curl 手动写一条测试文档验证。 ## 架构判断题：如果商品变更频率极高，Outbox + MQ 方案是否仍然适用？ 需要评估 MQ 吞吐和 ES 写入能力。如果单表 Outbox 扫描成为瓶颈，可以分表或改用 CDC 直接消费 Binlog。但无论哪种方案，对账和幂等机制都不能省略。 > 理解事务界，凭证同生灭。 > 实战对账路，偏差尽可截。 > 排障循链路，证据不可缺。 > 架构判高低，吞吐见真诀。 # 13、升华 ```text 工程结论：搜索索引的生命力不在“不丢”，而在“可发现、可重建” ``` MySQL 是真相源，ES 是搜索投影。这个定位一旦确立，很多看似复杂的一致性问题就变成了工程边界问题。 Outbox 留下凭证，MQ 负责传递，幂等保证结果，对账负责发现，重建负责修复。这四件事组合起来，才是一套可运行的索引一致性体系。 下一篇可以深入 CDC 方案，用 Debezium 捕获 Binlog，把 Outbox 的维护成本进一步降低。 > 真相在库中，投影在索引。 > 凭证留得住，偏差可追寻。 > 对账如明镜，重建如回春。 > 若问下一程，Binlog 待深论。 **系列项目：** 商品搜索与索引一致性系统 **已完成：** MySQL 商品存储、应用双写 ES、Outbox 事件、MQ 投递、幂等消费、DLQ、对账任务 **本篇：** 从双写陷阱到 Outbox + MQ + 对账的最终一致方案 **下一篇：** 基于 Debezium 的 CDC 实时索引方案
