ES在SpringBoot集成使用这个话题我其实拖了小半年才认真搞。之前项目里搜索一直用的MySQL的LIKE查询数据量过了百万之后性能惨不忍睹没办法只能把ES请进来。真正做的时候才发现网上教程七零八落有的讲ES客户端用得还是十年前那套TransportClient有的上来就直接甩一个高深莫测的聚合查询。这篇东西不玩虚的我就按自己从头到尾集成、迁移、优化的实际流程来写把SpringBoot下ES的版本选择、依赖配置、索引操作、异步写入、MySQL同步以及各种坑都挑明了说适合那些正准备做ES集成、或者已经被集成问题折磨得头疼的朋友直接抄作业。1. 环境准备与基础集成1.1 版本选型SpringBoot 3.x 就必须配 ES 8.x版本问题永远排在第一位因为ES和SpringBoot的版本兼容性真的能折腾死人。我最早在项目里用的是SpringBoot 2.3.12当时配套的ES客户端是7.9.3这套组合其实是稳定的。但后来项目升级到SpringBoot 3.0以后发现老的RestHighLevelClient被ES官方弃用了SpringData Elasticsearch也随之换了底层实现如果你还在网上搜到一堆基于HighLevelClient的代码在SpringBoot 3.x里是跑不起来的。我最终选择了SpringBoot 3.2.5配合Spring Data Elasticsearch 5.2.5对应的ES服务端版本是8.12.2。这里有个容易踩的坑SpringData Elasticsearch的版本号跟ES服务端版本不是强绑定的但必须保证大版本一致比如SpringData ES 5.x对应ES 8.x。还有一点ES 8.x默认开启了安全认证如果你本地测试不想用安全认证需要在启动ES时把xpack.security.enabled改成false不然后面客户端连接各种401报错。版本对应关系建议拿去参考这是我实测下来比较稳的搭配SpringBoot版本SpringData ES版本ES服务端版本备注2.7.x4.4.x7.17.x稳定网上资料多3.0.x5.0.x8.8.x新项目推荐3.2.x5.2.x8.11.x当前我用这套稳1.2 本地安装ESDocker方式最省心本地开发环境装ES我强烈推荐Docker一条命令就能把ES跑起来不用纠结JDK版本、环境变量之类的破事。我习惯把ES的data和plugins目录挂载到宿主机这样容器删了数据还在插件也不会丢。我用的Docker命令如下注意ES 8.x默认要传一些JVM参数docker run -d --name es \ -p 9200:9200 -p 9300:9300 \ -e discovery.typesingle-node \ -e xpack.security.enabledfalse \ -e ES_JAVA_OPTS-Xms512m -Xmx512m \ -v /data/es/data:/usr/share/elasticsearch/data \ -v /data/es/plugins:/usr/share/elasticsearch/plugins \ docker.elastic.co/elasticsearch/elasticsearch:8.12.2启动之后访问http://localhost:9200返回一个带version信息的JSON就说明成功了。这里提醒一句如果没有挂载data目录容器删了就什么都没了别问我怎么知道的。1.3 引入依赖和配置文件依赖我用的是spring-boot-starter-data-elasticsearch它内部已经封装好了ES客户端你不需要再额外引入elasticsearch-rest-client之类的依赖版本也有SpringBoot的依赖管理统一管着不容易出冲突。pom.xml里加这段dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-data-elasticsearch/artifactId /dependency然后是配置文件。在application.yml里SpringData ES的连接配置长这样spring: elasticsearch: uris: http://localhost:9200 connection-timeout: 5s socket-timeout: 60s注意这里只有spring.elasticsearch.uris不像老版本还区分host和port。如果你前端有nginx层代理也可以配置多个uri用逗号隔开但单个ES节点就没必要了。还有一个容易被忽略的点socket-timeout默认是10秒如果查询数据量大很容易超时建议调整到30秒到60秒。2. 索引操作与CRUD实战2.1 用Java代码创建索引和Mapping有了连接之后第一个任务就是创建索引。很多人习惯直接在Kibana的Dev Tools里写DSL但当项目需要自动化部署时用代码创建索引就非常有必要。SpringData ES里操作索引有三种姿势ElasticsearchOperations、ElasticsearchRestTemplate以及客户端原生API。这里我最推荐通过ElasticsearchOperations的indexOps()方法来搞因为它的API设计对Java开发者最友好。先定义一个索引对应的实体类比如用来存商品信息的Document(indexName product, createIndex false) public class Product { Id private String id; Field(type FieldType.Text, analyzer ik_max_word, searchAnalyzer ik_smart) private String name; Field(type FieldType.Keyword) private String category; Field(type FieldType.Double) private BigDecimal price; Field(type FieldType.Date, format DateFormat.date_time) private LocalDateTime createTime; }这里有个细节Field注解的analyzer指定的是索引分词器searchAnalyzer指定的是搜索分词器。我用的ik_max_word需要提前在ES里安装IK分词插件不然创建索引时会直接报错。安装IK插件可以下载对应版本的zip包放到plugins目录下解压然后重启ES容器。然后创建索引的代码public void createIndex() { IndexOperations indexOps elasticsearchOperations.indexOps(Product.class); if (!indexOps.exists()) { indexOps.create(); indexOps.putMapping(indexOps.createMapping()); } }这样就能把Product类的Mapping自动同步到ES里。但自动生成的Mapping有时候不能满足业务需求比如你想给某个字段设置norms为false或者doc_values为false就需要手动写Mapping的JSON然后通过indexOps.putMapping传进去。2.2 文档写入的三种姿势写入文档是最常用的操作SpringData ES提供了三种方式我按推荐程度排个序。第一种是直接用Repository接口继承了ElasticsearchRepository之后调用save方法就行public interface ProductRepository extends ElasticsearchRepositoryProduct, String { } productRepository.save(product);这种方式最简单适合单条写入或者低频业务场景。第二种是批量写入调用saveAll方法ListProduct products new ArrayList(); productRepository.saveAll(products);saveAll底层走的是Bulk API比单条循环调用save要高效得多实测在1000条数据场景下性能提升了至少6倍。第三种是直接使用ElasticsearchOperations的bulkIndex方法可以更灵活地控制批量行为ListIndexQuery queries products.stream() .map(p - new IndexQuery.builder() .withId(p.getId()) .withObject(p) .build()) .toList(); elasticsearchOperations.bulkIndex(queries, Product.class);不过到这里你会发现如果你的写入频率很高比如每秒上千条那无论哪种方式都还是有瓶颈这就是后面要讲的异步写入和批量调优的问题了。2.3 查询实战关键词过滤聚合ES的查询往往不是单一条件而是组合查询。SpringData ES对查询的封装有两种路子一种是方法名自动派生比如findByNameContaining另一种是用NativeQuery或者CriteriaQuery构建复杂查询。遇到复杂业务我一般直接用NativeQuery它的DSL可读性和灵活性最好。举个例子我现在要搜索名字包含手机的商品、品牌是华为、价格在3000到6000之间并且按价格升序排列NativeQuery nativeQuery new NativeQueryBuilder() .withQuery(q - q .bool(b - b .must(m - m.match(mm - mm.field(name).query(手机))) .filter(f - f.term(t - t.field(category).value(华为))) .filter(f - f.range(r - r .field(price) .gte(JsonData.of(3000)) .lte(JsonData.of(6000)))))) .withSort(s - s.field(f - f.field(price).order(SortOrder.Asc))) .build(); SearchHitsProduct hits elasticsearchOperations.search(nativeQuery, Product.class);这里的关键点是bool查询里must和filter的区别must会影响评分filter只做过滤不影响评分。对性能敏感的场景尽量把过滤条件塞到filter里ES会缓存过滤上下文的结果加快后续查询速度。聚合查询的话可以直接用aggregation()方法比如按分类统计商品数量NativeQuery aggQuery new NativeQueryBuilder() .withAggregation(category_count, a - a .terms(t - t.field(category))) .build();然后从SearchHits里取聚合结果这部分代码比较长但基本模式就是通过AggregationContainerOf去解析。2.4 路由字段routing的正确玩法如果你对ES有了解应该知道ES分片是按_routing值哈希取模来路由到具体分片的。默认使用_id作为路由值这样会产生一个问题当你查询某个特定用户的数据时请求会被广播到所有分片再汇总数据量大了之后就特别慢。这时候就需要给文档加一个路由字段比如用户ID保证同一个用户的文档都落在同一个分片上查询时指定同样的路由值ES就能直接定位到一个分片查询效率翻倍。在SpringData ES中实体类里可以用Routing注解或者直接在写入时设置Document(indexName order) public class Order { Id private String id; Routing private String userId; }查询的时候带上路由参数NativeQuery query new NativeQueryBuilder() .withQuery(q - q.matchAll(m - m)) .withRoute(userId) .build();这里有个需要注意的坑如果写入时指定了routing而查询时忘了指定那查询就会落到所有分片结果可能查到但性能会倒退。所以routing的使用一定要写入和查询同时兼顾最好把routing值统一存到一个上下文里比如放到savedThreadLocal或者传入查询封装对象中。3. 异步写入与性能优化3.1 同步写入到底慢在哪我刚集成ES那会儿同步写入的性能惨不忍睹。一条订单数据写一次ES业务响应时间直接加了30毫秒高频写入场景下整个接口都卡到崩溃。后来分析才知道单条写入的链路是业务线程发起HTTP请求到ESES写完后返回响应这个过程中业务线程一直在阻塞等待。如果ES写入放大或者网络抖动那延迟就更明显了。这就像你去银行柜台办一笔业务每笔都得排队等叫号效率肯定低。异步写入的思路就相当于你填好单子扔进大厅的投放箱柜员自己慢慢处理你填完直接走人整个过程不占你时间。3.2 SpringBoot的Async异步写入实现SpringBoot做异步写入最简单的方式是配合Async注解。首先在启动类或者配置类上开启异步支持Configuration EnableAsync public class AsyncConfig { Bean(name esWriteExecutor) public Executor esWriteExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); executor.setCorePoolSize(4); executor.setMaxPoolSize(8); executor.setQueueCapacity(2000); executor.setThreadNamePrefix(es-write-); executor.setRejectedExecutionHandler(new CallerRunsPolicy()); executor.initialize(); return executor; } }然后在写入方法上加上Async(esWriteExecutor)Async(esWriteExecutor) public void asyncWriteProduct(Product product) { productRepository.save(product); }这里有一个非常隐蔽的坑Async注解在自调用的时候是失效的。比如你在这个类内部调this.asyncWriteProduct()Spring的代理根本不会生效方法会在当前线程同步执行。要解决这个问题必须通过注入的Bean来调用或者把异步方法放到独立的Service里。线程池参数也不要乱配。核心线程数设成4不代表性能就最好要根据你的业务QPS和ES吞吐量动态调整。我先说一下我项目的参数并发写入峰值2000条/秒核心线程4最大线程8队列2000配合CallerRunsPolicy拒绝策略。这种配置在高峰期能让写入压力直接透传到ES如果ES吃不消就会把压力回传到调用方而不是无限囤积在内存队列里这对防止内存溢出很关键。3.3 Bulk批量写入的参数调优异步单条写入虽然解决了线程阻塞问题但依然没有把ES的Bulk API发挥到极致。批量写入才是ES高性能写入的王道。我后来从单条异步替换成批量异步性能又翻了几倍。批量写入的常见做法是攒一批数据再发送。实现上可以自己维护一个BlockingQueue定时消费批量提交。我常用的方案是用一个ScheduledExecutorService每500毫秒拉取一次队列中的数据凑满1000条就返回一次批量写入。这样兼顾了实时性最长500毫秒延迟和批量性单次Bulk写入上千条。伪代码是这样Component public class BulkProductWriter { private final BlockingQueueProduct queue new LinkedBlockingQueue(10000); Scheduled(fixedDelay 500) public void flush() { if (queue.isEmpty()) return; ListProduct batch new ArrayList(); queue.drainTo(batch, 1000); if (!batch.isEmpty()) { productRepository.saveAll(batch); } } }这里的queue.size和batch.size需要根据你服务的JVM内存和ES节点性能来调节。我试过批量设置到2000条ES节点8G内存CPU占比提升到70%然后开始报RejectedExecutionException说明ES节点吞吐到顶了。建议控制在1000条左右每个分片100~200条效果比较稳定。还有一点ES写入时如果遇到429错误表示ES正在限流或压力过大这时候不要傻傻重试应该退避一下或者降低写入速率。我在代码里就加了简单的重试逻辑如果返回429指数退避等待100毫秒再重试最多重试3次。4. 数据同步MySQL如何与ES保持同步4.1 同步方案横向对比我们项目中ES里的数据主要来源于MySQL业务写MySQL后需要同步到ES供搜索。同步方案选型是个大话题我直接给出对比表格这个方案也是按我们项目的真实需求评估得来的方案实时性侵入性运维成本适用场景Logstash分钟级低中需要维护管道全量定时增量对实时性要求不高的场景Canal秒级低高需要额外部署Canal高并发、近实时同步能解析MySQL binlog自研Java定时任务秒级-分钟级高需要写业务代码低数据量不大逻辑简单团队不想引外部组件我们当时的业务对实时性要求不算苛刻允许几十秒级延迟所以先用Logstash把全量迁移做了再配合一个定时任务兜底。如果你要做近实时的搜索比如电商订单状态变更那Canal是更合适的选择但它的部署复杂度也确实高。4.2 Logstash实战从MySQL增量同步到ESLogstash配置的核心在于jdbc_input插件。先写一个mysql_to_es.confinput { jdbc { jdbc_driver_library /path/to/mysql-connector-java-8.0.33.jar jdbc_driver_class com.mysql.cj.jdbc.Driver jdbc_connection_string jdbc:mysql://127.0.0.1:3306/mydb?useUnicodetruecharacterEncodingUTF-8serverTimezoneAsia/Shanghai jdbc_user root jdbc_password 123456 jdbc_paging_enabled true jdbc_page_size 5000 statement SELECT id, name, category, price, update_time FROM product WHERE update_time :sql_last_value schedule */10 * * * * * tracking_column update_time use_column_value true tracking_column_type timestamp } } filter { # 对同步过来的字段做清洗 } output { elasticsearch { hosts [http://localhost:9200] index product document_id %{id} document_type _doc } }注意几个关键点tracking_column必须随着数据表里的更新时间变化否则增量同步永远不会生效。sql_last_value是Logstash内置的游标它会记录上次同步的最后的update_time值可以持久化到磁盘以防止重启丢游标。Logstash的输出不太方便处理MySQL中的删除操作。如果业务有删除需求你需要在应用里标记逻辑删除同步时把ES文档改成delete状态或者另写一个清理脚本来处理。4.3 同步延迟与数据一致性兜底方案Logstash的定时同步总会有延迟在延迟窗口内如果用户搜索到旧数据体验会受影响。我们的做法是双保险Logstash每10秒一同步同时自己写一个定时任务每分钟扫一次MySQL最近更新的数据触发一次接口级的增量刷新。相当于把同步延迟压到最坏情况不超过1分钟同时Logstash失败后定时任务还能兜底。要注意的是千万不要在同步逻辑里引入复杂的分布式事务。ES和MySQL本身就是不同的存储系统强一致不太现实做好最终一致性就行。具体实现上每次同步时带上update_time字段在ES查询端加一个排序保证最新数据优先展示。5. 常见问题排查与避坑实录5.1 NoNodeAvailableException的典型原因当你第一次启动SpringBoot项目连ES时报NoNodeAvailableException不用慌80%的情况是下面几个原因第一ES服务没有真正启动。本地启动后要先确认9200端口通不通可以curl一下。第二ES配置了安全认证而SpringBoot客户端没有配账号密码。ES 8.x默认开启了xpack.security.enabledtrue需要在配置里添加username和password两个字段。第三SpringData版本和ES服务端版本不匹配。比如服务端是ES 7.xSpringData ES 4.x却用ES 8.x的协议去通信连接就被拒绝了。这类问题排查思路就一条先用curl http://localhost:9200验证ES是否正常再对比版本号最后看客户端日志里的具体异常类型基本上能定位。5.2 深分页性能杀手fromsize处理不了大偏移量ES默认分页用from和size但如果你要翻到1000页以后性能会灾难性下降。因为ES需要从所有分片里把前面一堆文档都取出来再丢弃。我当时对接需求时发现深度翻页导致ES节点CPU直接飙到90%查询RT超过5秒。解决办法有两个scroll和search_after。scroll适用于导出场景一次性把结果集快照下来慢慢消费而search_after用在上页的排序值作为游标适合交互式翻页。SpringData ES里实现search_after的关键是要在排序字段里指定一个唯一的字段比如id然后在查询参数里带上上一页最后一个文档的sort值。举个代码例子NativeQuery query new NativeQueryBuilder() .withQuery(q - q.matchAll(m - m)) .withSort(s - s.field(f - f.field(id).order(SortOrder.Asc))) .withMaxResults(20) .build(); // 第一次查询后取最后一个文档的sort值 Object[] searchAfter hits.getSearchHit(lastIndex).getSortValues(); // 下一页查询带上 query.setSearchAfter(searchAfter);记住一点search_after翻页过程中不允许跳页只能一页一页往后翻这对大多数分页展示场景是可以接受的。5.3 内存配置失误导致ES启动失败ES节点默认的堆内存是1G如果你没设置数据量一大就会频繁GC。而如果设置的堆内存过大比如30G又可能导致系统内存不够。常见错误是启动容器时没设置ES_JAVA_OPTS导致ES启动一段时间后报OutOfMemoryError。我建议本地开发时固定为512M到1G生产环境建议按机器内存的50%设置且不超过32G。Docker启动时注意加这个参数-E ES_JAVA_OPTS-Xms8g -Xmx8g这里又有个细节Xms和Xmx建议设置成相等避免JVM动态扩容引发性能抖动。设置完后要同时配置内核参数vm.max_map_count否则ES启动会报max virtual memory areas vm.max_map_count [65530] is too low解决办法是执行sysctl -w vm.max_map_count2621445.4 Docker Compose把SpringBoot与ES一起编排最后实际操作中我更喜欢用Docker Compose把ES、SpringBoot应用和Logstash放在一起管理这样可以省去大量手动配置环境的时间。一个典型的docker-compose.yml片段如下version: 3.8 services: es: image: docker.elastic.co/elasticsearch/elasticsearch:8.12.2 container_name: es environment: - discovery.typesingle-node - xpack.security.enabledfalse - ES_JAVA_OPTS-Xms1g -Xmx1g ports: - 9200:9200 volumes: - es_data:/usr/share/elasticsearch/data logstash: image: docker.elastic.co/logstash/logstash:8.12.2 container_name: logstash volumes: - ./logstash/mysql_to_es.conf:/usr/share/logstash/pipeline/logstash.conf depends_on: - es app: build: . container_name: springboot-app ports: - 8080:8080 depends_on: - es - logstash volumes: es_data:如果Logstash要集成自定义插件比如自定义的过滤插件需要先在容器里安装后commit新镜像或者在启动时挂载插件目录。热词里提到的logstash集成自定义插件就是这么个思路。6. 写在最后做完这个ES集成任务我最想吐槽的就是网上资料版本太老动不动让人用RestHighLevelClient结果SpringBoot 3.x环境下编译都过不了。我个人建议如果团队能接受一定学习成本直接拥抱SpringData Elasticsearch和官方Java API Client少走一点弯路。如果你正在做MySQL到ES的同步最省事的是先上Logstash跑半个小时把全量数据导进去再配合业务侧定时任务补增量。等到后期数据量和查询复杂度上来再去考虑引入Canal做秒级同步也不迟。最后再分享一个小技巧所有ES的索引创建、Mapping更新这类操作尽量在项目启动时自动执行而不是等运维手工去Kibana里敲命令这样能杜绝环境不一致带来的诡异问题。