简介这份资源是面向Java后端开发者的实战Demo聚焦Spring Boot整合Elasticsearch 8.3并借助RabbitMQ完成MySQL到搜索引擎的实时数据同步适合正在接触微服务、消息队列与全文检索的中高级开发者参考。压缩包共78个文件约90KB以21个java源码和21个class文件为核心配合17个xml配置、2个yml文件及mvnw、cmd等构建脚本完整呈现可运行的工程结构。内容围绕MySQL变更经RabbitMQ投递、Spring Boot消费后写入Elasticsearch的数据链路展开涉及Repository操作、消息监听、批量写入与错误重试等关键环节并留有测试与监控的扩展空间。目前已有209人学习下载可帮助读者快速理解关系型数据库与搜索引擎之间的同步思路掌握消息驱动架构的落地方式并作为自身项目改造的起点。1. 当 MySQL 写入后 ES 搜不到这套同步链路到底在解决什么电商后台刚上线那会儿运营在 MySQL 里改完商品标题前台搜索页刷新十遍还是旧数据。问题不在 Elasticsearch 慢而在没人把变更推过去。SpringBoot 整合 Elasticsearch 8.3 并通过 RabbitMQ 同步 MySQL 数据库本质就是给「写库」和「可搜索」之间架一条可靠的消息管道业务只写 MySQL一条消息丢进 RabbitMQ消费者负责把数据落到 ES 索引里。它适合中小团队做商品、文章、订单这类需要全文检索又要保证主库权威性的场景。难点从来不是 API 调用而是消息顺序、失败重试和全量增量怎么衔接——这三件事没想清楚Demo 跑通那天就是线上翻车的起点。2. 环境与依赖把 SpringBoot、ES 8.3、RabbitMQ 拉到同一张桌子上2.1 版本对齐比写代码更重要Elasticsearch 8.x 默认开启安全认证spring-boot-starter-data-elasticsearch的版本必须和 ES 服务端主版本一致否则会出现连接握手失败或者RestClient抛UnsupportedProductException。我一般用 SpringBoot 3.1.x 搭配 ES 8.3RabbitMQ 用 3.11 以上因为 3.11 对 quorum queue 的支持更稳后面做消费端幂等会用到。依赖清单如下注意spring-boot-starter-amqp和spring-boot-starter-data-elasticsearch都要显式引入不要指望传递依赖帮你带全。!-- pom.xml 关键依赖版本按你的 SpringBoot 父工程对齐 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-elasticsearch/artifactId /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency dependency groupIdcom.mysql/groupId artifactIdmysql-connector-j/artifactId scoperuntime/scope /dependency逻辑说明spring-boot-starter-data-elasticsearch在 8.x 里底层用的是新的ElasticsearchClient不再是 7.x 的RestHighLevelClient所以网上很多 7.x 的配置类直接抄过来会编译不过。参数上MySQL 驱动用mysql-connector-j这是 8.0.31 之后的新坐标老坐标mysql-connector-java已经停止维护。2.2 配置文件里三个必须写死的参数spring: elasticsearch: uris: https://127.0.0.1:9200 username: elastic password: your_password connection-timeout: 5s socket-timeout: 30s rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest listener: simple: acknowledge-mode: manual # 手动 ack防止消息丢了还不知道 prefetch: 20 # 每次拉 20 条别一次拉太多压垮 ES逻辑说明acknowledge-mode: manual是这套链路的关键自动 ack 会在消费者抛异常时把消息吞掉MySQL 改了但 ES 没更新你连后悔药都没得吃。prefetch设 20 是经验值设太大批量写 ES 容易触发429 Too Many Requests设太小吞吐上不去。socket-timeout给 30 秒因为 ES 批量写入在磁盘紧张时会卡住超时太短会误判失败。2.3 索引结构先定好别等数据灌进去再改商品索引的 mapping 建议提前用 Kibana 或ElasticsearchClient创建字段类型一旦写入再改就要 reindex。下面是一个最小可用的商品索引结构。PUT /product_index { settings: { number_of_shards: 1, number_of_replicas: 0 }, mappings: { properties: { id: { type: long }, title: { type: text, analyzer: standard }, price: { type: scaled_float, scaling_factor: 100 }, update_time: { type: date, format: yyyy-MM-dd HH:mm:ss } } } }逻辑说明scaled_float存价格比double省空间且避免浮点误差scaling_factor: 100表示保留两位小数。number_of_replicas在单机 Demo 里设 0否则集群健康状态会一直黄看着难受。update_time的格式要和 Java 侧序列化格式严格一致否则写入报mapper_parsing_exception。3. 同步链路怎么搭从 MySQL 变更到 RabbitMQ 消息再到 ES 文档3.1 消息生产端业务写完库再发消息顺序不能反常见做法是在 Service 层写完 MySQL 后手动发一条消息消息体只带主键 ID 和操作类型不带完整数据。这样做的理由是消息体越小越不容易在队列里堆积消费者拿到 ID 再回查 MySQL 拿最新数据能避免「消息里的数据比库里的旧」这种时序问题。Service public class ProductService { Autowired private ProductMapper productMapper; Autowired private RabbitTemplate rabbitTemplate; // 保存商品后发消息只传 id 和操作类型 Transactional(rollbackFor Exception.class) public void saveProduct(Product product) { productMapper.insert(product); // 事务提交后再发这里用 afterCommit 更稳Demo 先简化 rabbitTemplate.convertAndSend( product.exchange, // 交换机 product.sync, // 路由键 new SyncMessage(product.getId(), UPSERT) ); } }逻辑说明convertAndSend的第二个参数是路由键要和后面队列绑定的键一致。参数SyncMessage是一个简单 DTO只含id和op两个字段。注意这里有个坑如果发消息在事务提交前执行事务回滚了消息却发出去了消费者回查 MySQL 查不到数据会反复重试。生产环境建议用TransactionSynchronizationManager的afterCommit回调或者本地消息表方案。3.2 队列与交换机声明用 Bean 方式自动创建Configuration public class RabbitConfig { // 声明交换机durabletrue 表示重启后还在 Bean public DirectExchange productExchange() { return new DirectExchange(product.exchange, true, false); } // 声明队列用 quorum 类型更抗消息丢失 Bean public Queue productQueue() { return QueueBuilder.durable(product.sync.queue) .withArgument(x-queue-type, quorum) .build(); } // 绑定队列到交换机 Bean public Binding productBinding() { return BindingBuilder.bind(productQueue()) .to(productExchange()) .with(product.sync); } }逻辑说明DirectExchange的第二个参数durable为 true第三个autoDelete为 false这是生产配置。x-queue-type: quorum是 RabbitMQ 3.8 之后引入的队列类型基于 Raft 协议比经典队列更不容易丢消息代价是性能略低。如果 RabbitMQ 版本低于 3.8这个参数会报错去掉即可。3.3 消费端手动 ack 加回查 MySQL 再写 ESComponent public class ProductSyncConsumer { Autowired private ProductMapper productMapper; Autowired private ElasticsearchClient esClient; RabbitListener(queues product.sync.queue) public void onMessage(SyncMessage msg, Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException { try { // 回查 MySQL 拿最新数据避免消息体数据过期 Product product productMapper.selectById(msg.getId()); if (product null) { // 数据已删除从 ES 删掉对应文档 esClient.delete(d - d.index(product_index).id(String.valueOf(msg.getId()))); } else { // 写入或更新 ES 文档 esClient.index(i - i.index(product_index) .id(String.valueOf(product.getId())) .document(product)); } // 处理成功手动 ack channel.basicAck(tag, false); } catch (Exception e) { // 处理失败拒绝并重新入队注意设置重试上限 channel.basicNack(tag, false, true); } } }逻辑说明basicAck的第二个参数multiple为 false表示只确认当前这条。basicNack的第三个参数requeue为 true 会把消息重新放回队列但如果不设重试上限一条坏消息会无限循环。生产环境要么用死信队列兜底要么在消息头里记重试次数超过 3 次就丢到死信队列人工处理。esClient.index的id用 MySQL 主键保证同一条数据在 ES 里是覆盖写而不是新增。3.4 全量同步增量跑通后别忘了补历史数据增量链路跑通后历史数据还在 MySQL 里躺着。常见做法是写一个一次性任务分页查 MySQL 然后批量写 ES走bulk接口而不是单条index。public void fullSync() throws IOException { int pageSize 500; long lastId 0; while (true) { // 按主键游标分页避免深分页 ListProduct list productMapper.selectByCursor(lastId, pageSize); if (list.isEmpty()) break; ListBulkOperation ops new ArrayList(); for (Product p : list) { ops.add(BulkOperation.of(b - b .index(i - i.index(product_index) .id(String.valueOf(p.getId())) .document(p)))); } esClient.bulk(b - b.operations(ops)); lastId list.get(list.size() - 1).getId(); } }逻辑说明用主键游标分页而不是limit offset是因为offset越大 MySQL 扫描行数越多百万级数据会拖垮主库。pageSize设 500 是批量写入的折中值太大容易触发 ES 的413 Request Entity Too Large太小则网络往返次数多。BulkOperation是 ES 8.x Java Client 的批量操作封装比 7.x 的BulkRequest写法更简洁。4. 避坑与排查这五条血泪经验能省你两天4.1 消息发了但 ES 没数据先看 ack 模式现象MySQL 数据变了RabbitMQ 控制台显示消息已被消费但 ES 里查不到。原因acknowledge-mode配成了auto消费者抛异常时消息被自动确认等于丢了。解决改成manual并在 catch 块里显式basicNack同时给队列配死信交换机重试超过三次的消息进死信队列。4.2 批量写入报 429别急着加机器现象全量同步时 ES 返回429 Too Many Requests日志里一堆circuit_breaking_exception。原因prefetch设太大或者bulk批次太大ES 的写队列满了。解决把prefetch降到 10 以下bulk每批控制在 500 条以内同时检查 ES 的thread_pool.write.queue_size默认 1000不够可以适当调大但根本办法是降低写入速率。4.3 中文搜索搜不到多半是分词器没装现象标题里明明有「无线蓝牙耳机」搜「蓝牙」却返回空。原因ES 默认standard分词器把中文按单字切搜「蓝牙」匹配不上。解决装analysis-ik插件把title字段的analyzer改成ik_max_wordsearch_analyzer改成ik_smart。注意插件版本必须和 ES 8.3 严格对应版本错了 ES 起不来。4.4 消息顺序乱了后写的旧数据覆盖了新数据现象先改价格再改标题ES 里标题是新的但价格是旧的。原因RabbitMQ 单队列单消费者能保证顺序但如果你开了多个消费者并发消费顺序就没了。解决要么用单消费者要么在消息里带update_time消费时比较 ES 里已有文档的时间戳旧的直接丢弃。我一般用后者因为单消费者吞吐上不去。4.5 应用重启后消息重复消费现象服务重启RabbitMQ 把未 ack 的消息重新投递ES 里出现重复文档。原因basicNack的requeue为 true 或者消费者处理到一半进程挂了。解决ES 写入用主键 ID 做覆盖写天然幂等如果业务逻辑不是幂等的在 Redis 里记一个msgId的消费标记处理前先查标记。注意msgId要在生产端生成并放进消息头。5. 进阶技巧用版本号加死信队列把同步做成可运维的链路Demo 跑通只是起点真正上线要考虑的是「出问题了怎么查、怎么补」。我习惯在消息体里加一个version字段取 MySQL 记录的update_time时间戳消费端写入 ES 前先get一下文档比较版本号旧版本直接 ack 丢弃。这样即使消息乱序也不会污染数据。// 消费端版本比较防止旧消息覆盖新数据 GetResponseProduct existing esClient.get(g - g .index(product_index).id(String.valueOf(msg.getId())), Product.class); if (existing.found() existing.source().getUpdateTime() .after(product.getUpdateTime())) { channel.basicAck(tag, false); // 旧数据直接确认丢弃 return; }逻辑说明existing.source()拿到 ES 里已有文档比较updateTime。注意GetResponse在文档不存在时found()返回 false不会抛异常所以要先判断。这个方案的前提是 MySQL 的update_time字段在每次更新时自动刷新用ON UPDATE CURRENT_TIMESTAMP保证。死信队列的配置也不复杂在声明业务队列时加两个参数x-dead-letter-exchange指向死信交换机x-dead-letter-routing-key指向死信路由键。消费端basicNack时requeue设 false消息就会进死信队列。然后单独起一个消费者监听死信队列把消息落库到一张sync_fail_log表人工排查后重新投递。这套组合下来同步链路就从「黑匣子」变成了可观测、可干预的管道。最后说个我自己的习惯每次改完同步逻辑先往 MySQL 插一条测试数据盯着 RabbitMQ 控制台的消息速率和 ES 的_cat/indices文档数变化确认三个环节都对得上再继续。这个动作花不了一分钟但能帮你避开大部分「以为通了其实没通」的玄学问题。希望帮到你。本文还有配套的精品资源点击获取