一. 加入依赖
需要对应ES版本
1. <!--ElasticSearch 依赖-->
2. <dependency>
3. <groupId>org.elasticsearch.client</groupId>
4. <artifactId>elasticsearch-rest-high-level-client</artifactId>
5. <version>7.9.1</version>
6. </dependency>
7. <dependency>
8. <groupId>org.elasticsearch</groupId>
9. <artifactId>elasticsearch</artifactId>
10. <version>7.9.1</version>
11. </dependency>
12.
13. <!--阿里 fastjson-->
14. <dependency>
15. <groupId>com.alibaba</groupId>
16. <artifactId>fastjson</artifactId>
17. <version>1.2.58</version>
18. </dependency>
19.
20. <!--lombok-->
21. <dependency>
22. <groupId>org.projectlombok</groupId>
23. <artifactId>lombok</artifactId>
24. </dependency>二.配置 application.yml
1. @Getter
2. @Setter
3. @AllArgsConstructor
4. @NoArgsConstructor
5. @Configuration
6. public class EsEnvironmentProperties{
7.
8. @Value("${es.host}")
9. private String host; // ES请求地址
10.
11. @Value("${es.port}")
12. private int port; //端口
13.
14. @Value("${es.protocol}")
15. private String protocol; //协议
16.
17. // 可配置多ES
18. @Bean(name = "restHighLevelClientPre")
19. public RestHighLevelClient getRestHighLevelClientPre() {
20. return new RestHighLevelClient(RestClient
21. .builder(new HttpHost(host, port, protocol)));
22. }
23. }1. @Getter
2. @Setter
3. @AllArgsConstructor
4. @NoArgsConstructor
5. @Configuration
6. public class EsEnvironmentProperties{
7.
8. @Value("${es.host}")
9. private String host; // ES请求地址
10.
11. @Value("${es.port}")
12. private int port; //端口
13.
14. @Value("${es.protocol}")
15. private String protocol; //协议
16.
17. // 可配置多ES
18. @Bean(name = "restHighLevelClientPre")
19. public RestHighLevelClient getRestHighLevelClientPre() {
20. return new RestHighLevelClient(RestClient
21. .builder(new HttpHost(host, port, protocol)));
22. }
23. }1. es:
2. host: 127.0.0.1
3. port: 9200
4. protocol: http三. 封装ES操作
1. @Component
2. public class ESUtils {
3.
4. private static RestHighLevelClient restHighLevelClient;
5.
6. private static volatile ESUtils eSUtils;
7. private ESUtils(){}
8.
9. public static ESUtils getInstance(RestHighLevelClient restclient) {
10. restHighLevelClient = restclient;
11.
12. if (eSUtils == null) {
13. synchronized (ESUtils.class) {
14. if (eSUtils == null) {
15. eSUtils = new ESUtils();
16. }
17. }
18. }
19. return eSUtils;
20. }
21.
22. /**
23. * 查询
24. * @param index 索引
25. * @param searchSourceBuilder
26. */
27. public SearchResponse searchBySearchSourceBuilde(String index,
28. SearchSourceBuilder searchSourceBuilder) throws IOException {
29. // 组装SearchRequest请求
30. SearchRequest searchRequest = new SearchRequest(index);
31. searchRequest.source(searchSourceBuilder);
32. // 同步获取SearchResponse结果
33. SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
34.
35. return searchResponse;
36. }
37.
38. /**
39. * 批量插入 批量修改删除原理相同，也可混合 只需在 bulkRequest.add 不同请求即可
40. * @param jsonStrList 待插入集合 注：每个String元素需为json字符串
41. * @param index 索引
42. * @return
43. */
44. public BulkResponse bulkInsert(List<String> jsonStrList,String index) {
45. BulkRequest bulkRequest = new BulkRequest();
46.
47. for (String jsonStr : jsonStrList) {
48. IndexRequest indexRequest = new IndexRequest(index);
49. indexRequest.source(jsonStr, XContentType.JSON);
50.
51. bulkRequest.add(indexRequest); // 加入到批量请求bulk
52. }
53.
54. BulkResponse bulkResponse = null;
55. try {
56. bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
57. } catch (IOException e) {
58. e.printStackTrace();
59. }
60. return bulkResponse;
61. }
62. }具体使用
1. @Resource(name = "restHighLevelClientPre")
2. RestHighLevelClient restHighLevelClient;
3.
4. //查询
5. public void search() {
6. // user_city 完全匹配 Beijing 且 (2020-6-1 <= user_time <= 2020-6-2)
7. QueryBuilder queryBuilder = QueryBuilders.boolQuery()
8. .must(QueryBuilders.termQuery("user_city", "Beijing"))
9. .must(QueryBuilders.rangeQuery("user_time").gte("2020-6-1 00:00:00"))
10. .must(QueryBuilders.rangeQuery("user_time").lte("2020-6-2 00:00:00"));
11.
12. SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
13. searchSourceBuilder.query(queryBuilder);
14. // 按 user_time 升序排序
15. searchSourceBuilder.sort("user_time", SortOrder.ASC);
16. // 设置返回数量
17. searchSourceBuilder.size(1000);
18.
19. SearchResponse searchResponse = null;
20. try {
21. searchResponse = ESUtils.getInstance(restHighLevelClient)
22. .searchBySearchSourceBuilde("users", searchSourceBuilder);
23. } catch (IOException e) {
24. e.printStackTrace();
25. }
26. SearchHit[] hitsArr = searchResponse.getHits().getHits();
27.
28. List<User> userList = new ArrayList<>();
29. for (SearchHit searchHit : hitsArr) {
30. User user = new User();
31.
32. JSONObject source = JSONObject.parseObject(searchHit.getSourceAsString());
33. user.setUser_id(source.getString("user_id"));
34. user.setUser_name(source.getString("user_name"));
35. userList.add(user);
36. }
37. }
38.
39. //批量插入
40. public void bulkInsert(List<User> userList) {
41. List<String> jsonList = new ArrayList<>();
42.
43. for (User user : userList) {
44. // User 转为 Json字符串
45. jsonList.add(JSONObject.toJSONString(user));
46. }
47. BulkResponse bulkResponse = ESUtils.getInstance(restHighLevelClient).bulkInsert(jsonList, "users");
48. }
