devbox
一键难忘520
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

ElasticSearch与SpringBoot的集成使用_springboot集成elasticsearch的使用

作者：一键难忘520 | 2024-08-06 06:41:46

赞

踩

springboot集成elasticsearch的使用

一、ElasticSearch了解

        ElasticSearch是一个基于lucene的分布式检索服务器。相对于solr搜索，在大数据量和数据并发量上更有优势，同时，也有数据库的数据存储功能，但提供了比数据库更多的存储功能，如分词搜索、关联度搜索等，同时搜索速度也不是一个级别，达到百万数据/秒的查询数据。总结优点为：实时搜索、稳定、可靠、快速、安装方便等。

        ElasticSearch中的概念（或叫做结构）和数据库中进行对比：

        （1）index：索引，相当于数据库中的一个库，里面可以建立很多表，存储不同类型的数据

        （2）type：类型，相当于数据库中的一张表，存储json类型的数据

        （3）document：文档，一个文档相当于数据库中的一条数据

        （4）field：列，相当于数据库中的列，也就是一个属性

        （5）shards：分片，通俗理解，就是数据分成几块区域来存储，可以理解为mysql中的分库分表(不太恰当)

        （6）replicas：备份，就是分片的备份数,相当于数据库中的备份库

二、ElasticSearch+nodejs+head集成安装（安装环境：jdk8，window10）

        1、先安装ElasticSearch

            （注意：ElasticSearch目前最新版本达到了7.0.1，但不同的版本在不同的安装环境下回出现不同的兼容性问题，详见百度，目前遇到的问题有：用7.0.1版本时无法设置network.host为0.0.0.0，这样见无法让外部服务去访问当前服务器，故改成现在的6.7.2是可以的，并且，ElasticSearch是Java项目，需要依赖jdk，并且对jdk版本有要求，目前安装的是jdk8。另外，安装es服务时安装路径最好不要包含空格，否则在做通过logstash数据同步的时候会报莫名的错误）

            （1）安装：具体安装为在官网https://www.elastic.co/cn/downloads/elasticsearch#ga-release下载相应的版本后如：elasticsearch-6.7.2.zip，进行解压即可。

            （2）启动：到elasticsearch-6.7.2\bin目录下，启动elasticsearch.bat即可

            （3）判断是否安装启动成功：访问http://localhost:9200，出现如下图即为成功

               https://oscimg.oschina.net/oscnet/5e1b674e33b26fc3fed57cebf66eab3ade2.jpg

        2、安装nodejs

            （1）在https://nodejs.org/en/download/ 下载对应的版本，进行傻瓜式安装（默认安装成功后自动配置环境变量）

            （2）安装成功后使用 node -v查看nodejs的版本，使用npm -v查看npm的版本，如图：

                        https://oscimg.oschina.net/oscnet/da041197035cb1a6e13d26a44604cfdf750.jpg

            （3）在nodejs的根目录下，执行npm install -g grunt-cli命令，安装grunt，安装完成后执行grunt -version查看是否安装成功，如图：

                        https://oscimg.oschina.net/oscnet/e2299f56c609bfbeb044496a61ce40f3ac5.jpg

            3、安装head

                （1）网上下载elasticsearch-head.zip文件解压即可

                （2）修改elasticsearch-head文件目录下的Gruntfile.js文件，添加如下内容：

                          https://oscimg.oschina.net/oscnet/063c80c79ba7145b90398a2dd95c9d3be7a.jpg

                （3）修改elasticsearch-head\_site文件目录下的app.js文件内容，将红框中的内容修改为服务器地址，或是本机部署则不用修改。

                       https://oscimg.oschina.net/oscnet/4d20fdf6f76dfb599715f14660466c9ad4b.jpg

                  （4）进入elasticsearch-head的跟目录下执行npm install 命令，如图：

                       https://oscimg.oschina.net/oscnet/dbf7fa1a40c651d3c37000c3938b649f668.jpg

            （5）在elasticsearch-head的根目录下启动nodejs，执行grunt server 或者 npm run start,如图：

                        https://oscimg.oschina.net/oscnet/6e562213ca7cde8e9f75bc190e17152d094.jpg    

            （6）访问http://localhost:9100，出现如图所示，则head+nodejs安装成功

                        https://oscimg.oschina.net/oscnet/be9278b9c5ca1ef544226a5069af04acb08.jpg

               （7）、若想让head启动并链接elasticsearch成功，需先启动elasticsearch，若head和elasticsearch不在同一服务器上时，需要在elasticsearch中做如下配置：

                            修改elasticsearch.yml文件，在文件末尾加入：

                            http.cors.enabled: true 
                            http.cors.allow-origin: "*"
                            node.master: true
                            node.data: true

                            放开network.host: 192.168.0.1的注释并改为network.host: 0.0.0.0（这样外部服务器也能访问es服务）

                            放开cluster.name（集群名称，服务启动前修改后，以后不要再随意修改）；node.name（集群使用时回用到）；http.port（默认端口号即可）的注释

                            双击elasticsearch.bat重启es

                           修改完elasticsearch.yml文件完启动es后再访问http://localhost:9100，若head链接es成功后如下图：

                            https://oscimg.oschina.net/oscnet/d67a1c07a1b320e17b6ef9d3493c4e8dc05.jpg

        4、在head中对es数据进行操作，如，添加删除索引，对索引中的数据进行增删改差等操作。（详见百度）

