springboot项目elasticsearch6.x transportClient切换为restHighLevelClient（包含restClient）

背景：

项目中使用spring-data-elasticsearch做的集成（用的transportClient客户端9300端口）
由于项目需要迁移到欧洲部署，es使用阿里云提供的加密版，导致9300端口无法连接，故改用9200端口（http）

流程：

1.集成rest客户端，加入配置，在测试类中测通；
2.改造涉及到es的所有代码
3.删除涉及到transportClient的代码、配置、依赖
4.改造完成

添加依赖

<!-- Java Low Level REST Client --><dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-client</artifactId><version>6.3.2</version></dependency><!-- Java High Level REST Client --><dependency><groupId>org.elasticsearch.client</groupId><artifactId>elasticsearch-rest-high-level-client</artifactId><version>6.3.2</version></dependency>

yml配置

elasticSearch:host: 10.101.99.189port: 9200username: elasticpassword: 123456

ES restHighLevelClient配置

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;/*** es高级客户端配置** @author GaoJinShan*/
@Configuration
public class ElasticsearchConfiguration implements FactoryBean<RestHighLevelClient>, InitializingBean, DisposableBean {private static final Logger LOGGER = LoggerFactory.getLogger(ElasticsearchConfiguration.class);@Value("${elasticSearch.host}")private String host;@Value("${elasticSearch.port}")private int port;@Value("${elasticSearch.username}")private String username;@Value("${elasticSearch.password}")private String password;private RestHighLevelClient restHighLevelClient;@Overridepublic void destroy() throws Exception {try {LOGGER.info("Closing elasticSearch client");if (restHighLevelClient != null) {restHighLevelClient.close();}} catch (final Exception e) {LOGGER.error("Error closing ElasticSearch client: ", e);}}@Overridepublic RestHighLevelClient getObject() throws Exception {return restHighLevelClient;}@Overridepublic Class<RestHighLevelClient> getObjectType() {return RestHighLevelClient.class;}@Overridepublic boolean isSingleton() {return false;}@Overridepublic void afterPropertiesSet() throws Exception {buildClient();}protected void buildClient() {final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));RestClientBuilder builder = RestClient.builder(new HttpHost(host, port)).setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {@Overridepublic HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);}});restHighLevelClient = new RestHighLevelClient(builder);}}

