简介这份资源面向具备一定 Spring Boot 基础、希望掌握实时搜索与数据同步方案的后端开发者围绕 Spring Boot 整合 Elasticsearch 8.3 并通过 RabbitMQ 同步 MySQL 数据展开覆盖依赖配置、Repository 操作、消息监听与发送、事务事件监听、批量写入与错误重试等关键环节可用于电商检索、日志分析等需要关系型数据库与搜索引擎协同的场景。压缩包共 78 个文件约 90KB以 21 个 java 源码与 21 个 class 编译文件为主体辅以 17 个 xml 配置、2 个 yml 配置文件及证书、构建脚本等完整呈现可运行的工程骨架。已有 214 人学习下载读者可据此理清 MySQL 到 RabbitMQ 再到 Elasticsearch 的数据链路参考消息驱动同步与批量处理的实现思路并借助单元测试与监控配置排查常见问题。1. Spring Boot 整合 Elasticsearch 8.3 与 RabbitMQ先把版本对齐再动手上周帮一位同事排查环境现象很典型Spring Boot 3.2 项目里引了spring-boot-starter-data-elasticsearch本地跑得好好的换到另一台机器启动就报NoNodeAvailableException日志里连一个可用的节点都找不到。翻到最后发现Elasticsearch 服务端装的是 7.17客户端走的是 8.x 的 REST 协议安全认证和连接握手全对不上。这类问题在 spring boot rabbitmq elasticsearch8 的组合里特别常见——三个组件各自有版本节奏任何一个错半拍启动就翻车。这套环境适合两类人一是要在业务系统里做 MySQL 到 Elasticsearch 的准实时同步二是想用 RabbitMQ 把写库和建索引解耦避免双写不一致。装不好会怎样轻则索引建不出来重则消息堆积、数据对不上排查起来像开黑匣子。下面按我实际搭过一遍的顺序从版本选型一路写到验证跑通。2. 环境准备与版本选型三个组件的版本必须对齐2.1 版本对照表与选型理由这套组合的坑八成出在版本上。Elasticsearch 8.x 默认开启安全认证客户端必须带账号密码或证书Spring Boot 3.x 才正式支持 Elasticsearch 8 的 Java ClientRabbitMQ 的 Spring AMQP 版本又跟 Spring Boot 绑定。我一般先定 Spring Boot再反推另外两个。组件推荐版本说明JDK17Spring Boot 3.x 最低要求 17Spring Boot3.2.x内置 Spring Data Elasticsearch 5.2.x对应 ES 8.3Elasticsearch8.3.3与 Spring Data ES 5.2 官方兼容RabbitMQ3.12.x管理插件与 Spring AMQP 3.1 匹配MySQL8.0.x开启 binlog 便于后续扩展Maven3.9.x编译插件版本别太老选 8.3.3 而不是最新的 8.x是因为 Spring Data Elasticsearch 5.2 的兼容矩阵明确写到 8.3再往上客户端 API 有变动ElasticsearchOperations的部分方法签名会变。熟手可以追新但第一次搭环境稳比新重要。2.2 依赖清单与 pom 配置依赖不要贪多spring-boot-starter-data-elasticsearch已经带了 Java Client再手动引elasticsearch-java容易版本打架。RabbitMQ 用 starter 即可。dependencies !-- Web 层提供同步触发接口 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-web/artifactId /dependency !-- Elasticsearch 客户端版本由 Spring Boot 父工程管理 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-elasticsearch/artifactId /dependency !-- RabbitMQ用于解耦写库与建索引 -- dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-amqp/artifactId /dependency !-- MySQL 驱动与 JPA -- dependency groupIdcom.mysql/groupId artifactIdmysql-connector-j/artifactId scoperuntime/scope /dependency dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-jpa/artifactId /dependency /dependencies这里的关键点是不要给spring-boot-starter-data-elasticsearch写死版本号交给父工程spring-boot-starter-parent统一管理否则很容易出现客户端 8.5、服务端 8.3 的错配。MySQL 驱动从 8.0 起包名是com.mysql:mysql-connector-j老教程里的mysql:mysql-connector-java已经废弃照抄会拉不到包。2.3 本地起 Elasticsearch 8.3 与 RabbitMQ我用 Docker 起中间件省得污染本机。Elasticsearch 8.3 首次启动会自动生成elastic用户密码记得从日志里捞出来。# 启动 Elasticsearch 8.3.3单节点模式关闭不必要的内存占用 docker run -d --name es833 \ -p 9200:9200 -p 9300:9300 \ -e discovery.typesingle-node \ -e ES_JAVA_OPTS-Xms512m -Xmx512m \ -e xpack.security.enabledtrue \ docker.elastic.co/elasticsearch/elasticsearch:8.3.3 # 从日志里取初始密码形如 Password for the elastic user is: xxxx docker logs es833 21 | grep -A2 elastic user # 启动 RabbitMQ带管理插件方便看队列 docker run -d --name rabbit312 \ -p 5672:5672 -p 15672:15672 \ rabbitmq:3.12-managementdiscovery.typesingle-node是单机开发必须的不加会尝试组集群然后卡在选举。xpack.security.enabledtrue保持默认开启别图省事关掉——关掉之后客户端配置里不写认证能跑但换到测试环境开了安全认证就全挂这种“本地能跑线上不能跑”的玄学问题多半是这里埋的。RabbitMQ 默认账号密码都是guest只允许本机访问容器映射端口后从宿主机连没问题。3. 配置连接application.yml 里三个容易写错的参数3.1 Elasticsearch 连接配置Spring Boot 3.x 的 ES 配置前缀是spring.elasticsearch不是老版本的spring.data.elasticsearch。这一点写错配置根本不生效客户端会用默认localhost:9200且不带认证然后报 401。spring: elasticsearch: uris: http://localhost:9200 username: elastic password: 你的初始密码 connection-timeout: 5s socket-timeout: 30s data: elasticsearch: repositories: enabled: trueuris写http://前缀8.x 客户端支持 HTTPS但本地开发用 HTTP 更省事。connection-timeout设 5 秒避免服务没起来时启动线程卡太久socket-timeout给 30 秒批量建索引时别设太短。repositories.enabled打开后可以用ElasticsearchRepository接口简单 CRUD 不用手写。3.2 RabbitMQ 与 MySQL 配置RabbitMQ 的配置重点是确认机制和并发消费者数量MySQL 这边注意时区和批量抓取。spring: rabbitmq: host: localhost port: 5672 username: guest password: guest publisher-confirm-type: correlated # 开启发送确认 publisher-returns: true # 开启路由失败回调 listener: simple: acknowledge-mode: manual # 手动 ack防止消息丢失 prefetch: 10 # 每个消费者预取 10 条 concurrency: 2 max-concurrency: 4 datasource: url: jdbc:mysql://localhost:3306/demo?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8 username: root password: 你的密码 driver-class-name: com.mysql.cj.jdbc.Driverpublisher-confirm-type: correlated配合publisher-returns: true消息发到交换机但路由不到队列时能拿到回调不然消息静默丢失排查时毫无线索。acknowledge-mode: manual是同步场景的后悔药——消费者处理完建索引再手动 ack中途异常消息还能重回队列。prefetch: 10控制单个消费者未确认消息上限太大容易内存涨太小吞吐上不去10 到 20 是常见取值。3.3 索引映射与实体定义MySQL 表结构映射到 ES 时字段类型要显式声明别指望动态映射。日期字段尤其容易翻车MySQL 的datetime到 ES 里如果被识别成text范围查询直接失效。Document(indexName product) public class ProductDoc { Id private Long id; Field(type FieldType.Text, analyzer standard) private String name; Field(type FieldType.Keyword) private String category; Field(type FieldType.Double) private Double price; Field(type FieldType.Date, format DateFormat.date_hour_minute_second) private LocalDateTime updateTime; // getter/setter 省略 }Document(indexName product)对应索引名启动时如果配置了自动建索引会按注解生成 mapping。category用Keyword而不是Text因为它是精确过滤字段用Text会被分词聚合和 term 查询都对不上。updateTime显式指定DateFormat避免默认格式与写入值不匹配导致mapper_parsing_exception。4. 同步链路搭建MySQL 变更经 RabbitMQ 落到 Elasticsearch4.1 交换机、队列与绑定声明同步链路的核心是一个 topic 交换机加一个持久化队列。交换机按业务类型路由队列持久化保证 RabbitMQ 重启后消息不丢。Configuration public class RabbitConfig { public static final String EXCHANGE sync.exchange; public static final String QUEUE sync.product.queue; public static final String ROUTING_KEY product.#; Bean public TopicExchange syncExchange() { // durabletrue交换机持久化 return new TopicExchange(EXCHANGE, true, false); } Bean public Queue syncQueue() { // 持久化队列避免 broker 重启后队列消失 return QueueBuilder.durable(QUEUE).build(); } Bean public Binding syncBinding() { return BindingBuilder.bind(syncQueue()).to(syncExchange()).with(ROUTING_KEY); } }TopicExchange的第三个参数autoDelete设 false别让队列在无消费者时被自动删除。ROUTING_KEY用product.#通配后续加product.update、product.delete都能命中同一个队列扩展时不用改绑定。队列用QueueBuilder.durable而不是new Queue语义更清楚也方便后续加死信参数。4.2 写库后发消息事务边界要卡准最容易出错的地方是“库写成功了消息没发出去”或者“消息发了库回滚了”。常见做法是把发消息放在事务提交之后用TransactionSynchronization回调。Service public class ProductService { private final ProductRepository repository; private final RabbitTemplate rabbitTemplate; public ProductService(ProductRepository repository, RabbitTemplate rabbitTemplate) { this.repository repository; this.rabbitTemplate rabbitTemplate; } Transactional public void saveAndSync(Product product) { repository.save(product); // 事务提交后再发消息避免回滚后消息已投递 TransactionSynchronizationManager.registerSynchronization( new TransactionSynchronization() { Override public void afterCommit() { rabbitTemplate.convertAndSend( RabbitConfig.EXCHANGE, product.update, product.getId() ); } } ); } }afterCommit保证只有数据库事务真正提交才发消息这是双写一致性的第一道闸。消息体只传主键id不传整个对象——对象可能很大而且消费端应该以数据库为准重新查避免消息里的旧数据覆盖新数据。convertAndSend默认用 JDK 序列化生产环境建议配Jackson2JsonMessageConverter否则跨语言消费方解不了。4.3 消费端建索引手动 ack 与幂等处理消费端拿到 id 后回查 MySQL再写入 ES。手动 ack 配合幂等写入重复消息不会造成脏数据。Component public class ProductSyncConsumer { private final ProductRepository repository; private final ElasticsearchOperations esOperations; RabbitListener(queues RabbitConfig.QUEUE) public void onMessage(Long productId, Message message, Channel channel) throws IOException { long tag message.getMessageProperties().getDeliveryTag(); try { Product product repository.findById(productId).orElse(null); if (product ! null) { // save 是幂等的同一 id 重复写会覆盖 esOperations.save(toDoc(product)); } channel.basicAck(tag, false); } catch (Exception e) { // 处理失败拒绝并重回队列配合重试上限防止死循环 channel.basicNack(tag, false, true); } } }basicAck的第二个参数multiple设 false逐条确认批量确认在异常时容易误确认。basicNack第三个参数requeuetrue会把消息放回队列但要注意如果异常是数据本身的问题无限重回会形成死循环生产上一般配死信队列或重试次数上限。esOperations.save按Id覆盖写入天然幂等重复消费不会产生重复文档。5. 避坑与排查五个真实踩过的坑5.1 启动报 401 Unauthorized现象应用启动正常第一次调 ES 接口报ResponseException: 401 Unauthorized。原因spring.elasticsearch.username/password没配或者密码里含特殊字符没加引号。解决从容器日志确认初始密码YAML 里用单引号包住密码如果改过密码用curl -u elastic:密码 http://localhost:9200先验证凭据本身可用。5.2 索引字段类型与预期不符现象price字段做范围查询查不到数据。原因索引是动态映射生成的price被识别成text。解决删掉索引DELETE /product确认实体上Field(type FieldType.Double)生效后重启重建或者用PUT /product/_mapping手动补 mapping。动态映射只适合原型正式环境一律显式声明。5.3 消息发出但队列没收到现象convertAndSend不报错管理后台队列消息数为 0。原因routing key 与绑定不匹配或者交换机名写错。解决打开publisher-returns后注册ReturnsCallback打印退回消息在 RabbitMQ 管理页的 Exchanges 里看绑定关系确认 routing key 能被product.#命中。5.4 消费端反复重试同一条消息现象日志里同一条消息被消费多次ES 数据没更新。原因basicNack的requeuetrue遇到不可恢复异常比如 id 在库里不存在形成死循环。解决区分可重试与不可重试异常数据不存在直接 ack 丢弃并记日志或者给队列配x-dead-letter-exchange超过重试次数进死信队列人工处理。5.5 时间字段写入报 mapper_parsing_exception现象写入 ES 时报日期格式解析失败。原因LocalDateTime序列化格式与 mapping 里声明的DateFormat不一致。解决统一用yyyy-MM-ddTHH:mm:ss格式实体上Field(type FieldType.Date, format DateFormat.date_hour_minute_second)并在 Jackson 配置里注册JavaTimeModule别让默认序列化输出带时区偏移的格式。6. 装完怎么验证三步确认链路真的通了环境装完不等于链路通我一般按“中间件 → 单组件 → 全链路”三步验证每步都有明确的观察点。第一步确认两个中间件本身可用。ES 用curl查集群健康RabbitMQ 看管理页队列是否声明成功# ES 返回 status 为 yellow 或 green 即正常单节点 yellow 是副本未分配可接受 curl -u elastic:你的密码 http://localhost:9200/_cluster/health?pretty # 确认索引 mapping 里 price 是 double、updateTime 是 date curl -u elastic:你的密码 http://localhost:9200/product/_mapping?pretty第二步单独验证 ES 写入。调一次业务接口触发saveAndSync然后直接查 ES 文档# 按 id 查文档能看到 _source 里字段齐全即写入成功 curl -u elastic:你的密码 http://localhost:9200/product/_doc/1?pretty第三步验证消息链路。在 RabbitMQ 管理页http://localhost:15672看sync.product.queue的 Ready 和 Unacked 计数调用接口后 Ready 短暂上升再归零说明消费正常如果 Ready 持续上涨说明消费者没起来或 ack 卡住。再故意把 ES 停掉调一次接口观察消息是否留在队列里、恢复后是否被重新消费——这一步能验证手动 ack 和重回队列是否真的生效。验证点正常表现异常时先看哪ES 集群健康yellow/green容器日志、端口占用索引 mapping字段类型与实体一致是否动态映射抢先建了索引队列 Ready 计数调用后短暂上升后归零消费者是否注册、ack 模式文档内容_source字段完整消费端回查 SQL 与转换逻辑这套验证我吃过亏有一次只看了接口返回 200 就以为通了结果 ES 里一条数据没有原因是消费者所在的包没被扫描到RabbitListener根本没生效。从那以后我每次搭完这套环境都强制走一遍“停 ES 看消息堆积、恢复看消费归零”的动作比看日志靠谱得多。希望帮到你。本文还有配套的精品资源点击获取