三、springboot+elasticsearch集成及简单使用

       1、先搭建好单独的springboot项目，以及部署好es服务（这里我是在之前搭建的SpringBoot多模块项目集成es）

       2、相关依赖

  1. <!-- SpringBoot与elasticsearch整合的相关依赖 -->
  2. <dependency>
  3.      <groupId>org.springframework.boot</groupId>
  4.      <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
  5. </dependency>

      3、配置文件

  1. #增加elasticsearch配置
  2. spring.data.elasticsearch.repositories.enabled = true
  3. spring.data.elasticsearch.cluster-nodes=127.0.0.1:9300
  4. spring.data.elasticsearch.cluster-name=my-application-111

      4、相关代码

    Story
  1. package net.cnki.es.entity;
  2.  
  3. import org.springframework.data.annotation.Id;
  4. import org.springframework.data.elasticsearch.annotations.Document;
  5. import org.springframework.data.elasticsearch.annotations.Field;
  6. import org.springframework.data.elasticsearch.annotations.FieldType;
  7.  
  8. import java.util.Date;
  9.  
  10. @Document(indexName = "story",type="doc",shards=5,replicas=1)
  11. public class Story {
  12.     /**
  13.      * @Document 作用在类，标记实体类为文档对象，一般有两个属性：
  14.      * 		indexName:对应索引库名称
  15.      * 		type:对应索引库中的类型（从其他文档中查看说是相当于关系型数据库中的一个表）
  16.      * 		shards:分片数量，默认5个
  17.      * 		replicas:副本数量，默认1个
  18.      * @Id 作用在成员变量，标记一个字段作为id主键
  19.      * @Field 作用在成员变量，标记为文档的字段，并指定字段映射属性：
  20.      * 		type：字段类型，是枚举：FieldType，可以是text、long、short、date、integer、object等
  21.      * 			text：存储数据时候，会自动分词，并生成索引
  22.      * 			keyword：存储数据时候，不会分词建立索引
  23.      * 			Numerical：数值类型，分两类
  24.      * 				基本数据类型：long、interger、short、byte、double、float、half_float
  25.      * 				浮点数的高精度类型：scaled_float
  26.      * 					需要指定一个精度因子，比如10或100。elasticsearch会把真实值乘以这个因子后存储，取出时再还原。
  27.      * 			Date：日期类型
  28.      * 				elasticsearch可以对日期格式化为字符串存储，但是建议我们存储为毫秒值，存储为long，节省空间。
  29.      * 		index：是否索引，布尔类型，默认是true
  30.      * 		store：是否存储，布尔类型，默认是false
  31.      * 		analyzer：分词器名称，这里的ik_max_word即使用ik分词器
  32.      * 				IK分词器有两种类型，分别是ik_smart分词器和ik_max_word分词器。
  33.      * 				ik_smart: 会做最粗粒度的拆分，比如会将“中华人民共和国国歌”拆分为“中华人民共和国,国歌”。
  34.      * 				ik_max_word: 会将文本做最细粒度的拆分，比如会将“中华人民共和国国歌”拆分为“中华人民共和国,中华人民,中华,华人,人民共和国,人民,人,民,共和国,共和,和,国国,国歌”，会穷尽各种可能的组合；
  35.      */
  36.     @Id
  37.     private Integer id;
  38.     @Field(type= FieldType.text,analyzer = "ik_smart",searchAnalyzer = "ik_smart")
  39.     private String volume;
  40.     @Field(type=FieldType.text,analyzer = "ik_smart",searchAnalyzer = "ik_smart")
  41.     private String chapter;
  42.     @Field(type=FieldType.text,analyzer = "ik_smart",searchAnalyzer = "ik_smart")
  43.     private String content;
  44.     @Field(type=FieldType.Date)
  45.     private Date createdate;
  46.  
  47.     /**
  48.      * @return id
  49.      */
  50.     public Integer getId() {
  51.         return id;
  52.     }
  53.  
  54.     /**
  55.      * @param id
  56.      */
  57.     public void setId(Integer id) {
  58.         this.id = id;
  59.     }
  60.  
  61.     public String getVolume() {
  62.         return volume;
  63.     }
  64.  
  65.     public void setVolume(String volume) {
  66.         this.volume = volume;
  67.     }
  68.  
  69.     public String getChapter() {
  70.         return chapter;
  71.     }
  72.  
  73.     public void setChapter(String chapter) {
  74.         this.chapter = chapter;
  75.     }
  76.  
  77.     public String getContent() {
  78.         return content;
  79.     }
  80.  
  81.     public void setContent(String content) {
  82.         this.content = content;
  83.     }
  84.  
  85.     public Date getCreatedate() {
  86.         return createdate;
  87.     }
  88.  
  89.     public void setCreatedate(Date createdate) {
  90.         this.createdate = createdate;
  91.     }
  92.  
  93.     @Override
  94.     public String toString() {
  95.         return "Story{" +
  96.                 "id=" + id +
  97.                 ", volume='" + volume + '\'' +
  98.                 ", chapter='" + chapter + '\'' +
  99.                 ", content='" + content + '\'' +
  100.                 ", createdate=" + createdate +
  101.                 '}';
  102.     }
  103. }

        StoryService

  1. package net.cnki.es.service;
  2.  
  3. import com.github.pagehelper.PageInfo;
  4. import net.cnki.es.entity.Story;
  5. import org.springframework.data.domain.Page;
  6.  
  7. import java.util.Iterator;
  8. import java.util.List;
  9. import java.util.Map;
  10.  
  11.  
  12.  
  13. public interface StoryService{
  14.  
  15.  
  16.     /**
  17.      * 多条件模糊分页查询
  18.      * @param searchMap
  19.      * 		   查询条件以searchMap中key、value形式传入，其中key为要查询的字段名称，value为查询字段值
  20.      * @param ordersMap
  21.      * 		  排序条件以ordersMap中key、value形式传入，其中key为排序的字段名称，value为排序规则：desc/asc
  22.      * @param pageUtil
  23.      * 		  分页对象信息
  24.      * @param clazz
  25.      * @return
  26.      * 		  返回一个map数据，包含：
  27.      * 		 map.("data",dataListInfo<T>);
  28.      * 		 map.("pageEsInfo",pageEsInfo);
  29.      */
  30.     <T> Map<String,Object> queryByMapParamsForPage(Map<String,Object> searchMap, Map<String, Object> ordersMap, PageInfo<T> pageUtil, Class<T> clazz);
  31.  
  32.     /**
  33.      * 多条件模糊高亮显示分页查询
  34.      * @param searchMap
  35.      * 		   查询条件以searchMap中key、value形式传入，其中key为要查询的字段名称，value为查询字段值
  36.      * @param ordersMap
  37.      * 		   排序条件以ordersMap中key、value形式传入，其中key为排序的字段名称，value为排序规则：desc/asc
  38.      * @param pageUtil
  39.      * 		   分页对象信息
  40.      * @param clazz
  41.      * @param color
  42.      * 		   设置高亮显示的颜色
  43.      * @return
  44.      * 		   返回一个map数据，包含：
  45.      * 		  map.("data",dataListInfo<T>);
  46.      * 		  map.("pageEsInfo",pageEsInfo);
  47.      */
  48.     <T> Map<String,Object> highLigthQueryForPage(Map<String,Object> searchMap, Map<String, Object> ordersMap, PageInfo<T> pageUtil, Class<T> clazz, String color);
  49.  
  50.     /**
  51.      * 根据id删除es服务器上的数据（由于MySQL数据库同步数据到es时，无法同步MySQL中删除的历史数据，故需在es服务器上自行删除）
  52.      * @param id
  53.      * @param clazz
  54.      * @return
  55.      */
  56.     <T> String deleteEsDataById(String id, Class<T> clazz);
  57.     /**
  58.      * 新增数据到es服务器
  59.      * @param id
  60.      * 		  当前数据的id值
  61.      * @param modelJson
  62.      * 		  新增对象的实体类，以json字符串格式传入
  63.      * @param index
  64.      * 		  新增对象的索引值
  65.      * @param type
  66.      * 		  新增对象的类型值
  67.      * @return
  68.      */
  69.     String addEsData(String id, String modelJson, String index, String type);
  70.     /**
  71.      * 更新数据到es服务器
  72.      * @param id
  73.      * 		  当前数据的id值
  74.      * @param modelMap
  75.      * 		  更新对象属性值，以map格式传入
  76.      * @param index
  77.      * 		  更新对象的索引值
  78.      * @param type
  79.      * 		  更新对象的类型值
  80.      * @param clazz
  81.      * 		  更新对象类
  82.      * @return
  83.      */
  84.     <T> String updEsData(String id, Map<String,Object> modelMap, String index, String type, Class<T> clazz);
  85.     /**
  86.      * 判断es服务器中是否有该id
  87.      * @param id
  88.      * @return
  89.      */
  90.     String  getIdById(String id);
  91.  
  92.     /**
  93.      * 保存单条记录
  94.      * @param story
  95.      */
  96.     void save(Story story);
  97.     /**
  98.      * 保存多条记录
  99.      * @param list
  100.      */
  101.     void saveAll(List<Story> list);
  102.  
  103.     /**
  104.      * 查询所有记录
  105.      * @return
  106.      */
  107.     Iterator<Story> findAll();
  108.  
  109.     Page<Story> findByContent(String content);
  110.  
  111.     Page<Story> findByFirstCode(String firstCode);
  112.  
  113.     Page<Story> findBySecordCode(String secordCode);
  114.  
  115.     Page<Story> query(String key);
  116.  
  117.     void  getStoryData();
  118.  
  119. }