import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.TimeUnit;@Service
public class ElasticSearchUtil {private static final Logger LOGGER = LoggerFactory.getLogger(ElasticSearchUtil.class);@Autowiredprivate RestHighLevelClient restHighLevelClient;private BulkProcessor bulkProcessor;@PostConstructpublic void init() {BulkProcessor.Listener listener = new BulkProcessor.Listener() {@Overridepublic void beforeBulk(long executionId, BulkRequest request) {int numberOfActions = request.numberOfActions();LOGGER.info("Executing bulk [{}] with {} requests", executionId, numberOfActions);}@Overridepublic void afterBulk(long executionId, BulkRequest request, BulkResponse response) {if (response.hasFailures()) {LOGGER.error("Bulk [{}] executed with failures,response = {}", executionId, response.buildFailureMessage());} else {LOGGER.info("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());}BulkItemResponse[] responses = response.getItems();}@Overridepublic void afterBulk(long executionId, BulkRequest request, Throwable failure) {LOGGER.error("Failed to execute bulk", failure);}};BulkProcessor bulkProcessor = BulkProcessor.builder(restHighLevelClient::bulkAsync, listener)// 1000条数据请求执行一次bulk.setBulkActions(1000)// 5mb的数据刷新一次bulk.setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB))// 并发请求数量, 0不并发, 1并发允许执行.setConcurrentRequests(0)// 固定1s必须刷新一次.setFlushInterval(TimeValue.timeValueSeconds(1L))// 重试5次，间隔1s.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 5)).build();this.bulkProcessor = bulkProcessor;}@PreDestroypublic void destroy() {try {bulkProcessor.awaitClose(30, TimeUnit.SECONDS);} catch (InterruptedException e) {LOGGER.error("Failed to close bulkProcessor", e);}LOGGER.info("bulkProcessor closed!");}/*** 修改** @param request*/public void update(UpdateRequest request) {this.bulkProcessor.add(request);}/*** 新增** @param request*/public void insert(IndexRequest request) {this.bulkProcessor.add(request);}

使用示例

查询：

    @Autowiredprivate RestHighLevelClient restHighLevelClient;@Testpublic void contextLoads() throws IOException, InvocationTargetException, IllegalAccessException {//"admin_log"是es中的索引//SearchRequest是包装查询请求SearchRequest searchRequest = new SearchRequest("admin_log");//sourceBuilder 是用来构建查询条件SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();//添加查询条件就和transportClient中的一样了sourceBuilder.query(QueryBuilders.boolQuery().must(QueryBuilders.termsQuery("appName.keyword", "aaaaaa-mng")));searchRequest.source(sourceBuilder);SearchResponse searchResponse = restHighLevelClient.search(searchRequest);SearchHit[] searchHits = searchResponse.getHits().getHits();//AdminLog是我定义的实体类ArrayList<AdminLog> adminLogs = new ArrayList<>();//用来转换存储查询到的for (SearchHit hit : searchHits) {Map<String, Object> sourceAsMap = hit.getSourceAsMap();AdminLog adminLog = new AdminLog();//map赋值给实体类对象（key和对象属性务必一致）BeanUtils.populate(adminLog, sourceAsMap);adminLogs.add(adminLog);}}

聚合示例：

    @Autowiredprivate RestHighLevelClient restHighLevelClient;@Testpublic void aggsSearch() {SearchRequest searchRequest = new SearchRequest("admin_log");SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();searchRequest.source(sourceBuilder);sourceBuilder.aggregation(AggregationBuilders.terms("xxx_NAME").field("appName.keyword").size(10000));try {SearchResponse searchResponse = restHighLevelClient.search(searchRequest);Terms aggregationResult = searchResponse.getAggregations().get("xxx_NAME");List<? extends Terms.Bucket> buckets = aggregationResult.getBuckets();for (Terms.Bucket tb:buckets) {System.out.println(tb.getKey());}} catch (IOException e) {e.printStackTrace();}}

restLowLevelClient的使用

    @Autowiredprivate RestHighLevelClient restHighLevelClient;@Testpublic void restClientTest() throws IOException {Map<String, String> params = Collections.emptyMap();String queryString = "{\"sort\": [\n" +"  {\n" +"    \"esCreateTime\": {\n" +"      \"order\": \"desc\"\n" +"    }\n" +"  }\n" +"]}";HttpEntity entity = new NStringEntity(queryString, ContentType.APPLICATION_JSON);//rest高级客户端是基于低级客户端的，利用getLowLevelClient()函数从高级客户端中获取就ok了Response response = restHighLevelClient.getLowLevelClient().performRequest("GET", "/admin_log/_search", params, entity);}

transportClient和restHighLevelClient改造前后对比

注：本文restHighLevelClient配置部分参考自https://my.oschina.net/u/3795437/blog/3022874

springboot项目elasticsearch6.x transportClient切换为restHighLevelClient（包含restClient）相关推荐

springboot集成elasticsearch6.8.23设置密码xpack连接，及遇到的None of the configured nodes are available
遇到一个 None of the configured nodes are available 的坑一.背景: 因现网扫描出来几个漏洞,目前版本使用的springboot2.1.17.RELEASE ...
Docker 部署SpringBoot项目不香吗？
点击上方"方志朋",选择"设为星标" 回复"666"获取新整理的面试文章作者:流星007 链接:toutiao.com/i68433912 ...
Kotlin开发springboot项目（一）
Kotlin开发springboot项目(一) Kotlin语言与Xtend语言有很多相似之处为什么会存在这么多JVM语言? 现存的语言提供了太过受限制的功能,要不就是功能太过繁杂,导致语言的臃肿和 ...
使用IDEA在SpringBoot项目中连接数据库
使用IDEA在SpringBoot项目中连接数据库文章目录使用IDEA在SpringBoot项目中连接数据库前言连接数据库正常操作数据库移除数据库连接前言每次我在 IDEA 中用 Sp ...
docker部署springboot_创建SpringBoot项目部署到docker全过程
目录 docker 介绍安装 docker Ubuntu 安装 docker CentOS 安装 docker 通过脚本安装拉取 java 环境创建 springboot 项目打包 sprin ...
docker 搭建指定版本的cas_Docker搭建-生成SpringBoot项目脚手架-各版本
start.spring.io是用来生成SpringBoot项目脚手架的,因为网络的原因,某些情况下无法正常访问.有时候我们也需要一些历史的版本.基于这两个原因,我们可以将它的代码拉下来,在本地打包, ...
IDEA配置Docker一键部署SpringBoot项目（企业级做法）
Docker一键部署SpringBoot项目前言一.开启Linux的Docker远程访问 1.1 安装vim工具: 1.2 修改`daemon.json`的配置: 1.3 加入配置: 1.4 重新 ...
全新SpringBoot整合Elasticsearch6.xxx搜索引擎实战
SpringBoot整合Elasticsearch6.xxx搜索引擎实战一.导入相关依赖包二.配置application.properties文件三.ES存储实体类及注解 1.测试创建索引四. ...
SpringBoot 项目构建 Docker 镜像调优实践
点击上方"Java知音",选择"置顶公众号" 技术文章第一时间送达! 作者:超级小豆丁 http://www.mydlq.club/article/16/ PS ...
从零开始带你部署springboot项目到ubuntu服务器05
从零开始带你部署springboot项目到ubuntu服务器 1 重装系统 2 更换源 3 部署 1 安装jdk 2 安装tomcat 3 安装mysql 1) 查看本地mysql版本 2)在服务器上 ...

springboot项目elasticsearch6.x transportClient切换为restHighLevelClient（包含restClient）

背景：

添加依赖

yml配置

ES restHighLevelClient配置

使用示例

查询：

聚合示例：

restLowLevelClient的使用

transportClient和restHighLevelClient改造前后对比

springboot项目elasticsearch6.x transportClient切换为restHighLevelClient（包含restClient）相关推荐

最新文章

热门文章