聚合

聚合（aggregations）可以让我们极其方便的实现对数据的统计、分析、运算
聚合常见的有三类：

  • 桶（Bucket）聚合：用来对文档做分组
    • TermAggregation：按照文档字段值分组，例如按照品牌值分组、按照国家分组
    • Date Histogram：按照日期阶梯分组，例如一周为一组，或者一月为一组
  • 度量（Metric）聚合：用以计算一些值，比如：最大值、最小值、平均值等
    • Avg：求平均值
    • Max：求最大值
    • Min：求最小值
    • Stats：同时求max、min、avg、sum等

  • 管道（pipeline）聚合：其它聚合的结果为基础做聚合

    Bucket聚合语法：

    image.png

    Metric聚合语法：

    例如stat聚合：就可以获取min、max、avg等结果。
    image.png

    RestAPI实现聚合

    聚合条件语法
    image.png
    聚合结果解析
    image.png

    1. void testMetric() throws IOException {
    2.       //1.创建对应请求
    3.       SearchRequest request = new SearchRequest("hotel");
    4.       //2.DSL
    5.       request.source().size(0);
    6.       request.source().aggregation(AggregationBuilders
    7.               .terms("brandAgg")
    8.               .field("brand")
    9.               .size(20)
    10.               .order(BucketOrder.aggregation("_count",true))
    11.       );
    12.       //3.发请求
    13.       SearchResponse response = client.search(request, RequestOptions.DEFAULT);
    14.       //4.解析结果
    15.       Aggregations aggregations = response.getAggregations();
    16.       Terms brandAgg = aggregations.get("brandAgg");
    17.       List<? extends Terms.Bucket> buckets = brandAgg.getBuckets();
    18.       for (Terms.Bucket bucket : buckets) {
    19.           String key = bucket.getKeyAsString();
    20.           System.out.println(key);
    21.       }
    23.   }

    案例：实现搜索栏条件
    image.png ```java package cn.itcast.hotel.service.impl;

import cn.itcast.hotel.mapper.HotelMapper; import cn.itcast.hotel.pojo.Hotel; import cn.itcast.hotel.pojo.HotelDoc; import cn.itcast.hotel.pojo.PageResult; import cn.itcast.hotel.pojo.RequestParams; import cn.itcast.hotel.service.IHotelService; import com.alibaba.fastjson.JSON; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.geo.GeoPoint; import org.elasticsearch.common.unit.DistanceUnit; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.index.query.QueryBuilders; import org.elasticsearch.index.query.functionscore.FunctionScoreQueryBuilder; import org.elasticsearch.index.query.functionscore.ScoreFunctionBuilders; import org.elasticsearch.search.SearchHit; import org.elasticsearch.search.SearchHits; import org.elasticsearch.search.aggregations.Aggregation; import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.Aggregations; import org.elasticsearch.search.aggregations.bucket.terms.Terms; import org.elasticsearch.search.sort.SortBuilders; import org.elasticsearch.search.sort.SortOrder; import org.elasticsearch.search.suggest.Suggest; import org.elasticsearch.search.suggest.SuggestBuilder; import org.elasticsearch.search.suggest.SuggestBuilders; import org.elasticsearch.search.suggest.completion.CompletionSuggestion; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service;

import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map;

@Service public class HotelService extends ServiceImpl implements IHotelService {

  1. @Autowired
  2. private RestHighLevelClient client;
  4. /**
  5.  * 根据关键字搜索酒店信息
  6.  *
  7.  * @param params 请求参数对象，包含用户输入的关键字
  8.  * @return 酒店文档列表
  9.  */
  10. @Override
  11. public PageResult search(RequestParams params) {
  12.     try {
  13.         //1.
  14.         SearchRequest request = new SearchRequest("hotel");
  16.         //2.准备DSL并添加条件
  17.         addQuery(request, params);
  18.         //分页
  19.         int size = params.getSize();
  20.         int begin = (params.getPage() - 1) * size;
  21.         request.source().from(begin).size(size);
  22.         //距离排序
  23.         String location = params.getLocation();
  24.         if (location != null && !location.equals("")) {
  25.             request.source().sort(SortBuilders
  26.                     .geoDistanceSort("location", new GeoPoint(location))
  27.                     .order(SortOrder.ASC)
  28.                     .unit(DistanceUnit.KILOMETERS)
  29.             );
  30.         }
  31.         //发送请求
  32.         SearchResponse response = client.search(request, RequestOptions.DEFAULT);
  33.         // 解析响应
  34.         return handleResponse(response);
  35.     } catch (IOException e) {
  36.         throw new RuntimeException(e);
  37.     }
  38. }
  40. /**
  41.  * 条件查询拼接DSL
  42.  *
  43.  * @param request
  44.  * @param params
  45.  * @return
  46.  */
  47. private void addQuery(SearchRequest request, RequestParams params) {
  48.     //2.1query
  49.     //构建boolQuery,多条件使用bool查询
  50.     BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
  51.     String key = params.getKey();
  52.     //判断搜索词
  53.     if (key == null || "".equals(key)) {
  54.         boolQuery.must(QueryBuilders.matchAllQuery());
  55.     } else {
  56.         boolQuery.must(QueryBuilders.matchQuery("all", key));
  57.     }
  58.     // 3.城市条件
  59.     if (params.getCity() != null && !params.getCity().equals("")) {
  60.         boolQuery.filter(QueryBuilders.termQuery("city", params.getCity()));
  61.     }
  62.     // 4.品牌条件
  63.     if (params.getBrand() != null && !params.getBrand().equals("")) {
  64.         boolQuery.filter(QueryBuilders.termQuery("brand", params.getBrand()));
  65.     }
  66.     // 5.星级条件
  67.     if (params.getStarName() != null && !params.getStarName().equals("")) {
  68.         boolQuery.filter(QueryBuilders.termQuery("starName", params.getStarName()));
  69.     }
  70.     //6.价格
  71.     if (params.getMinPrice() != null && params.getMaxPrice() != null) {
  72.         boolQuery.filter(QueryBuilders.rangeQuery("price")
  73.                 .gte(params.getMinPrice())
  74.                 .lte(params.getMaxPrice()));
  75.     }
  76.     //function_score
  77.     FunctionScoreQueryBuilder functionScoreQuery = QueryBuilders.functionScoreQuery(
  78.             // 原始查询，相关性算分的查询
  79.             boolQuery,
  80.             // function score的数组
  81.             new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{
  82.                     // 其中的一个function score 元素
  83.                     new FunctionScoreQueryBuilder.FilterFunctionBuilder(
  84.                             //过滤条件
  85.                             QueryBuilders.termQuery("isAD", true),
  86.                             //算分函数
  87.                             ScoreFunctionBuilders.weightFactorFunction(10)
  88.                     )
  89.             }
  90.     );
  91.     //放入条件
  92.     request.source().query(functionScoreQuery);
  93. }
  95. /**
  96.  * 搜索栏条件添加
  97.  *
  98.  * @param params
  99.  * @return
  100.  */
  101. @Override
  102. public Map<String, List<String>> getFilters(RequestParams params) {
  103.     try {
  104.         //1.
  105.         SearchRequest request = new SearchRequest("hotel");
  106.         //2.准备DSL并添加条件
  107.         addQuery(request, params);
  108.         //2.正常条件
  109.         request.source().size(0);
  110.         //3.聚合条件
  111.         addFilter(request);
  112.         //4.发送请求
  113.         SearchResponse response = client.search(request, RequestOptions.DEFAULT);
  114.         //解析结果
  115.         Aggregations aggregations = response.getAggregations();
  116.         //放入不同map
  117.         Map<String, List<String>> map = new HashMap<>();
  118.         List<String> cityList = getAggField(aggregations, "cityAgg");
  119.         map.put("city", cityList);
  120.         List<String> starList = getAggField(aggregations, "starAgg");
  121.         map.put("starName", starList);
  122.         List<String> brandList = getAggField(aggregations, "brandAgg");
  123.         map.put("brand", brandList);
  125.         return map;
  126.     } catch (IOException e) {
  127.         throw new RuntimeException(e);
  128.     }
  129. }
  131. /**
  132.  * 自动补全
  133.  *
  134.  * @param key
  135.  * @return
  136.  */
  137. @Override
  138. public List<String> getSuggestions(String key) {
  139.     try {
  140.         //1.请求
  141.         SearchRequest request = new SearchRequest("hotel");
  142.         //2.DSL的java代码
  143.         request.source().suggest(new SuggestBuilder().addSuggestion(
  144.                 "suggestions",
  145.                 SuggestBuilders.completionSuggestion("suggestion")
  146.                         .prefix(key)
  147.                         .skipDuplicates(true)
  148.                         .size(10)
  149.         ));
  150.         //3.发送请求获取参数
  151.         SearchResponse response = client.search(request, RequestOptions.DEFAULT);
  152.         //4.解析参数
  153.         Suggest suggest = response.getSuggest();
  154.         // 4.1.根据补全查询名称，获取补全结果
  155.         CompletionSuggestion suggestions= suggest.getSuggestion("suggestions");
  156.         // 4.2.获取options
  157.         List<CompletionSuggestion.Entry.Option> options = suggestions.getOptions();
  158.         List<String>list=new ArrayList<>();
  159.         //4.3遍历放入集合
  160.         for (CompletionSuggestion.Entry.Option option : options) {
  161.             String string = option.getText().string();
  162.             list.add(string);
  163.         }
  164.         return list;
  165.     } catch (IOException e) {
  166.         throw new RuntimeException(e);
  167.     }
  168. }
  170. /**
  171.  * 新增、修改
  172.  * @param id
  173.  */
  174. @Override
  175. public void insertById(Long id) {
  176.     try {
  177.         //查询对应id的酒店信息
  178.         Hotel hotel = getById(id);
  179.         HotelDoc hotelDoc=new HotelDoc(hotel);
  180.         //请求
  181.         IndexRequest request = new IndexRequest("hotel").id(id.toString());
  182.         //组装DSL
  183.         request.source(JSON.toJSONString(hotelDoc), XContentType.JSON);
  184.         //发请求
  185.         client.index(request,RequestOptions.DEFAULT);
  186.     } catch (IOException e) {
  187.         throw new RuntimeException(e);
  188.     }
  189. }
  191. /**
  192.  * 删除
  193.  * @param id
  194.  */
  195. @Override
  196. public void deleteById(Long id) {
  197.     try {
  198.         //创建
  199.         DeleteRequest request = new DeleteRequest("hotel",id.toString());
  200.         client.delete(request,RequestOptions.DEFAULT);
  201.     } catch (IOException e) {
  202.         throw new RuntimeException(e);
  203.     }
  204. }
  207. /**
  208.  * 得到对应搜索框信息
  209.  *
  210.  * @param aggregations
  211.  * @param aggName
  212.  * @return
  213.  */
  214. private List<String> getAggField(Aggregations aggregations, String aggName) {
  215.     Terms aggTerms = aggregations.get(aggName);
  216.     List<? extends Terms.Bucket> buckets = aggTerms.getBuckets();
  217.     List<String> list = new ArrayList<>();
  218.     for (Terms.Bucket bucket : buckets) {
  219.         String key = bucket.getKeyAsString();
  220.         list.add(key);
  221.     }
  222.     return list;
  223. }
  225. /**
  226.  * 聚合条件
  227.  *
  228.  * @param request
  229.  */
  230. private void addFilter(SearchRequest request) {
  231.     request.source().aggregation(AggregationBuilders
  232.             .terms("brandAgg")
  233.             .field("brand")
  234.             .size(100));
  235.     request.source().aggregation(AggregationBuilders
  236.             .terms("starAgg")
  237.             .field("starName")
  238.             .size(100));
  239.     request.source().aggregation(AggregationBuilders
  240.             .terms("cityAgg")
  241.             .field("city")
  242.             .size(100)
  243.     );
  244. }
  246. /**
  247.  * 结果解析
  248.  *
  249.  * @param response
  250.  * @return
  251.  */
  252. private PageResult handleResponse(SearchResponse response) {
  253.     //解析结果
  254.     SearchHits searchHits = response.getHits();
  255.     //总数
  256.     long total = searchHits.getTotalHits().value;
  257.     //集合
  258.     SearchHit[] hits = searchHits.getHits();
  259.     List<HotelDoc> hotelDocList = new ArrayList<>();
  260.     for (SearchHit hit : hits) {
  261.         //酒店对象JSON
  262.         String hitSourceAsString = hit.getSourceAsString();
  263.         //转换对象hotelDoc
  264.         HotelDoc hotelDoc = JSON.parseObject(hitSourceAsString, HotelDoc.class);
  265.         //距离
  266.         Object[] sortValues = hit.getSortValues();
  267.         if (sortValues != null && sortValues.length > 0) {
  268.             hotelDoc.setDistance(sortValues[0]);
  269.         }
  270.         hotelDocList.add(hotelDoc);
  271.     }
  272.     return new PageResult(total, hotelDocList);
  273. }

}

  1. <a name="oioHr"></a>
  2. ### 自动补全
  3. 拼音分词器：<br />安装方式与IK分词器一样，分三步：<br />    ①解压<br />    ②上传到虚拟机中，elasticsearch的plugin目录<br />    ③重启elasticsearch<br />    ④测试<br /> 
  4. ```java
  5. POST /_analyze{  "text": "如家酒店还不错",  "analyzer": "pinyin"}

