如何在java项目中使用ES_java 使用es

1. 引入es的依赖库

    <!-- elasticsearch-->
           <dependency>
               <groupId>org.springframework.boot</groupId>
               <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
           </dependency>
   1
   2
   3
   4
   5

2. 创建一个接口来从es里面查询数据 service文件中

    /**
        * 从 ES 查询
        *
        * @param postQueryRequest
        * @return
        */
       Page<Post> searchFromEs(PostQueryRequest postQueryRequest);
   
   // postQueryRequest是封装的请求参数类，主要需要传入的字段就是searchText（搜索关键词）
   1
   2
   3
   4
   5
   6
   7
   8
   9

3. 实现该接口 （ES负责的是静态查询，将查询结果对应的文档id找到之后再返回到mysql里面查询更加完整的数据）

   public Page<Post> searchFromEs(PostQueryRequest postQueryRequest) {
           // 01. 将所有的参数给单独提取出来
           String searchText = postQueryRequest.getSearchText();
           // 02. 指定查询的数据的页数 es 起始页为 0
           long current = postQueryRequest.getCurrent() - 1;
           long pageSize = postQueryRequest.getPageSize();
   
           // 03. 创建一个查询对象
           BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
   
           // 04. 查询条件过滤
           boolQueryBuilder.filter(QueryBuilders.termQuery("isDelete", 0));
           // 05. 判断是否有传入以下的查询条件，如果有的话就加入到查询条件中
           // 按关键词检索
           if (StringUtils.isNotBlank(searchText)) {
               boolQueryBuilder.should(QueryBuilders.matchQuery("title", searchText));
               boolQueryBuilder.should(QueryBuilders.matchQuery("description", searchText));
               boolQueryBuilder.should(QueryBuilders.matchQuery("content", searchText));
               boolQueryBuilder.minimumShouldMatch(1);  // 至少匹配一个
           }
           // 分页
           PageRequest pageRequest = PageRequest.of((int) current, (int) pageSize);
           // 构造排序的查询
   //        NativeSearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder)
   //                .withPageable(pageRequest).withSorts(sortBuilder).build();
   
           // 构造没有排序的查询
           NativeSearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(boolQueryBuilder)
                   .withPageable(pageRequest).build();
   
           // 将所有的查询结果都取出来
           SearchHits<PostEsDTO> searchHits = elasticsearchRestTemplate.search(searchQuery, PostEsDTO.class);
   
           Page<Post> page = new Page<>();
           // 将查询的结果添加到page对象里面
           page.setTotal(searchHits.getTotalHits());
   
           List<Post> resourceList = new ArrayList<>();
           // 查出结果后，从 db 获取最新动态数据（比如点赞数）     es负责进行静态数据的筛选，然后在回表到mysql里面将所有的数据信息查出来
           if (searchHits.hasSearchHits()) {
               List<SearchHit<PostEsDTO>> searchHitList = searchHits.getSearchHits();
               // 将查询到的文档id使用列表进行存储
               List<Long> postIdList = searchHitList.stream().map(searchHit -> searchHit.getContent().getId())
                       .collect(Collectors.toList());
               // 根据这个文档id查询 mysql里面的数据  使用列表存储
               List<Post> postList = baseMapper.selectBatchIds(postIdList);
               if (postList != null) {
                   // 根据文章id进行分组
                   Map<Long, List<Post>> idPostMap = postList.stream().collect(Collectors.groupingBy(Post::getId));
   
                   // 如果查询到的mysql的集合里面包含上面es中查询到的文档id就取出来放到resource列表中存储
                   postIdList.forEach(postId -> {
                       if (idPostMap.containsKey(postId)) {
                           resourceList.add(idPostMap.get(postId).get(0));
                       } else {
                           // 从 es 清空 db 已物理删除的数据   不包含的话就直接删除
                           String delete = elasticsearchRestTemplate.delete(String.valueOf(postId), PostEsDTO.class);
                           log.info("delete post {}", delete);
                       }
                   });
               }
           }
           // 设置到records中支持分页存储
           page.setRecords(resourceList);
           return page;
       }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15
   16
   17
   18
   19
   20
   21
   22
   23
   24
   25
   26
   27
   28
   29
   30
   31
   32
   33
   34
   35
   36
   37
   38
   39
   40
   41
   42
   43
   44
   45
   46
   47
   48
   49
   50
   51
   52
   53
   54
   55
   56
   57
   58
   59
   60
   61
   62
   63
   64
   65
   66