StoryServiceImpl
  1. package net.cnki.es.service.impl;
  2.  
  3. import java.net.URL;
  4. import java.util.*;
  5. import java.util.Map.Entry;
  6. import com.github.pagehelper.PageInfo;
  7. import net.cnki.es.dao.EsDataQueryRepository;
  8. import net.cnki.es.entity.Story;
  9. import net.cnki.es.service.StoryService;
  10. import org.elasticsearch.action.search.SearchResponse;
  11. import org.elasticsearch.action.update.UpdateRequest;
  12. import org.elasticsearch.action.update.UpdateResponse;
  13. import org.elasticsearch.index.query.BoolQueryBuilder;
  14. import org.elasticsearch.index.query.QueryBuilder;
  15. import org.elasticsearch.index.query.QueryBuilders;
  16. import org.elasticsearch.search.SearchHit;
  17. import org.elasticsearch.search.SearchHits;
  18. import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
  19. import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder.Field;
  20. import org.jsoup.Jsoup;
  21. import org.jsoup.nodes.Document;
  22. import org.jsoup.nodes.Element;
  23. import org.jsoup.select.Elements;
  24. import org.springframework.beans.factory.annotation.Autowired;
  25. import org.springframework.data.domain.Page;
  26. import org.springframework.data.domain.PageRequest;
  27. import org.springframework.data.domain.Pageable;
  28. import org.springframework.data.domain.Sort;
  29. import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
  30. import org.springframework.data.elasticsearch.core.SearchResultMapper;
  31. import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage;
  32. import org.springframework.data.elasticsearch.core.aggregation.impl.AggregatedPageImpl;
  33. import org.springframework.data.elasticsearch.core.query.IndexQuery;
  34. import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
  35. import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
  36. import org.springframework.data.elasticsearch.core.query.SearchQuery;
  37. import org.springframework.data.elasticsearch.core.query.UpdateQuery;
  38. import org.springframework.stereotype.Service;
  39.  
  40. import com.alibaba.druid.util.StringUtils;
  41.  
  42. @Service
  43. public class StoryServiceImpl implements StoryService {
  44.  
  45.     @Autowired
  46.     private ElasticsearchTemplate elasticsearchTemplate;
  47.  
  48.     @Autowired
  49.     private EsDataQueryRepository esRepository;
  50.  
  51.     private Pageable pageable = PageRequest.of(0,10);
  52.  
  53.  
  54.     @SuppressWarnings({ "deprecation", "static-access" })
  55.     @Override
  56.     public <T> Map<String, Object> queryByMapParamsForPage(Map<String, Object> searchMap, Map<String, Object> ordersMap,
  57.                                                            PageInfo<T> pageEsInfo, Class<T> clazz) {
  58.         Map<String, Object> map = new HashMap<String,Object>();
  59.         //封装字段查询条件
  60.         BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
  61.         if(searchMap!=null && !searchMap.isEmpty()) {
  62.             for(Entry<String, Object> vo : searchMap.entrySet()) {
  63.                 String key = vo.getKey();
  64.                 String value = (String) vo.getValue();
  65.                 if(!StringUtils.isEmpty(value)) {
  66.                     boolQueryBuilder = boolQueryBuilder.must(QueryBuilders.matchQuery(key, value));
  67.                 }
  68.             }
  69.         }
  70.         QueryBuilder queryBuilder = boolQueryBuilder;
  71.         SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(queryBuilder).build();
  72.         //获取封装分页查询条件,默认取第一页
  73.         Pageable pageable= null;
  74.         if(pageEsInfo!=null) {
  75.             Integer currentPageNum = pageEsInfo.getPageNum();
  76.             if(currentPageNum!=null&&currentPageNum>0) {
  77.                 currentPageNum = currentPageNum-1;
  78.             }else {
  79.                 map.put("success", false);
  80.                 map.put("msg", "当前页数有误，请输入大于0的正整数");
  81.                 return map;
  82.             }
  83.             Integer pageSize = pageEsInfo.getPageSize();
  84.             if(pageSize==null || pageSize==0) {
  85.                 map.put("success", false);
  86.                 map.put("msg", "每页条数有误，请输入大于0的正整数");
  87.                 return map;
  88.             }
  89.             //获取封装排序条件
  90.             Sort sort = null;
  91.             if(ordersMap!=null && !ordersMap.isEmpty()) {
  92.                 for(Entry<String, Object> vo : ordersMap.entrySet()) {
  93.                     String key = vo.getKey();
  94.                     String value = (String) vo.getValue();
  95.                     if("desc".equals(value)) {
  96.                         sort = sort.by(key).descending();
  97.                     }else {
  98.                         sort = sort.by(key).ascending();
  99.                     }
  100.                 }
  101.             }
  102.             if(sort!=null) {
  103.                 pageable = new PageRequest(currentPageNum, pageSize,sort);
  104.             }else {
  105.                 pageable = new PageRequest(currentPageNum, pageSize);
  106.             }
  107.  
  108.             searchQuery = new NativeSearchQueryBuilder().withQuery(queryBuilder).withPageable(pageable).build();
  109.         }
  110.  
  111.         List<T> list = elasticsearchTemplate.queryForList(searchQuery, clazz);
  112.         Long count = elasticsearchTemplate.count(searchQuery);
  113.         pageEsInfo.setTotal(count.intValue());
  114.         pageEsInfo.setSize(count.intValue());
  115.         map.put("data", list);
  116.         map.put("pageEsInfo", pageEsInfo);
  117.         return map;
  118.     }
  119.  
  120.     @SuppressWarnings({ "deprecation", "static-access" })
  121.     @Override
  122.     public <T> Map<String, Object> highLigthQueryForPage(Map<String, Object> searchMap, Map<String, Object> ordersMap,
  123.                                                          PageInfo<T> pageEsInfo, Class<T> clazz,String color) {
  124.         Map<String,Object> mapRel = new HashMap<String,Object>();
  125.         BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
  126.  
  127.         //封装字段查询条件，及高亮显示字段
  128.         String preTag = "<font color='red'>";//google的色值
  129.         String postTag = "</font>";
  130.         //不传默认红色
  131.         if(!StringUtils.isEmpty(color)) {
  132.             preTag = preTag.replace("red", color);
  133.         }
  134.         int size = 0;
  135.         for(Entry<String, Object> vo : searchMap.entrySet()) {
  136.             if(!StringUtils.isEmpty((String)vo.getValue())) {
  137.                 size = size+1;
  138.             }
  139.         }
  140.         HighlightBuilder.Field[] highlightFields = new Field[size];
  141.  
  142.         if(searchMap!=null && !searchMap.isEmpty()) {
  143.             int i=0;
  144.             for(Entry<String, Object> vo : searchMap.entrySet()) {
  145.                 if(!StringUtils.isEmpty((String)vo.getValue())) {
  146.                     String key = vo.getKey();
  147.                     String value = (String) vo.getValue();
  148.                     if(!StringUtils.isEmpty(value)) {
  149.                         boolQueryBuilder = boolQueryBuilder.must(QueryBuilders.matchQuery(key, value));
  150.  
  151.                         Field field = new HighlightBuilder.Field(key).preTags(preTag).postTags(postTag);
  152.                         highlightFields[i] = field;
  153.                         i = i+1;
  154.                     }
  155.                 }
  156.             }
  157.         }
  158.         NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
  159.                 .withQuery(boolQueryBuilder).withHighlightFields(highlightFields).build();
  160.         //封装分页信息
  161.         Pageable pageable= null;
  162.         if(pageEsInfo!=null) {
  163.             Integer currentPageNum = pageEsInfo.getPageNum();
  164.             if(currentPageNum!=null&&currentPageNum>0) {
  165.                 currentPageNum = currentPageNum-1;
  166.             }else {
  167.                 mapRel.put("success", false);
  168.                 mapRel.put("msg", "当前页数有误，请输入大于0的正整数");
  169.                 return mapRel;
  170.             }
  171.             Integer pageSize = pageEsInfo.getPageSize();
  172.             if(pageSize==null || pageSize==0) {
  173.                 mapRel.put("success", false);
  174.                 mapRel.put("msg", "每页条数有误，请输入大于0的正整数");
  175.                 return mapRel;
  176.             }
  177.             //封装排序条件
  178.             Sort sort = null;
  179.             if(ordersMap!=null && !ordersMap.isEmpty()) {
  180.                 for(Entry<String, Object> vo : ordersMap.entrySet()) {
  181.                     String key = vo.getKey();
  182.                     String value = (String) vo.getValue();
  183.                     if("desc".equals(value)) {
  184.                         sort = sort.by(key).descending();
  185.                     }else {
  186.                         sort = sort.by(key).ascending();
  187.                     }
  188.                 }
  189.             }
  190.             if(sort!=null) {
  191.                 pageable = new PageRequest(currentPageNum, pageSize,sort);
  192.             }else {
  193.                 pageable = new PageRequest(currentPageNum, pageSize);
  194.             }
  195.  
  196.             searchQuery = new NativeSearchQueryBuilder()
  197.                     .withQuery(boolQueryBuilder).withPageable(pageable)
  198.                     .withHighlightFields(highlightFields).build();
  199.         }
  200.  
  201.  
  202.  
  203.         Page<T> page = elasticsearchTemplate.queryForPage(searchQuery, clazz, new SearchResultMapper() {
  204.  
  205.             @SuppressWarnings({ "unchecked", "hiding" })
  206.             @Override
  207.             public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> clazz, Pageable pageable) {
  208.                 ArrayList<Map<String,Object>> poems = new ArrayList<Map<String,Object>>();
  209.                 SearchHits hits = response.getHits();
  210.                 for (SearchHit searchHit : hits) {
  211.                     if (hits.getHits().length <= 0) {
  212.                         return null;
  213.                     }
  214.                     Map<String,Object> mapResult = searchHit.getSourceAsMap();
  215.                     for(Entry<String, Object> vo : searchMap.entrySet()) {
  216.                         String key = vo.getKey();
  217.                         //将高亮内容设置进去
  218.                         String valueSearch = (String) vo.getValue();
  219.                         if(!StringUtils.isEmpty(valueSearch)) {
  220.                             String highLightMessage = searchHit.getHighlightFields().get(key).fragments()[0].toString();
  221.                             mapResult.put(key, highLightMessage);
  222.                         }
  223.  
  224.                     }
  225.                     poems.add(mapResult);
  226.                 }
  227.                 if (poems.size() > 0) {
  228.                     return new AggregatedPageImpl<T>((List<T>) poems);
  229.                 }
  230.                 return null;
  231.             }
  232.         });
  233.  
  234.         List<T> poems = new ArrayList<T>();
  235.         if(page!=null) {
  236.             poems = (List<T>) page.getContent();
  237.         }
  238.         Long count = elasticsearchTemplate.count(searchQuery);
  239.         int num = (int) count.longValue();
  240.         pageEsInfo.setTotal(count.intValue());
  241.         pageEsInfo.setSize(count.intValue());
  242.         //pageEsInfo.setPages(count.intValue());
  243.         if(num%pageEsInfo.getPageSize()==0){
  244.             pageEsInfo.setPages(num%pageEsInfo.getPageSize());
  245.         }else {
  246.             pageEsInfo.setPages(num%pageEsInfo.getPageSize()+1);
  247.         }
  248.         pageEsInfo.setList(poems);
  249.         mapRel.put("data", poems);
  250.         mapRel.put("pageEsInfo", pageEsInfo);
  251.  
  252.         return mapRel;
  253.     }
  254.  
  255.     @Override
  256.     public <T> String deleteEsDataById(String id, Class<T> clazz) {
  257.         String delId = elasticsearchTemplate.delete(clazz, id);
  258.         return delId;
  259.     }
  260.  
  261.     @Override
  262.     public String addEsData(String id, String modelJson, String index, String type) {
  263.         IndexQuery indexQuery = new IndexQuery();
  264.         indexQuery.setId(id);
  265.         indexQuery.setSource(modelJson);
  266.         indexQuery.setIndexName(index);
  267.         indexQuery.setType(type);
  268.         String  strId = elasticsearchTemplate.index(indexQuery);
  269.         return strId;
  270.     }
  271.  
  272.     @Override
  273.     public <T> String updEsData(String id, Map<String, Object> modelMap, String index, String type, Class<T> clazz) {
  274.         UpdateQuery updateQuery = new UpdateQuery();
  275.         updateQuery.setId(id);
  276.         updateQuery.setClazz(clazz);
  277.         updateQuery.setIndexName(index);
  278.         updateQuery.setType(type);
  279.  
  280.         UpdateRequest updateRequest = new UpdateRequest();
  281.         updateRequest.doc(modelMap);
  282.         updateQuery.setUpdateRequest(updateRequest);
  283.  
  284.         UpdateResponse up = elasticsearchTemplate.update(updateQuery);
  285.  
  286.         return up.getId();
  287.     }
  288.  
  289.     @Override
  290.     public String getIdById(String id) {
  291.         // 构造搜索条件
  292.         SearchQuery searchQuery = new NativeSearchQueryBuilder().withQuery(QueryBuilders.boolQuery().must(QueryBuilders.matchQuery("id", id))).build();
  293.         List<String> ids = elasticsearchTemplate.queryForIds(searchQuery);
  294.         if(ids!=null&&ids.size()>0) {
  295.             return ids.get(0);
  296.         }
  297.         return null;
  298.     }
  299.  
  300.     @Override
  301.     public void save(Story docBean) {
  302.         esRepository.save(docBean);
  303.     }
  304.  
  305.     @Override
  306.     public void saveAll(List<Story> list) {
  307.         esRepository.saveAll(list);
  308.     }
  309.  
  310.     @Override
  311.     public Iterator<Story> findAll() {
  312.         return esRepository.findAll().iterator();
  313.     }
  314.  
  315.     @Override
  316.     public Page<Story> findByContent(String content) {
  317.         return esRepository.findByContent(content,pageable);
  318.     }
  319.  
  320.     @Override
  321.     public Page<Story> findByFirstCode(String firstCode) {
  322.         return esRepository.findByFirstCode(firstCode,pageable);
  323.     }
  324.  
  325.     @Override
  326.     public Page<Story> findBySecordCode(String secordCode) {
  327.         return esRepository.findBySecordCode(secordCode,pageable);
  328.     }
  329.  
  330.     @Override
  331.     public Page<Story> query(String key) {
  332.         return esRepository.findByContent(key,pageable);
  333.     }
  334.  
  335.     /**
  336.      * 爬取小说数据
  337.      * 小说名：剑来
  338.      */
  339.     @Override
  340.     public void  getStoryData()  {
  341.         try {
  342.             int htmlPages = 2324764;
  343.             int id = 1;
  344.             String urlstart = "http://www.shuquge.com/txt/8659/";
  345.             String url = "http://www.shuquge.com/txt/8659/2324752.html";
  346.             String volume = "第一卷 笼中雀";
  347.             while (htmlPages<=38147404) {
  348.                 //拼接url
  349.                 //url = urlstart + htmlPages + ".html";
  350.                 Document document = Jsoup.parse(new URL(url), 30000);
  351.                 //获取小说内容元素
  352.                 Elements element = document.getElementsByClass("content");
  353.                 //获取结果集的第一个元素
  354.                 Element li = element.first();
  355.                 //章节
  356.                 String chapter = li.getElementsByTag("h1").eq(0).html();
  357.                 //内容
  358.                 String content = li.getElementById("content").html();
  359.                 Story story = new Story();
  360.                 //小说详情页没有第几卷，根据章节判断
  361.                 if (url.contains("2324837")) {
  362.                     volume = "第二卷 山水郎";
  363.                 }
  364.                 story.setId(id);
  365.                 story.setVolume(volume);
  366.                 story.setChapter(chapter);
  367.                 story.setContent(content);
  368.                 Date date = new Date();
  369.                 story.setCreatedate(date);
  370.                 System.out.println(story.toString());
  371.                 //插入es
  372.                 save(story);
  373.                 //获取下一章节地址
  374.                 Element nextChapterElement = document.getElementsByClass("page_chapter").first();
  375.                 String nextChapter = nextChapterElement.getElementsByTag("a").eq(2).attr("href");
  376.                 if(nextChapter.contains("index.html")){
  377.                     return;
  378.                 }
  379.                 url = urlstart + nextChapter;
  380.                 htmlPages++;
  381.                 id++;
  382.             }
  383.         }catch(Exception e){
  384.         }
  385.     }
  386. }

