ElasticsearchJavaRESTClient初始化、添加索引及数据

上一篇：Elasticsearch基础与概念
下一篇：Elasticsearch Java REST Client 查询
所有代码均为参考官方文档：官方文档

原文说明

Java 高级 REST 客户端在 Java 低级 REST 客户端之上工作。它的主要目标是公开 API 特定的方法，这些方法接受请求对象作为参数并返回响应对象，以便由客户端自己处理请求编组和响应解组。

每个 API都可以同步或异步调用。同步方法返回一个响应对象，而名称以async后缀结尾的异步方法需要一个侦听器参数，一旦收到响应或错误，就会通知该侦听器参数（在低级客户端管理的线程池上）。

Java High Level REST Client 依赖于 Elasticsearch 核心项目。它接受与相同的请求参数TransportClient并返回相同的响应对象。

java API 文档：JAVA doc
低版本的：Java REST Client
高版本的：Elasticsearch Java API Client

注意安装版本要与客户端引入依赖保持一致，否则会出现版本问题
高版本的初始化方式与低版本的不一样，建议参考对应版本的文档
高版本的依赖会出现一部分jakarta.json加载不到，可引入下面依赖来解决

org.eclipse.parsson jakarta.json 1.0.0

Elasticsearch Maven

# 所需的最低 Java 版本是1.8# 所需依赖 org.elasticsearch.client elasticsearch-rest-high-level-client 6.7.2 org.elasticsearch elasticsearch 6.7.2

初始化客户端