创建表时添加suggestion字段用于补全
只有在创建表的时候添加

// 酒店数据索引库
PUT /hotel
{
  "settings": {
    "analysis": {
      "analyzer": {
        "text_anlyzer": {
          "tokenizer": "ik_max_word",
          "filter": "py"
        },
        "completion_analyzer": {
          "tokenizer": "keyword",
          "filter": "py"
        }
      },
      "filter": {
        "py": {
          "type": "pinyin",
          "keep_full_pinyin": false,
          "keep_joined_full_pinyin": true,
          "keep_original": true,
          "limit_first_letter_length": 16,
          "remove_duplicated_term": true,
          "none_chinese_pinyin_tokenize": false
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "id":{
        "type": "keyword"
      },
      "name":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart",
        "copy_to": "all"
      },
      "address":{
        "type": "keyword",
        "index": false
      },
      "price":{
        "type": "integer"
      },
      "score":{
        "type": "integer"
      },
      "brand":{
        "type": "keyword",
        "copy_to": "all"
      },
      "city":{
        "type": "keyword"
      },
      "starName":{
        "type": "keyword"
      },
      "business":{
        "type": "keyword",
        "copy_to": "all"
      },
      "location":{
        "type": "geo_point"
      },
      "pic":{
        "type": "keyword",
        "index": false
      },
      "all":{
        "type": "text",
        "analyzer": "text_anlyzer",
        "search_analyzer": "ik_smart"
      },
      "suggestion":{
          "type": "completion",
          "analyzer": "completion_analyzer"
      }
    }
  }
}

image.png
实例DSL：
image.png
解析：
image.png

@Override
public List<String> getSuggestions(String prefix) {
    try {
        // 1.准备Request
        SearchRequest request = new SearchRequest("hotel");
        // 2.准备DSL
        request.source().suggest(new SuggestBuilder().addSuggestion(
            "suggestions",
            SuggestBuilders.completionSuggestion("suggestion")
            .prefix(prefix)
            .skipDuplicates(true)
            .size(10)
        ));
        // 3.发起请求
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        // 4.解析结果
        Suggest suggest = response.getSuggest();
        // 4.1.根据补全查询名称，获取补全结果
        CompletionSuggestion suggestions = suggest.getSuggestion("suggestions");
        // 4.2.获取options
        List<CompletionSuggestion.Entry.Option> options = suggestions.getOptions();
        // 4.3.遍历
        List<String> list = new ArrayList<>(options.size());
        for (CompletionSuggestion.Entry.Option option : options) {
            String text = option.getText().toString();
            list.add(text);
        }
        return list;
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

ES索引和数据库的数据同步问题：

三种数据同步方法
方式一：同步调用

  • 优点：实现简单，粗暴
  • 缺点：业务耦合度高

方式二：异步通知

  • 优点：低耦合，实现难度一般
  • 缺点：依赖mq的可靠性

方式三：监听binlog——相对高级

  • 优点：完全解除服务间耦合
  • 缺点：开启binlog增加数据库负担、实现复杂度高

流程：

需要用到的模块都需要导入
<!--amqp-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

在yml中申明rabbitMq的端口。。
spring:
  rabbitmq:
    host: 121.36.164.132 # 主机名
    port: 5672 # 端口
    virtual-host: / # 虚拟主机
    username: itcast # 用户名
    password: 123321 # 密码

可以用配置类声明也可以直接用注解，注解详细见上一片ES

@Configuration
public class MqConfig {
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(MqConstants.HOTEL_EXCHANGE, true, false);
    }

    @Bean
    public Queue insertQueue(){
        return new Queue(MqConstants.HOTEL_INSERT_QUEUE, true);
    }

    @Bean
    public Queue deleteQueue(){
        return new Queue(MqConstants.HOTEL_DELETE_QUEUE, true);
    }

    @Bean
    public Binding insertQueueBinding(){
        return BindingBuilder.bind(insertQueue()).to(topicExchange()).with(MqConstants.HOTEL_INSERT_KEY);
    }

    @Bean
    public Binding deleteQueueBinding(){
        return BindingBuilder.bind(deleteQueue()).to(topicExchange()).with(MqConstants.HOTEL_DELETE_KEY);
    }
}

@PostMapping
    public void saveHotel(@RequestBody Hotel hotel) {
        hotelService.save(hotel);

        rabbitTemplate.convertAndSend(MQConstants.HOTEL_EXCHANGE
                ,MQConstants.HOTEL_INSERT_KEY
                ,hotel.getId());
    }

    @PutMapping()
    public void updateById(@RequestBody Hotel hotel) {
        if (hotel.getId() == null) {
            throw new InvalidParameterException("id不能为空");
        }
        hotelService.updateById(hotel);
        rabbitTemplate.convertAndSend(MQConstants.HOTEL_EXCHANGE
                ,MQConstants.HOTEL_INSERT_KEY
                ,hotel.getId());
    }

    @DeleteMapping("/{id}")
    public void deleteById(@PathVariable("id") Long id) {
        hotelService.removeById(id);
        rabbitTemplate.convertAndSend(MQConstants.HOTEL_EXCHANGE
                ,MQConstants.HOTEL_DELETE_KEY
                ,id);
    }

编写监听器

@Component
public class HotelListener {

    @Autowired
    private IHotelService hotelService;

    /**
     * 监听酒店新增或修改的业务
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_INSERT_QUEUE)
    public void listenHotelInsertOrUpdate(Long id){
        hotelService.insertById(id);
    }

    /**
     * 监听酒店删除的业务
     * @param id 酒店id
     */
    @RabbitListener(queues = MqConstants.HOTEL_DELETE_QUEUE)
    public void listenHotelDelete(Long id){
        hotelService.deleteById(id);
    }
}

面试相关问题：

数据同步的解决方案（面试）

有以下几种解决方案
同步调用：一般出现在传统类型的项目中，耦合度高，速度较慢
异步通知：使用MQ队列来完成异步和一定程度的解耦，依赖MQ的可靠性
监听binlog：可以通过阿里与的canal组件来监听mysql的binlog日志来实现，可以完全的解耦，开启binlog会增加数据库负担、实现复杂度高

ES内部的实现原理

数据插入的分片如何确定

先通过hash算法算出数据的hash值，然后用这个值去和分片的数量进行%运算，得到的值就是放入第几个分片
，也因此，索引库一旦创建，分片数量就不能修改

数据的查询

分为两个阶段
第一阶段是分散阶段，在这个阶段，协调节点会把请求分发到每一个分片
第二阶段是聚集阶段，这个阶段，协调节点会汇总分片的数据结果，并经过处理后将最终结果返回给用户

脑裂（面试）

数据脑裂：在一个集群中出现了不止一个master，这就是脑裂
解决：
配置方面：使用奇数个节点，因为选取master要满足半数以上原则
增加内存，防止内存不够带来的无法响应
增加响应超时时间，把心跳响应时间从默认的30s调整为更长的时间，来避免网络波动带来的无法接受心跳消息
较少访问数量，不要在短时间内对master进行太高的写请求并发，防止高并发带来的无法处理其他请求信息

ES的集群故障转移

集群的master节点会监控集群中的节点状态，如果发现有节点宕机，会立即将宕机节点的分片数据迁移到其他节点，确保数据安全，这就是故障转移