EsDataQueryRepository
  1. package net.cnki.es.dao;
  2.  
  3. import net.cnki.es.entity.Story;
  4. import org.springframework.data.domain.Page;
  5. import org.springframework.data.domain.Pageable;
  6. import org.springframework.data.elasticsearch.annotations.Query;
  7. import org.springframework.data.elasticsearch.repository.ElasticsearchRepository;
  8.  
  9. public interface EsDataQueryRepository  extends ElasticsearchRepository<Story, Long> {
  10.     //默认的注释
  11.     //@Query("{\"bool\" : {\"must\" : {\"field\" : {\"content\" : \"?\"}}}}")
  12.     Page<Story> findByContent(String content, Pageable pageable);
  13.  
  14.     @Query("{\"bool\" : {\"must\" : {\"field\" : {\"firstCode.keyword\" : \"?\"}}}}")
  15.     Page<Story> findByFirstCode(String firstCode, Pageable pageable);
  16.  
  17.     @Query("{\"bool\" : {\"must\" : {\"field\" : {\"secordCode.keyword\" : \"?\"}}}}")
  18.     Page<Story> findBySecordCode(String secordCode, Pageable pageable);
  19. }

StoryController
  1. package net.cnki.es.controller;
  2.  
  3. import com.alibaba.fastjson.JSON;
  4. import com.github.pagehelper.PageHelper;
  5. import com.github.pagehelper.PageInfo;
  6. import net.cnki.es.entity.Story;
  7. import net.cnki.es.service.StoryService;
  8. import org.springframework.beans.factory.annotation.Autowired;
  9. import org.springframework.stereotype.Controller;
  10. import org.springframework.util.StringUtils;
  11. import org.springframework.web.bind.annotation.GetMapping;
  12. import org.springframework.web.bind.annotation.RequestMapping;
  13. import org.springframework.web.bind.annotation.RequestParam;
  14. import org.springframework.web.bind.annotation.ResponseBody;
  15.  
  16. import java.util.Date;
  17. import java.util.HashMap;
  18. import java.util.Iterator;
  19. import java.util.Map;
  20.  
  21. @Controller
  22. @RequestMapping("es")
  23. public class StoryController {
  24.     @Autowired
  25.     private StoryService storyService;
  26.  
  27.     /**
  28.      * 查询页面
  29.      * @return
  30.      */
  31.     @GetMapping("/search")
  32.     public String searchPage() {
  33.         //ClassUtils.getDefaultClassLoader().getResource("templates").getPath()+
  34.         return "pages/elasticsearch/search.html";
  35.     }
  36.     /**
  37.      * 查询所有数据
  38.      */
  39.     @RequestMapping("/all")
  40.     @ResponseBody
  41.     public Iterator<Story> all(){
  42.         return storyService.findAll();
  43.     }
  44.  
  45.     /**
  46.      * 抓取网页数据
  47.      */
  48.     @RequestMapping("/getStoryData")
  49.     public void getStoryData(){
  50.         storyService.getStoryData();
  51.     }
  52.  
  53.     /**
  54.      * 分页查询
  55.      * @return
  56.      */
  57.     @RequestMapping("getStoryList")
  58.     @ResponseBody
  59.     public PageInfo getStoryList(@RequestParam(value = "limit",defaultValue = "10")Integer limit,
  60.                                  @RequestParam(value = "page",defaultValue = "1")Integer page,String content,String chapter) {
  61.         Map<String,Object> mapParams = new HashMap<String,Object>();
  62.         mapParams.put("content", content);
  63.         mapParams.put("chapter", chapter);
  64.         PageInfo<Story> pageInfo = new PageInfo<Story>();
  65.         PageHelper.startPage(page,limit);
  66.         pageInfo.setPageNum(page);
  67.         pageInfo.setPageSize(limit);
  68.  
  69.         Map<String,Object> mapOrders = new HashMap<String,Object>();
  70.         mapOrders.put("id", "desc");
  71.  
  72.         Map<String,Object> map = new HashMap<String,Object>();
  73.         map= storyService.highLigthQueryForPage(mapParams, mapOrders, pageInfo, Story.class,"");
  74.         return (PageInfo)map.get("pageEsInfo");
  75.     }
  76.  
  77.  
  78.     /**
  79.      * 分页查询
  80.      */
  81.     @RequestMapping("searchByPage")
  82.     @ResponseBody
  83.     public String getSyslogListTest2() {
  84.  
  85.         Map<String,Object> mapParams = new HashMap<String,Object>();
  86.         mapParams.put("content", "平安");
  87.  
  88.         PageInfo<Story> pageInfo = new PageInfo<Story>();
  89.         pageInfo.setPageNum(3);
  90.         pageInfo.setPageSize(5);
  91.  
  92.         Map<String,Object> mapOrders = new HashMap<String,Object>();
  93.         mapOrders.put("_id", "desc");
  94.  
  95.         Map<String,Object> map = new HashMap<String,Object>();
  96.         map = storyService.queryByMapParamsForPage(mapParams, mapOrders, pageInfo, Story.class);
  97.         return JSON.toJSONString(map);
  98.     }
  99.  
  100.     /**
  101.      * 高亮分页查询
  102.      * @return
  103.      */
  104.     @RequestMapping("searchByHighPage")
  105.     @ResponseBody
  106.     public String getSyslogListTest1() {
  107.         Map<String,Object> mapParams = new HashMap<String,Object>();
  108.         mapParams.put("content", "东");
  109.  
  110.         PageInfo<Story> pageInfo = new PageInfo<Story>();
  111.         pageInfo.setPageNum(1);
  112.         pageInfo.setPageSize(5);
  113.  
  114.         Map<String,Object> mapOrders = new HashMap<String,Object>();
  115.         mapOrders.put("id", "desc");
  116.  
  117.         Map<String,Object> map = new HashMap<String,Object>();
  118.         map = storyService.highLigthQueryForPage(mapParams, mapOrders, pageInfo, Story.class,"#53FF53");
  119.         return JSON.toJSONString(map);
  120.     }
  121.  
  122.     /**
  123.      * 删除记录
  124.      * @return
  125.      */
  126.     @RequestMapping("delById")
  127.     @ResponseBody
  128.     public String delSyslogByIdTest() {
  129.         String id = "1";
  130.         if(!StringUtils.isEmpty(id)) {
  131.             String delId = storyService.deleteEsDataById(id, Story.class);
  132.             if(id.equals(delId)) {
  133.                 return "true";
  134.             }else {
  135.                 return "false";
  136.             }
  137.         }else {
  138.             return "false";
  139.         }
  140.     }
  141.  
  142.     /**
  143.      * 添加记录
  144.      * @return
  145.      */
  146.     @RequestMapping("addSyslog")
  147.     @ResponseBody
  148.     public String addSyslogInfoTest() {
  149.         Story sysLog = new Story();
  150.         sysLog.setId(100);
  151.         sysLog.setVolume("skq");
  152.         sysLog.setChapter("手动添加");
  153.         sysLog.setContent("shoudong");
  154.         sysLog.setCreatedate(new Date());
  155.         String modelJson = JSON.toJSONString(sysLog);
  156.         String id = sysLog.getId().toString();
  157.         String strId = storyService.addEsData(id, modelJson, "syslog", "syslog");
  158.         return strId;
  159.     }
  160.  
  161.     /**
  162.      * 更新记录
  163.      * @return
  164.      */
  165.     @RequestMapping("updSyslog")
  166.     @ResponseBody
  167.     public String updSyslogInfoTest() {
  168.         Map<String,Object> modelMap = new HashMap<String,Object>();
  169.         modelMap.put("id", 100);
  170.         modelMap.put("user_name", "skq12");
  171.         modelMap.put("type", "sd");
  172.         String strId = storyService.updEsData("100", modelMap, "syslog", "syslog", Story.class);
  173.         return strId;
  174.     }
  175. }

           5、测试es全文检索，在service中写了爬取剑来小说的方法，将爬取的每一章内容直接存到es中，效果如下图所示：

 

 

 