4. 然后在需要查询es里面数据的地方调用该接口就行

    @Override
       public Page<PostVO> doSearch(String searchText, int pageSize, int pageNum) {
           PostQueryRequest postQueryRequest = new PostQueryRequest();
           postQueryRequest.setSearchText(searchText);
           postQueryRequest.setPageSize(pageSize);
           postQueryRequest.setCurrent(pageNum);
           // 这里因为不能在传入request参数  所以就将request参数从requestHolder里面获取
           ServletRequestAttributes requestAttributes = (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
           HttpServletRequest request = requestAttributes.getRequest();
           // 将查询帖子的接口转到es里面去进行查询    调用es查询的接口
           Page<Post> postPage = postService.searchFromEs(postQueryRequest);
           Page<PostVO> postVOPage = postService.getPostVOPage(postPage, request);
   //        Page<PostVO> postResult = postService.listPostVOByPage(postQueryRequest, request);
           return postVOPage;
       }
   1
   2
   3
   4
   5
   6
   7
   8
   9
   10
   11
   12
   13
   14
   15

5. 上面的接口实现出现了一个问题，就是我们的es里面根本就没有任何的数据，所以也就无从查起了

   1. es里面一般负责静态数据的查询，然后将结构查询出来之后（关联的文档id）再到mysql里面进行动态数据的查询

   2. 这时我们需要将mysql的数据同步到es里面，但不是全部的数据同步，而是只需要同步部分字段的数据即可，比如这里需要同步id，title，content, description

   3. 数据库数据同步的方式有很多种，这里我们采用的时定时任务的方式进行数据同步（占用资源少，不用引入第三方插件，适用于不需要太实时同步数据的场景）

   • 创建一个es表的实体类

     package com.yupi.springbootinit.model.dto.post;
     
     import cn.hutool.core.collection.CollUtil;
     import cn.hutool.json.JSONUtil;
     import com.yupi.springbootinit.model.entity.Post;
     import lombok.Data;
     
     import org.apache.commons.lang3.StringUtils;
     import org.springframework.beans.BeanUtils;
     import org.springframework.data.annotation.Id;
     import org.springframework.data.elasticsearch.annotations.Document;
     import org.springframework.data.elasticsearch.annotations.Field;
     import org.springframework.data.elasticsearch.annotations.FieldType;
     
     import java.io.Serializable;
     import java.util.Date;
     import java.util.List;
     
     /**
      * 帖子 ES 包装类
      *
      **/
     // todo 取消注释开启 ES（须先配置 ES）
     @Document(indexName = "post")   // es的文档索引（表名）
     @Data
     public class PostEsDTO implements Serializable {
     
         private static final String DATE_TIME_PATTERN = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'";
     
         /**
          * id  必须打上id注解
          */
         @Id
         private Long id;
     
         /**
          * 标题
          */
         private String title;
     
         /**
          * 内容
          */
         private String content;
     
         /**
          * 标签列表
          */
         private List<String> tags;
     
         /**
          * 创建用户 id
          */
         private Long userId;
     
         /**
          * 创建时间  解析时间使得java时间和es时间符合
          */
         @Field(index = false, store = true, type = FieldType.Date, format = {}, pattern = DATE_TIME_PATTERN)
         private Date createTime;
     
         /**
          * 更新时间
          */
         @Field(index = false, store = true, type = FieldType.Date, format = {}, pattern = DATE_TIME_PATTERN)
         private Date updateTime;
     
         /**
          * 是否删除
          */
         private Integer isDelete;
     
         private static final long serialVersionUID = 1L;
     
         /**
          * 对象转包装类
          *
          * @param post
          * @return
          */
         public static PostEsDTO objToDto(Post post) {
             if (post == null) {
                 return null;
             }
             PostEsDTO postEsDTO = new PostEsDTO();
             BeanUtils.copyProperties(post, postEsDTO);
             String tagsStr = post.getTags();
             if (StringUtils.isNotBlank(tagsStr)) {
                 postEsDTO.setTags(JSONUtil.toList(tagsStr, String.class));
             }
             return postEsDTO;
         }
     
         /**
          * 包装类转对象
          *
          * @param postEsDTO
          * @return
          */
         public static Post dtoToObj(PostEsDTO postEsDTO) {
             if (postEsDTO == null) {
                 return null;
             }
             Post post = new Post();
             BeanUtils.copyProperties(postEsDTO, post);
             List<String> tagList = postEsDTO.getTags();
             if (CollUtil.isNotEmpty(tagList)) {
                 post.setTags(JSONUtil.toJsonStr(tagList));
             }
             return post;
         }
     }
     
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35
     36
     37
     38
     39
     40
     41
     42
     43
     44
     45
     46
     47
     48
     49
     50
     51
     52
     53
     54
     55
     56
     57
     58
     59
     60
     61
     62
     63
     64
     65
     66
     67
     68
     69
     70
     71
     72
     73
     74
     75
     76
     77
     78
     79
     80
     81
     82
     83
     84
     85
     86
     87
     88
     89
     90
     91
     92
     93
     94
     95
     96
     97
     98
     99
     100
     101
     102
     103
     104
     105
     106
     107
     108
     109
     110
     111
     112
     113

   • 创建一个全量的同步任务（只在项目启动的时候进行一次同步，用完之后记得将注解注释掉，这样下次就不会再执行）

     // todo 取消注释开启任务
     //@Component
     @Slf4j
     // CommandLineRunner是一个接口用于在程序启动之后进行一些初始化方法执行，可以重写里面的run方法即可
     public class FullSyncPostToEs implements CommandLineRunner {
     
         @Resource
         private PostService postService;
     
         @Resource
         private PostEsDao postEsDao;
     
         @Override
         public void run(String... args) {
             // 01. 查询里面所有的数据
             List<Post> postList = postService.list();
             if (CollUtil.isEmpty(postList)) {
                 return;
             }
             // 02. 将post查询的所有数据的tags取出来 转换为一个新的对象PostEsDTO  同时将里面的tags由json格式转为string格式
             List<PostEsDTO> postEsDTOList = postList.stream().map(post -> PostEsDTO.objToDto(post)).collect(Collectors.toList());
     //        List<PostEsDTO> postEsDTOList = postList.stream().map(PostEsDTO::objToDto).collect(Collectors.toList());
             // 03. 一次最多同步500条数据到es里面
             final int pageSize = 500;
             int total = postEsDTOList.size();
             log.info("FullSyncPostToEs start, total {}", total);
             for (int i = 0; i < total; i += pageSize) {
                 int end = Math.min(i + pageSize, total);
                 log.info("sync from {} to {}", i, end);
                 postEsDao.saveAll(postEsDTOList.subList(i, end));
             }
             log.info("FullSyncPostToEs end, total {}", total);
         }
     }
     
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35

   • 增量同步（通过定时任务的方式来判断数据更新的时间从而进行数据的同步）

     @Component
     @Slf4j
     public class IncSyncPostToEs {
     
         @Resource
         private PostMapper postMapper;
     
         @Resource
         private PostEsDao postEsDao;
     
         /**
          * 每分钟执行一次
          */
         @Scheduled(fixedRate = 60 * 1000)
         public void run() {
             // 查询5分钟内的数据
             Date fiveMinutesAgoDate = new Date(new Date().getTime() - 5 * 60 * 1000L);
             // sql查询  写在mapper里面的
             List<Post> postList = postMapper.listPostWithDelete(fiveMinutesAgoDate);
             if (CollUtil.isEmpty(postList)) {
                 log.info("no inc post");
                 return;
             }
             // 将数据进行转换为新的对象格式 PostEsDTO类型
             List<PostEsDTO> postEsDTOList = postList.stream()
                     .map(PostEsDTO::objToDto)
                     .collect(Collectors.toList());
             // 最多只能同步500条数据
             final int pageSize = 500;
             int total = postEsDTOList.size();
             log.info("IncSyncPostToEs start, total {}", total);
             for (int i = 0; i < total; i += pageSize) {
                 int end = Math.min(i + pageSize, total);
                 log.info("sync from {} to {}", i, end);
                 postEsDao.saveAll(postEsDTOList.subList(i, end));
             }
             log.info("IncSyncPostToEs end, total {}", total);
         }
     }
     1
     2
     3
     4
     5
     6
     7
     8
     9
     10
     11
     12
     13
     14
     15
     16
     17
     18
     19
     20
     21
     22
     23
     24
     25
     26
     27
     28
     29
     30
     31
     32
     33
     34
     35
     36
     37
     38
     39

总结：ES引入到java客户端使用的方法

1. 首先引入对应的ES依赖库
2. 然后进行ES的数据同步（创建ES实体封装类，编写定时任务）
3. 数据同步成功之后编写对应的查询接口以及接口实现类，实现类里面具体执行对es数据库的查询【注意这里返回的不是es里面的数据，而是返回的关联的文档ID，将该ID取出来之后去mysql数据库里面查询和ID相等的完整的动态数据才是我们需要的结果】
4. 调用接口获取数据，响应给前端即可