本文使用的SpringBoot版本为2.5.4，elasticsearch版本为7.15.0
1. pom文件如下2. application.yml配置如下4.0.0 org.springframework.boot spring-boot-starter-parent2.5.4 com.yh es-demo0.0.1-SNAPSHOT es-demo es-demo 1.8 7.15.0 org.springframework.boot spring-boot-starter-weborg.springframework.boot spring-boot-configuration-processortrue org.projectlombok lomboktrue org.springframework.boot spring-boot-starter-testtest org.elasticsearch.client elasticsearch-rest-high-level-client${elasticsearch.version} org.elasticsearch elasticsearch${elasticsearch.version} com.alibaba fastjson1.2.70 test org.springframework.boot spring-boot-maven-pluginorg.projectlombok lombok
elasticsearch:
host-arr:
- 127.0.0.1:9200
二、ES SpringBoot配置
1. 配置ES服务器地址
package com.yh.esdemo.config;
import lombok.Getter;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@Component
@Getter
@Setter
@ConfigurationProperties(prefix = "elasticsearch")
public class EsHostConfig {
private String[] hostArr;
}
2. 配置RestHighLevelClient
package com.yh.esdemo.config;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class EsRestClientConfig {
@Autowired
private EsHostConfig esHostConfig;
@Bean
public RestHighLevelClient restHighLevelClient() {
String[] hostArr = esHostConfig.getHostArr();
int size = hostArr.length;
HttpHost[] httpHostArr = new HttpHost[size];
for (int i = 0; i < size; i++) {
String[] split = hostArr[i].split(":");
httpHostArr[i] = new HttpHost(split[0], Integer.parseInt(split[1]), "http");
}
return new RestHighLevelClient(RestClient.builder(httpHostArr));
}
}
配置完成后只需在相关业务中注入RestHighLevelClient即可使用
三、ES基本使用 索引操作 1. 创建索引 @Test
public void addIndex() throws Exception {
//获取 IndicesClient ，用于创建 Index
IndicesClient indices = restHighLevelClient.indices();
//定义一个 CreateIndexRequest ，参数为索引的名称
CreateIndexRequest createIndexRequest = new CreateIndexRequest("yh_customer");
//CreateIndexRequest配置相关参数 number_of_shards：分片数 number_of_replicas：每个分片的副本数
createIndexRequest.settings(
Settings.builder()
.put("index.number_of_shards", 1)
.put("index.number_of_replicas", 0)
);
Map name = new HashMap<>();
name.put("type", "text");
name.put("analyzer", "ik_max_word");//指定分词器
name.put("search_analyzer", "ik_smart");
Map timestamp = new HashMap<>();
timestamp.put("type", "date");
timestamp.put("format", "yyyy-MM-dd HH:mm:ss");//指定时间格式
Map uuid = new HashMap<>();
uuid.put("type", "keyword");
Map properties = new HashMap<>();
properties.put("name", name);
properties.put("timestamp", timestamp);
properties.put("uuid", uuid);
Map mapping = new HashMap<>();
mapping.put("properties", properties);
//创建mapping mapping可以理解为此索引中数据的数据结构
createIndexRequest.mapping(mapping);
//创建索引 此方法同步返回结果 异步使用 createAsync
CreateIndexResponse createIndexResponse = indices.create(createIndexRequest, RequestOptions.DEFAULT);
//创建结果（true/false）
System.out.println(createIndexResponse.isAcknowledged());
//输出创建成功的索引名称
System.out.println(createIndexResponse.index());
}
最终输出:
@Test
public void delIndex() throws Exception {
IndicesClient indices = restHighLevelClient.indices();
DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest("yh_customer");
AcknowledgedResponse deleteIndexResponse = indices.delete(deleteIndexRequest, RequestOptions.DEFAULT);
System.out.println(deleteIndexResponse.isAcknowledged());
}
增删改查
1. 添加数据（单个）
@Test
public void addDoc() throws Exception {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String timestamp = simpleDateFormat.format(new Date());
Map jsonMap = new HashMap<>();
jsonMap.put("uuid", "534523dqw4qweq24324");
jsonMap.put("name", "张三");
jsonMap.put("timestamp", timestamp);
//创建 IndexRequest 指定id
IndexRequest indexRequest = new IndexRequest("yh_customer").id("1").source(jsonMap);
IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
System.out.println(indexResponse);
}
最终输出：
@Test
public void bulkTest() {
//批量添加使用 BulkRequest
BulkRequest bulkRequest = new BulkRequest();
for (int i = 0; i < 100; i++) {
Customer customer = new Customer(UUID.randomUUID().toString().replace("-", ""), i + "号客户", LocalDateTime.now());
//String data = JSONObject.toJSonString(customer);
String data = JSONObject.toJSonStringWithDateFormat(customer, "yyyy-MM-dd HH:mm:ss");
IndexRequest request = new IndexRequest("yh_customer").source(data, XContentType.JSON);
bulkRequest.add(request);
}
//同步
// try {
// BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
// boolean hasFailures = bulk.hasFailures();
// if (hasFailures) {
// List collect = Arrays.stream(bulk.getItems()).filter(BulkItemResponse::isFailed).collect(Collectors.toList());
// System.out.println(collect.size());
// String msg = bulk.buildFailureMessage();
// System.out.println(msg);
// }
// } catch (IOException e) {
// e.printStackTrace();
// }
//异步
restHighLevelClient.bulkAsync(bulkRequest, RequestOptions.DEFAULT, new ActionListener() {
//成功回调
@Override
public void onResponse(BulkResponse bulkItemResponses) {
System.out.println("success");
}
//失败回调
@Override
public void onFailure(Exception e) {
System.out.println("failure");
e.printStackTrace();
}
});
try {
TimeUnit.MINUTES.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
最终输出：
@Test
public void updateDoc() throws Exception {
UpdateRequest updateRequest = new UpdateRequest("yh_customer", "1");
Map jsonMap = new HashMap<>();
jsonMap.put("name", "李四");
updateRequest.doc(jsonMap);
UpdateResponse update = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
System.out.println(update);
}
最终输出:
@Test
public void get() throws Exception {
GetRequest getRequest = new GetRequest("yh_customer");
getRequest.id("1");
GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
System.out.println(getResponse);
}
最终输出：
@Test
public void getSource() throws Exception {
GetSourceRequest getRequest = new GetSourceRequest("yh_customer", "1");
GetSourceResponse source = restHighLevelClient.getSource(getRequest, RequestOptions.DEFAULT);
System.out.println(source);
}
最终输出：
@Test
public void exists() throws Exception {
GetRequest getRequest = new GetRequest("yh_customer");
getRequest.id("1");
boolean exists = restHighLevelClient.exists(getRequest, RequestOptions.DEFAULT);
System.out.println(exists);
}
最终输出:
@Test
public void search() throws Exception {
SearchRequest searchRequest = new SearchRequest("yh_customer");//没有参数，查询所有索引
//构建搜索条件
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
//根据name字段搜索
searchSourceBuilder.query(QueryBuilders.matchQuery("name", "1"));
searchRequest.source(searchSourceBuilder);
SearchResponse search = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
//返回命中数据
SearchHits hits = search.getHits();
for (SearchHit hit : hits.getHits()) {
String sourceAsString = hit.getSourceAsString();
Customer customer = JSONObject.parseObject(sourceAsString, Customer.class);
System.out.println(customer);
}
}
最终输出:
@Test
public void simpleSearch() throws Exception {
//创建搜索request
SearchRequest searchRequest = new SearchRequest("bank");
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
searchSourceBuilder.query(QueryBuilders.matchAllQuery());//检索所有字段
SearchRequest request = searchRequest.source(searchSourceBuilder);
SearchResponse search = restHighLevelClient.search(request, RequestOptions.DEFAULT);
SearchHits hits = search.getHits();//命中条数
System.out.println("总条数：" + hits.getTotalHits().value);
for (SearchHit hit : hits) {//默认只检索出10条
Staff staff = JSONObject.parseObject(hit.getSourceAsString(), Staff.class);
System.out.println(staff);
}
}
最终输出：
需要先导入官方提供的测试数据（使用kibana）
@Test
public void aggSearch() throws Exception {
//创建搜索request
SearchRequest searchRequest = new SearchRequest("bank");
SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
//查询条件
searchSourceBuilder.query(QueryBuilders.matchAllQuery());
//统计各个年龄的平均工资
searchSourceBuilder.aggregation(AggregationBuilders.terms("ageAgg").field("age").subAggregation(AggregationBuilders.avg("balanceAvgAgg").field("balance")));
//所有人平均工资
searchSourceBuilder.aggregation(AggregationBuilders.avg("allBalanceAvgAgg").field("balance"));
//构建request
SearchRequest request = searchRequest.source(searchSourceBuilder);
//search
SearchResponse searchResponse = restHighLevelClient.search(request, RequestOptions.DEFAULT);
//Aggregations aggregations = searchResponse.getAggregations();
//Aggregation接口有很多实现类 Avg（平均值）
// Avg avg = aggregations.get("balanceAgg");
// System.out.println(avg.getValue());
Aggregations aggregations = searchResponse.getAggregations();
Terms ageAgg = aggregations.get("ageAgg");
List buckets = ageAgg.getBuckets();
for (Terms.Bucket bucket : buckets) {//只查出了10条数据
System.out.println("年龄:" + bucket.getKey());
System.out.println("人数:" + bucket.getDocCount());
Avg avg = bucket.getAggregations().get("balanceAvgAgg");
System.out.println("平均工资:" + avg.getValue());
System.out.println("=================================");
}
Avg avg = aggregations.get("allBalanceAvgAgg");
System.out.println("所有人平均工资:" + avg.getValue());
}
最终输出：
年龄:31 人数:61 平均工资:28312.918032786885 ================================= 年龄:39 人数:60 平均工资:25269.583333333332 ================================= 年龄:26 人数:59 平均工资:23194.813559322032 ================================= 年龄:32 人数:52 平均工资:23951.346153846152 ================================= 年龄:35 人数:52 平均工资:22136.69230769231 ================================= 年龄:36 人数:52 平均工资:22174.71153846154 ================================= 年龄:22 人数:51 平均工资:24731.07843137255 ================================= 年龄:28 人数:51 平均工资:28273.882352941175 ================================= 年龄:33 人数:50 平均工资:25093.94 ================================= 年龄:34 人数:49 平均工资:26809.95918367347 ================================= 所有人平均工资:25714.837
demo地址：https://github.com/yangh124/es-demo