四、利用logstash实现MySQL中的数据全量/增量同步到elasticsearch服务器中（window10环境）

        1、logstash的下载安装（logstash的安装时路径不要包含空格，某种在数据同步的时候会数据同步不成功）

             logstash的官网下载地址：https://www.elastic.co/downloads/logstash，注意下载的版本要与elasticsearch版本必须一直，如当前elasticsearch的版本是6.7.2，则logstash的版本也必须是6.7.2

            注意，在网上各种查找资料发现好多资料都需集成logstash-jdbc-input插件才能实现数据同步，后台才发现这个和版本有关系，在elasticsearch5.X及之后的版本是不需要集成该插件即可

            下载好相应的版本后解压即可（注意解压的路径，最好不要有中文和空格）。

        2、logstash配置

           （1）在logstash-6.7.2路径下创建空文件夹，如mysql，用来存放相关配置文件等

           （2）在创建的新文件夹中（mysql文件夹）中放入驱动包：mysql-connector-java.jar

           （3）在创建的新文件夹中（mysql文件夹）中创建一个sql文件，如find.sql，从这里开始，就是logstash同步数据库的核心操作了，在这里创建的sql文件主要内容是：mysql需要同步Elasticsearch的具体数据的查询方式，如果是全量同步，只需要select * from [table]即可

            （4）在创建的新文件夹中（mysql文件夹）中创建一个conf文件，如jdbc.conf文件，该文件用于链接数据库和elasticsearch，其内容为：（注意，有的可能不识别注解，运行时需要将注解去掉）

  1. input {
  2.    stdin {
  3.     }
  4.     jdbc {
  5.       jdbc_connection_string => "jdbc:mysql://localhost:3306/quick_platform?characterEncoding=utf8&useSSL=false&serverTimezone=UTC"
  6.       jdbc_user => "root"
  7.       jdbc_password => "cnki_TTOD_7"
  8.       jdbc_driver_library => "D:\ELK\logstash-6.7.2\mysql\mysql-connector-java-8.0.11.jar"
  9.       jdbc_driver_class => "com.mysql.cj.jdbc.Driver"
  10.       jdbc_paging_enabled => "true"
  11.       jdbc_page_size => "50000"
  12. 	  clean_run => false
  13.       jdbc_default_timezone => "Asia/Shanghai"
  14.       statement => "SELECT * FROM story where createdate > :sql_last_value order by createdate desc"
  15.       schedule => "* * * * *"
  16.       type => "sy"
  17.       lowercase_column_names => false
  18.       record_last_run => true
  19.       use_column_value => true
  20.       tracking_column => "createdate"
  21.       tracking_column_type => "timestamp"
  22.       last_run_metadata_path => "D:\ELK\logstash-6.7.2\mysql\lw_last_time"
  23.       clean_run => false
  24.     }
  25. }
  26. filter {
  27.     json {
  28.         source => "message"
  29.         remove_field => ["message"]
  30.     }
  31. }
  32. output {
  33.   elasticsearch {
  34.         hosts => "127.0.0.1:9200"
  35.         index => "mysql_story"
  36.     }
  37.     stdout {
  38.         codec => json_lines
  39.     }
  40. }

