当测试了es连接和对索引相关操作后，接下来我们需要针对数据进行相关简单的操作
新增数据的接口只是用来测试，实际开发过程中直接通过canal同步数据即可
public void addOneData() throws Exception {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
// Person p1 = new Person(1, "张三", 18, new Date());
// Person p1 = new Person(2, "李四", 20, new Date());
Person p1 = new Person(3, "王五", 22, new Date());
String docId = client.index(b -> b
.index(indexName)
.id(p1.getId().toString()) //设置存储文件的id，可不设置，如不设置，es会自动分配一个id（索引要为字符串格式）
.document(p1)
.refresh(Refresh.True) // Make it visible for search
).id();
System.out.println(docId);
return;
} catch (Exception e) {
e.printStackTrace();
log.error("新增es单个数据失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
public void addBatchData() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
Person p1 = new Person(7, "赵六2", 19, new Date());
Person p2 = new Person(5, "孙七2", 20, new Date());
Person p3 = new Person(6, "周八", 20, new Date());
List<Person> personList = new ArrayList<>();
personList.add(p1);
personList.add(p2);
BulkRequest.Builder br = new BulkRequest.Builder();
//此种方式如id存在则会直接覆盖，如id不存在则会直接新增
//如果索引不存在，则会直接新增索引
for (Person person : personList) {
br.operations(op -> op
.index(idx -> idx
.index(indexName)
.id(person.getId().toString())
.document(person)
)
);
}
BulkResponse bulk = client.bulk(br.build());
//下面方式如果设置了create，update方法，如此id存在，则添加失败。如此id不存在，则修改失败
// BulkResponse bulk = client.bulk(_0 -> _0
// .operations(_1 -> _1
// .create(_2 -> _2
// .index(indexName)
// .id(p1.getId().toString())
// .document(p1)
// ))
// .operations(_1 -> _1
// .update(_2 -> _2
// .index(indexName)
// .id(p3.getId().toString())
// .action(_3 -> _3
// .docAsUpsert(true)
// .doc(p3))
// )
// )
// );
//每条都成功才会返回成功，有一条不成功error状态都为true
if (bulk.errors()) {
System.out.println("有部分数据操作失败");
for (BulkResponseItem item : bulk.items()) {
if (item.error() != null) {
System.out.println(item.error().reason());
}
}
}
System.out.println(JSON.toJSONString(bulk));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("es批量新增数据失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
public void judgeDataIfExistById() {
ElasticsearchClient client = null;
String docId = "9"; //要查询的文档id
try {
client = this.getEsClient();
GetResponse<Person> person = client.get(b -> b
.index(indexName)
.id(docId)
, Person.class
);
System.out.println("此id数据是否存在：" + person.found());
return;
} catch (Exception e) {
e.printStackTrace();
log.error("根据id获取es数据失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
public void getDataList() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
SearchResponse<Person> search = client.search(b -> b
.index(indexName)
, Person.class
);
// long totalNum = search.hits().total().value(); //查询总条数
List<Person> list = search.hits().hits().stream().map(p -> p.source()).collect(Collectors.toList());
System.out.println(JSON.toJSONString(list));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询es列表失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
查询多索引内容要确保索引的实体要一样
public void getDataListMoreIndex() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
//此案例为同时查询存在的索引和不存在的索引
MsearchResponse<Person> msearch = client.msearch(_0 -> _0
.searches(_1 -> _1
.header(_3 -> _3.index(indexName))
.body(_3 -> _3.query(_4 -> _4.matchAll(_5 -> _5)))
).searches(_1 -> _1
.header(_3 -> _3.index("non-existing"))
.body(_3 -> _3.query(_4 -> _4.matchAll(_5 -> _5)))
)
, Person.class);
System.out.println(JSON.toJSONString(msearch));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询es多索引内容失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
public void getDataById() {
ElasticsearchClient client = null;
String docId = "1"; //要查询的文档id
try {
client = this.getEsClient();
Person person = client.get(b -> b
.index(indexName)
.id(docId)
, Person.class
).source();
//如果不知道返回的实体类，直接以ObjectNode接收
// GetResponse response = client.get(g -> g
// .index(indexName)
// .id(docId),
// ObjectNode.class
// );
//
// if (response.found()) {
// ObjectNode json = response.source();
// String name = json.get("name").asText();
// System.out.println("Person name " + name);
// } else {
// System.out.println("Person not found");
// }
System.out.println(JSON.toJSONString(person));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("根据id获取es数据失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
public ApiResult getAllContent() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
SearchResponse<Map> search = client.search(b -> b
.index(indexName)
//es默认返回10000条数据，加上下面配置才能返回真正的总条数
.trackTotalHits(t -> t.enabled(true))
, Map.class
);
long totalNum = search.hits().total().value(); //查询总条数
return totalNum ;
} catch (Exception e) {
e.printStackTrace();
log.error("查询总条数失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
return ApiResult.error("查询失败");
}
筛选姓名带有 “ 三 ” 的人员信息
public void getDataListContentOne() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
String searchText = "三";
SearchResponse<Person> response = client.search(s -> s
.index(indexName)
.query(q -> q
.match(t -> t //在众多可用的查询变体中选择一个。我们在这里选择匹配查询（全文搜索）
.field("name")
.query(searchText)
)
),
Person.class
);
TotalHits total = response.hits().total();
boolean isExactResult = total.relation() == TotalHitsRelation.Eq;
if (isExactResult) {
log.info("There are " + total.value() + " results");
} else {
log.info("There are more than " + total.value() + " results");
}
List<Hit<Person>> hits = response.hits().hits();
for (Hit<Person> hit : hits) {
Person product = hit.source();
System.out.println("Found Person " + product.getName() + ", score " + hit.score());
}
return;
} catch (Exception e) {
e.printStackTrace();
log.error("es筛选查询失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
筛选姓名带有“三”并且年龄大于等于18的人员信息
public void getDataListContentTwo() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
String searchText = "三";
double bigAge = 18;
//添加姓名查询条件
Query byName = MatchQuery.of(m -> m
.field("name")
.query(searchText)
)._toQuery();
//添加年龄查询条件
Query byMaxAge = RangeQuery.of(r -> r
.field("age")
.gte(JsonData.of(bigAge)) //equals：等于 gt：大于 lt：小于 gte：大于等于 lte：小于等于
)._toQuery();
//联合查询条件放入查询请求中
SearchResponse<Person> response = client.search(s -> s
.index(indexName)
.query(q -> q
.bool(b -> b
.filter(byName) //filter不计算评分，效率更高。must计算评分，效率稍低
.filter(byMaxAge)
// .must(byName) //must表示匹配所有条件
// .must(byMaxAge)
)
),
Person.class
);
List<Hit<Person>> hits = response.hits().hits();
for (Hit<Person> hit : hits) {
Person person = hit.source();
System.out.println("Found person " + person.getName() + ", score " + hit.score());
}
return;
} catch (Exception e) {
e.printStackTrace();
log.error("es筛选查询失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
public void getDataListOne() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
String searchText = "三";
SearchResponse<Person> response = client.search(s -> s
.index(indexName)
.from(0) //从第0条开始查
.size(15), //一次查15条
Person.class
);
List<Person> list = search.hits().hits().stream().map(p -> p.source()).collect(Collectors.toList());
return;
} catch (Exception e) {
e.printStackTrace();
log.error("es筛选查询失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
排序和分页可以结合使用，大家根据自己需要自行组合
public void getDataListTwo() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
SearchResponse<Person> search = client.search(b -> b
.index(indexName)
.sort(_2 -> _2
.field(_3 -> _3
.field("uploadTime") //按上传时间倒序排列
.order(SortOrder.Desc)))
, Person.class
);
// long totalNum = search.hits().total().value(); //查询总条数
List<Person> list = search.hits().hits().stream().map(p -> p.source()).collect(Collectors.toList());
System.out.println(JSON.toJSONString(list));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询es列表失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
在开发过程中，不想一下子把所有字段都返回给前端，则需要指定返回特定字段，有两种方式，个人测试时第二种效率稍高，大家根据自己需要自行查看
public void getDataListContentFour() {
ElasticsearchClient client = null;
try {
client = EsUtils.getEsClient(esIp, esPort);
SearchResponse<Map> response = client.search(_1 -> _1
.index(indexName)
.source(_2 -> _2
.fetch(false)) //关闭source总字段列表
.fields(_2 -> _2.field("id")) //返回特定字段值，需要多个返回值则继续追加
.fields(_2 -> _2.field("name"))
.from(0) //从第0条开始查
.size(15), //一次查15条
Map.class
);
List<Map<String, JsonData>> list = response.hits().hits().stream().map(m -> m.fields()).collect(Collectors.toList());
//返回的格式为List>>>，要将此结构改为List>>
List<HashMap<String, ? extends Serializable>> collect = list.stream().map(map ->
map.entrySet().stream().collect(Collectors.toMap(
item -> item.getKey(),
//返回结果如果为字符串，则会有""符号，则通过是否有""符号判断字段类型
item -> Optional.ofNullable(item.getValue().toJson().asJsonArray().get(0).toString().contains("\"") ?
item.getValue().toJson().asJsonArray().get(0).toString().replaceAll("\"", "") :
Long.valueOf(item.getValue().toJson().asJsonArray().get(0).toString())).orElse(null),
// item -> Optional.ofNullable(item.getValue().toJson().asJsonArray().get(0).toString().replaceAll("\"", "")).orElse(null),
(oldVal, currVal) -> oldVal, HashMap::new)))
.collect(Collectors.toList());
System.out.println(collect);
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("查询失败");
}
第二种是直接source源中设置过滤的字段，也可以达到返回特定字段的效果
public void getDataListContentFive() {
ElasticsearchClient client = null;
try {
client = EsUtils.getEsClient(esIp, esPort);
SearchResponse<Map> response = client.search(_1 -> _1
.index(indexName)
.source(_2 -> _2
.filter(_3 -> _3
.includes("id", "name", "age")//需要显示的字段
.excludes("")))//需要排除的字段
.from(0) //从第0条开始查
.size(15), //一次查15条
Map.class
);
List<Map> list = response.hits().hits().stream().map(m -> m.source()).collect(Collectors.toList());
System.out.println(list);
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("查询失败");
}
在开发过程中，有时候多个查询参数不是固定的，则需要根据是否有参数来判断是否需要添加查询条件，则需要实现追加参数的功能
@Test
public void getDatListThree() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
SearchRequest.Builder request = new SearchRequest.Builder();
request.index(indexName)//设置索引
.source(_2 -> _2
.filter(_3 -> _3 //设置过滤显示字段
.includes("name", "age","birthday")
.excludes("")))
.query(_2 -> _2 //查询条件一
.bool(_3 -> _3
.must(_4 -> _4 //must为筛选匹配满足所有条件的记录
.match(_5 -> _5
.field("name")
.query("张三")))
)
);
request.query(_2 -> _2
.bool(_3 -> _3
.must(_4 -> _4 //追加筛选条件 年龄为24的记录
.match(_5 -> _5
.field("age")
.query(18))))
);
SearchResponse<Map> search = client.search(request.build(), Map.class);
List<Map> list = search.hits().hits().stream().map(m -> m.source()).collect(Collectors.toList());
System.out.println(JSON.toJSONString(list));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询es列表失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}
上述写法是将整个查询内容都放在外面，还有一种实现方式是生成动态query内容，其余内容按原来写法
@Test
public void getDatListFour() {
ElasticsearchClient client = null;
try {
client = this.getEsClient();
BoolQuery.Builder queryBuilder = new BoolQuery.Builder();
//添加的第一个筛选条件
queryBuilder.must(_1 -> _1
.match(_2 -> _2
.field("name")
.query("张三")));
//添加的第二个筛选条件
queryBuilder.must(_1 -> _1
.match(_2 -> _2
.field("age")
.query(18)));
SearchResponse<Map> search = client.search(_1 -> _1
.index(indexName)
.source(_2 -> _2
.filter(_3 -> _3 //设置过滤显示字段
.includes("name", "age", "birthday")
.excludes("")))
.query(queryBuilder.build()._toQuery())
.from(0) //从第0条开始查
.size(15), //一次查15条
Map.class
);
List<Map> list = search.hits().hits().stream().map(m -> m.source()).collect(Collectors.toList());
System.out.println(JSON.toJSONString(list));
return;
} catch (Exception e) {
e.printStackTrace();
log.error("查询es列表失败" + e);
} finally {
try {
client._transport().close();
} catch (IOException e) {
e.printStackTrace();
}
}
System.out.println("失败");
}