package com.wdz.es.config.es;import org.apache.http.HttpHost;import org.apache.http.auth.AuthScope;import org.apache.http.auth.UsernamePasswordCredentials;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.springframework.beans.factory.annotation.Value;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class ElasticsearchConfig { @Value("${elasticsearch.hosts}") private String hosts; @Value("${elasticsearch.username}") private String username; @Value("${elasticsearch.password}") private String password; @Bean(destroyMethod = "close") public RestHighLevelClient restHighLevelClient() { // 设置密码 BasicCredentialsProvider provider = new BasicCredentialsProvider(); provider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password)); RestHighLevelClient restHighLevelClient = new RestHighLevelClient( RestClient.builder(setHost()).setHttpClientConfigCallback( new RestClientBuilder.HttpClientConfigCallback() { @Override public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpAsyncClientBuilder) { httpAsyncClientBuilder.disableAuthCaching(); return httpAsyncClientBuilder.setDefaultCredentialsProvider(provider); } } )); return restHighLevelClient; } @Bean(destroyMethod = "close") public RestHighLevelClient restHighLevelClient() { RestHighLevelClient restHighLevelClient = new RestHighLevelClient(RestClient.builder(setHost())); return restHighLevelClient; } private HttpHost[] setHost(){ String[] split = hosts.split(","); HttpHost[] hhs = new HttpHost[split.length]; for (int i = 0; i < split.length; i++) { String host = split[i]; String[] ht = host.split(":"); hhs[i] = new HttpHost(ht[0], Integer.parseInt(ht[1]), "http"); } return hhs; }}

参数hosts可以多个用户逗号分隔，在代码中进行解析，格式：ip:port

校验和创建索引

private boolean checkCreate(String index) throws IOException { GetIndexRequest request = new GetIndexRequest(index); boolean exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT); if (!exists) { CreateIndexRequest create = new CreateIndexRequest(index); CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(create, RequestOptions.DEFAULT); System.out.println(JSONObject.toJSONString(createIndexResponse)); } return exists;}// 该方法可以放在业务实现类中，在服务启动时校验一次@PostConstructpublic void checkCreate(){ // 索引参数，相当于数据库中的一张表 String index = "edu-app-user"; GetIndexRequest request = new GetIndexRequest(index); try { boolean exists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT); if (!exists) { CreateIndexRequest create = new CreateIndexRequest(index); CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(create, RequestOptions.DEFAULT); System.out.println(JSONObject.toJSONString(createIndexResponse)); } } catch (IOException e) { e.printStackTrace(); }}

同步添加数据

为了方便调试，所有接口使用get方式

# 方式一：使用map方式进行添加数据，推荐使用@GetMapping("add")public IndexResponse add(String id, String name, String mobile, Integer age) { Map jsonMap = new HashMap<>(); jsonMap.put("id", id); jsonMap.put("name", name); jsonMap.put("mobile", mobile); jsonMap.put("age", age); // 参数1: 索引，同一个文档（或者说数据库中的表名）唯一， // 参数2：文档类型，在高版本中已经优化，可不需要该参数， // 参数3：id,内部id参数值，也可默认,该参数是String类型 IndexRequest request = new IndexRequest("edu-app-user", "doc", id).source(jsonMap); // 选填数据 request.opType(DocWriteRequest.OpType.CREATE); IndexResponse response = null; try { // 校验索引是否存在，否则 checkCreate("edu-app-user"); // 执行添加，默认的请求操作方式 response = restHighLevelClient.index(request, RequestOptions.DEFAULT); } catch (IOException e) { e.printStackTrace(); } return response;}

# 方式二 拼接json方式，该方式不推荐使用，容易出错@GetMapping("add2")public IndexResponse add2(String id, String name, String mobile, Integer age) { IndexRequest request = new IndexRequest("edu-app-user", "doc", id); String jsonString = "{" + ""id":" + id + "," + ""name":" + name + "," + ""mobile":" + mobile + "," + ""age":" + age + "}"; request.source(jsonString, XContentType.JSON.mediaType()); IndexResponse response = null; try { response = restHighLevelClient.index(request, RequestOptions.DEFAULT); } catch (IOException e) { e.printStackTrace(); } return response;}# XContentType 中定义多种类型，默认是XContentType.JSON = "application/json",会出现编码问题

# 方式三 推荐使用，更简洁易读@GetMapping("add3")public IndexResponse add3(String id, String name, String mobile, Integer age) throws IOException { XContentBuilder builder = XContentFactory.jsonBuilder(); builder.startObject(); { builder.field("id", id); builder.field("name", name); builder.field("mobile", mobile); builder.timeField("updateTime", new Date()); builder.field("age", age); } builder.endObject(); IndexRequest request = new IndexRequest("edu-app-user", "doc", id); request.source(builder); IndexResponse response = null; try { response = restHighLevelClient.index(request, RequestOptions.DEFAULT); } catch (IOException e) { e.printStackTrace(); } return response;}

# 方式四：直接在request.source()中进行参数操作@GetMapping("add4")public IndexResponse add4(String id, String name, String mobile, Integer age) { IndexRequest request = new IndexRequest("edu-app-user", "doc", id) .source("id", id, "name", name, "age", age, "mobile", mobile); IndexResponse response = null; try { response = restHighLevelClient.index(request, RequestOptions.DEFAULT); } catch (IOException e) { e.printStackTrace(); } return response;}

# 其他可选方法# 过期时间指定，两种方式request.timeout(Timevalue.timevalueSeconds(1)); request.timeout("1s");# 刷新策略request.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL); request.setRefreshPolicy("wait_for");

异步添加数据

同步与异步区别是方法上：(springboot 中的es版本和显式引入的依赖会出现冲突问题，直接显式定义es版本即可解决)

6.7.2

@GetMapping("addAsync")public void addAsync(String id, String name, String mobile, Integer age) { IndexRequest request = new IndexRequest("edu-app-user", "doc", id) .source("id", id, "name", name, "age", age, "mobile", mobile); restHighLevelClient.indexAsync(request, RequestOptions.DEFAULT, new EduAppUserListener()); }

# 监听类package com.wdz.es.config.es;import org.elasticsearch.action.ActionListener;import org.elasticsearch.action.index.IndexResponse;public class EduAppUserListener implements ActionListener { @Override public void onResponse(IndexResponse response) { System.out.println("异步保存成功："+ response); } @Override public void onFailure(Exception e) { System.out.println("异步保存失败："+ e.getMessage()); }}

上一篇：Elasticsearch基础与概念
下一篇：Elasticsearch Java REST Client 查询