网上的第二中配置，部分不同而已：

  1. input {
  2.  
  3.     stdin {
  4.  
  5.     }
  6.  
  7.     jdbc {
  8.  
  9.       jdbc_connection_string => "jdbc:mysql://localhost:3306/你的数据库名字"
  10.  
  11.  
  12.  
  13.         jdbc_user => "你的数据库用户名"
  14.  
  15.  
  16.  
  17.         jdbc_password => "你的数据库密码"
  18.  
  19.  
  20.  
  21.         jdbc_driver_library => "C:/logstash/bin/mysql-connector-java-5.1.44-bin.jar"
  22.  
  23.  
  24.  
  25.         jdbc_driver_class => "com.mysql.jdbc.Driver"
  26.  
  27.  
  28.  
  29.         jdbc_paging_enabled => "true"
  30.  
  31.  
  32.  
  33.         jdbc_page_size => "50000"
  34.  
  35.         statement => "SELECT id（主键）,其他内容 FROM 你的表"
  36.  
  37.         schedule => "* * * * *"
  38.  
  39.     }
  40.  
  41. }
  42.  
  43.  
  44.  
  45. output {
  46.  
  47.      stdout {
  48.  
  49.         codec => json_lines
  50.  
  51.     }
  52.  
  53.     elasticsearch {
  54.  
  55.         hosts => "localhost:9200"
  56.  
  57.         index => "你要创建的索引名"
  58.  
  59.         document_type => "你要创建的索引类型"
  60.  
  61.         document_id => "%{id}"
  62.  
  63.     }
  64.  
  65. }

另一个说明比较详细的版本：

  1. input {
  2.  
  3.     stdin {
  4.  
  5.     }
  6.  
  7.     jdbc {
  8.  
  9.       # 连接的数据库地址和哪一个数据库，指定编码格式，禁用SSL协议，设定自动重连
  10.  
  11.       jdbc_connection_string => "jdbc:mysql://数据库地址:端口号/数据库名?characterEncoding=UTF-8&useSSL=false&autoReconnect=true"
  12.  
  13.       # 你的账户密码
  14.  
  15.       jdbc_user => "账号"
  16.  
  17.       jdbc_password => "密码"
  18.  
  19.       # 连接数据库的驱动包，建议使用绝对地址
  20.  
  21.       jdbc_driver_library => "mysql/mysql-connector-java-5.1.45-bin.jar"
  22.  
  23.       # 这是不用动就好
  24.  
  25.       jdbc_driver_class => "com.mysql.jdbc.Driver"
  26.  
  27.       jdbc_paging_enabled => "true"
  28.  
  29.       jdbc_page_size => "50000"
  30.  
  31.  
  32.  
  33.     #处理中文乱码问题
  34.  
  35.       codec => plain { charset => "UTF-8"}
  36.  
  37.  
  38.  
  39.        #使用其它字段追踪，而不是用时间
  40.  
  41.       use_column_value => true
  42.  
  43.        #追踪的字段      
  44.  
  45.     tracking_column => testid      
  46.  
  47.     record_last_run => true     
  48.  
  49.     #上一个sql_last_value值的存放文件路径, 必须要在文件中指定字段的初始值     
  50.  
  51.     last_run_metadata_path => "mysql/station_parameter.txt"
  52.  
  53.  
  54.  
  55.       jdbc_default_timezone => "Asia/Shanghai"
  56.  
  57.  
  58.  
  59.       statement_filepath => "mysql/jdbc.sql"
  60.  
  61.       
  62.  
  63.     #是否清除 last_run_metadata_path 的记录,如果为真那么每次都相当于从头开始查询所有的数据库记录
  64.  
  65.     clean_run => false
  66.  
  67.  
  68.  
  69.       # 这是控制定时的，重复执行导入任务的时间间隔，第一位是分钟
  70.  
  71.       schedule => "*/5 * * * *"
  72.  
  73.       type => "jdbc"
  74.  
  75.     }
  76.  
  77. }
  78.  
  79.  
  80.  
  81. filter {
  82.  
  83.     json {
  84.  
  85.         source => "message"
  86.  
  87.         remove_field => ["message"]
  88.  
  89.     }
  90.  
  91. }
  92.  
  93.  
  94.  
  95. output {
  96.  
  97.     elasticsearch {
  98.  
  99.         # 要导入到的Elasticsearch所在的主机
  100.  
  101.         hosts => "192.168.105.180:9200"
  102.  
  103.         # 要导入到的Elasticsearch的索引的名称
  104.  
  105.         index => "db_anytest"
  106.  
  107.         # 类型名称（类似数据库表名）
  108.  
  109.         document_type => "table_anytest"
  110.  
  111.         # 主键名称（类似数据库主键）
  112.  
  113.         document_id => "%{testid}"
  114.  
  115.         # es 账号
  116.  
  117.         user => elastic
  118.  
  119.         password => changeme
  120.  
  121.         
  122.  
  123.     }
  124.  
  125.  
  126.  
  127.     stdout {
  128.  
  129.         # JSON格式输出
  130.  
  131.         codec => json_lines
  132.  
  133.     }
  134.  
  135. }

        3、启动logstash开始同步数据库

              （1）确保elasticsearch服务已启动，并且要同步的表里有相应的数据

              （2）cmd一个新窗口，进入到D:\logstash-6.7.2\bin

              （3）运行命令logstash -f ../mysql/jdbc.conf  ，其中logstash -f表示运行指令， ../mysql/jdbc.conf表示我们配置的jdbc.conf文件路径，成功启动后，可以在终端中看见运行的sql和同步的数据，如图：https://oscimg.oschina.net/oscnet/22c18bfb83c90d6361283daeb10b2a038f9.jpg

               （4）同步成功后即可在head中查看同步到elasticsearch中的数据，其中timestamp和version是elastisearch自己添加的字段。​​​​​​​

                （5）注意点：

                        a、在同步的时候，如果是首次全量同步的话，可以不需要在elasticsearch中去新建索引和类型，同步的时候会根据配置自动创建

                        b、若是增量更新的话，在sql中添加查询条件即可，如

                    where  testid >= :sql_last_start

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/一键难忘520/article/detail/936259
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号