1. 准备阶段

1.1. 简介

本案例介绍一个elascticsearch模块的实现。该模块属于一个商城项目，将mysql数据库同步入elascticsearch，当用户使用搜索功能的时候，从es中搜索并展示数据。页面如下。

在本模块中，总共实现了以下几个功能：

1. 创建索引
2. 数据全量导入
3. 搜索框输入的自动补全
4. 查询分类的聚合查询（动态改变分类信息）
5. 搜索查询功能（分页）

1.2. 环境配置

依赖

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!--elasticsearch-->
<dependency>
    <groupId>org.elasticsearch.client</groupId>
    <artifactId>elasticsearch-rest-high-level-client</artifactId>
</dependency>
<!--FastJson-->
<dependency>
    <groupId>com.alibaba</groupId>
    <artifactId>fastjson</artifactId>
    <version>1.2.71</version>
</dependency>

配置

es:
  host: 192.168.50.131
  port: 9200

索引

PUT /item

{
    "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":"text",
                    "analyzer":"keyword"
        },
        "name":{
            "type":"text",
                    "analyzer": "text_anlyzer",
                    "search_analyzer": "ik_smart",
                    "copy_to": "all"
        },
        "price":{
            "type":"integer"
        },
        "stock":{
            "type":"integer",
                    "index":"false"
        },
        "image":{
            "type":"keyword",
                    "index":"false"
        },
        "category":{
            "type":"keyword",
                    "copy_to": "all"
        },
        "brand":{
            "type":"keyword",
                    "copy_to": "all"
        },
        "spec":{
            "type":"keyword",
                    "index":"false"
        },
        "sold":{
            "type":"integer",
                    "index":"false"
        },
        "comment_count":{
            "type":"integer",
                    "index":"false"
        },
        "isAD":{
            "type":"boolean"
        },
        "status":{
            "type":"integer"
        },
        "create_time":{
            "type":"date",
                    "index":"false"
        },
        "update_time":{
            "type":"date",
                    "index":"false"
        },
        "all":{
            "type": "text",
                    "analyzer": "text_anlyzer",
                    "search_analyzer": "ik_smart"
        },
        "suggestion":{
            "type": "completion",
                    "analyzer": "completion_analyzer"
        }
    }
}

2. 代码实现

2.1. 其他

静态文件

EsItemConstant.java

public interface EsItemConstant {
    // 索引名字
    String INDEX_NAME = "item";
    // 自动补全字段
    String SUGGESTION_NAME = "itemSuggestion";
    // 自动补全显示条数
    Integer SUGGESTION_RESULT_NUM = 10;
    // 索引创建部分，此处是上面1.2中的索引
    String INDEX_TEMPLATE = "xxxxxxxxxxxxxxxxxxxx";
}

配置文件

EsConfig.java

@Configuration
public class EsConfig {

    @Value("${es.host}")
    private String host;

    @Value("${es.port}")
    private Integer port;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        return new RestHighLevelClient(RestClient.builder(new HttpHost(host, port)));
    }
}

Bean

实体类 Item.java

@Data
@TableName("tb_item")
public class Item {
    @TableId(type = IdType.AUTO)
    private Long id;//商品id
    private String name;//商品名称
    private Long price;//价格（分）
    private Integer stock;//库存数量
    private String image;//商品图片
    private String category;//分类名称
    private String brand;//品牌名称
    private String spec;//规格
    private Integer sold;//销量
    private Integer commentCount;//评论数
    private Integer status;//商品状态 1-正常，2-下架
    @TableField("isAD")
    private Boolean isAD;//商品状态 1-正常，2-下架
    private Date createTime;//创建时间
    private Date updateTime;//更新时间

    /**
     * 商品状态-正常
     */
    public static final int STATUS_NORMAL = 1;
    /**
     * 商品状态-下架
     */
    public static final int STATUS_OFF_SHELF = 2;

}

对应ES的对象 ItemDoc.java

@Data
@NoArgsConstructor
public class ItemDoc {

    private Long id;//商品id
    private String name;//商品名称
    private Long price;//价格（分）
    private Integer stock;//库存数量
    private String image;//商品图片
    private String category;//分类名称
    private String brand;//品牌名称
    private String spec;//规格
    private Integer sold;//销量
    private Integer commentCount;//评论数
    private Integer status;//商品状态 1-正常，2-下架
    @TableField("isAD")
    private Boolean isAD;//商品状态 1-正常，2-下架
    private Date createTime;//创建时间
    private Date updateTime;//更新时间

    private List<String> suggestion;

    public ItemDoc(Item item){
        this.id = item.getId();
        this.name = item.getName();
        this.price = item.getPrice();
        this.stock = item.getStock();
        this.image = item.getImage();
        this.category = item.getCategory();
        this.brand = item.getBrand();
        this.spec = item.getSpec();
        this.sold = item.getSold();
        this.commentCount = item.getCommentCount();
        this.status = item.getStatus();
        this.isAD = item.getIsAD();
        this.createTime = item.getCreateTime();
        this.updateTime = item.getUpdateTime();
        this.suggestion = new ArrayList<>();
        this.suggestion.add(this.brand);
        this.suggestion.add(this.category);
    }
}

2.2. controller

@RestController
@RequestMapping("search")
public class ItemSearchController {

    @Autowired
    private ISearchService searchService;

    /**
     * 自动补全
     *
     * @param key
     * @return
     */
    @GetMapping("/suggestion")
    public List<String> suggestion(@RequestParam("key") String key) {
        return searchService.suggestion(key);
    }

    /**
     * 索引聚合
     *
     * @param params
     * @return
     */
    @PostMapping("/filters")
    public ItemFilterVO getFilters(@RequestBody ItemFilterDTO params) {
        return searchService.filter(params);
    }

    /**
     * 搜索
     *
     * @param params
     * @return
     */
    @PostMapping("/list")
    public PageDTO<ItemDoc> createIndex(@RequestBody ItemFilterDTO params) {
        return searchService.pageQuery(params);
    }

}

2.3. service

import com.alibaba.fastjson.JSON;
import com.hmall.client.ItemClient;
import com.hmall.common.dto.ItemFilterDTO;
import com.hmall.common.dto.PageDTO;
import com.hmall.common.pojo.Item;
import com.hmall.common.pojo.ItemDoc;
import com.hmall.common.vo.ItemFilterVO;
import com.hmall.constant.EsItemConstant;
import com.hmall.service.ISearchService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
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.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.common.lucene.search.function.CombineFunction;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
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.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
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 org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@Slf4j
@Service
public class SearchService implements ISearchService {

    /**
     * ES客户端
     */
    @Autowired
    private RestHighLevelClient client;

    /**
     * 商品feign接口
     */
    @Autowired
    private ItemClient itemClient;

    /**
     * 创建索引item
     */
    public void creatIndex() {
        // 设置创建的index的名字
        CreateIndexRequest request = new CreateIndexRequest(EsItemConstant.INDEX_NAME);
        // 设置请求参数
        request.source(EsItemConstant.INDEX_TEMPLATE, XContentType.JSON);
        // 通过客户端对象发送一个请求
        try {
            CreateIndexResponse response = client
                    .indices()
                    .create(request, RequestOptions.DEFAULT);
            log.info("========索引创建成功！{}========", response.isAcknowledged());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 将数据全量导入ES
     */
    public void addAllDataToEs() {
        // 测试查询，获取总数
        PageDTO<Item> testData = itemClient.findByPage(1, 1);
        // 排除异常情况
        if (testData == null || testData.getList() == null || testData.getList().size() == 0 || testData.getTotal() == null || testData.getTotal() == 0) {
            log.error("feign调用失败，数据错误");
            return;
        }
        // 每页查询大小为500
        int pageSize = 500;
        // 获取总条数
        Long totalNum = testData.getTotal();
        // 获取总页数
        Long totalPage = (totalNum + pageSize - 1) / pageSize;

        // 循环获取页数并赋值给es
        for (int i = 1; i <= totalPage.intValue(); i++) {
            System.out.println("============【" + i * pageSize + "】=============");

            // 通过feign调用获取分页数据
            PageDTO<Item> eachPage = itemClient.findByPage(i, pageSize);

            List<ItemDoc> docList = new ArrayList<>();
            // 循环遍历分页信息，放入数据
            for (Item item : eachPage.getList()) {
                docList.add(new ItemDoc(item));
            }
            // 存入es
            addListToEs(docList);
        }
    }

    /**
     * 自动补全
     *
     * @param key
     * @return
     */
    @Override
    public List<String> suggestion(String key) {

        SearchRequest request = new SearchRequest(EsItemConstant.INDEX_NAME);
        request.source().suggest(
                new SuggestBuilder()
                        .addSuggestion(EsItemConstant.SUGGESTION_NAME,
                                SuggestBuilders
                                        .completionSuggestion("suggestion")
                                        .prefix(key)
                                        // 查询条数
                                        .size(EsItemConstant.SUGGESTION_RESULT_NUM)
                                        .skipDuplicates(true)));

        try {
            // 发请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 解析响应
            Suggest suggest = response.getSuggest();
            // 获取自动补全相关信息
            CompletionSuggestion itemSuggestion = suggest.getSuggestion(EsItemConstant.SUGGESTION_NAME);
            List<CompletionSuggestion.Entry.Option> options = itemSuggestion.getOptions();
            return options.stream().map(each -> each.getText().toString()).collect(Collectors.toList());
        } catch (IOException e) {
            log.error("自动补全失败！", e);
            return new ArrayList<>();
        }
    }

    /**
     * 聚合查询
     *
     * @param params
     * @return
     */
    @Override
    public ItemFilterVO filter(ItemFilterDTO params) {
        ItemFilterVO itemFilterVO = new ItemFilterVO();

        SearchRequest request = new SearchRequest(EsItemConstant.INDEX_NAME);
        // 设置查询条件
        request.source().query(setupQueryConditions(params));
        // 设置分类聚合条件
        request.source().aggregation(AggregationBuilders.terms("categoryAggs").field("category").size(10));
        // 设置品牌聚合条件
        request.source().aggregation(AggregationBuilders.terms("brandAggs").field("brand").size(10));

        // 设置排序
        setupSort(request, params);

        try {
            // 发请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 解析响应
            Aggregations aggregations = response.getAggregations();

            // 获取分类聚合条件的值
            itemFilterVO.setCategory(getKeyListFromAggs(aggregations, "categoryAggs"));
            // 获取品牌聚合条件的值
            itemFilterVO.setBrand(getKeyListFromAggs(aggregations, "brandAggs"));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        return itemFilterVO;

    }

    /**
     * 分页查询
     * @param params
     * @return
     */
    @Override
    public PageDTO<ItemDoc> pageQuery(ItemFilterDTO params) {
        SearchRequest request = new SearchRequest(EsItemConstant.INDEX_NAME);
        // 设置查询条件
        request.source().query(setupQueryConditions(params));
        // 设置分页条件
        request.source().from((params.getPage() - 1) * params.getSize());
        request.source().size(params.getSize());
        // 设置排序
        setupSort(request, params);
        try {
            // 发请求
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);
            // 解析响应
            return parseResponse(response);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 把列表中的数据添加到es
     *
     * @param docList
     */
    private void addListToEs(List<ItemDoc> docList) {

        // 批量请求
        BulkRequest bulkRequest = new BulkRequest();

        for (ItemDoc itemDoc : docList) {
            // 设置创建的index的名字
            IndexRequest request = new IndexRequest(EsItemConstant.INDEX_NAME).id(itemDoc.getId().toString());
            // 设置请求参数
            request.source(JSON.toJSONString(itemDoc), XContentType.JSON);
            // 添加到批量请求中
            bulkRequest.add(request);
        }

        try {
            // 执行批量任务
            BulkResponse response = client.bulk(bulkRequest, RequestOptions.DEFAULT);
            if (response.hasFailures()) {
                String msg = response.buildFailureMessage();
                log.error("添加失败！{}", msg);
            }
        } catch (IOException e) {
            log.error("批量导入异常！", e);
        }

    }

    /**
     * 设置排序条件
     *
     * @param request
     * @param param
     */
    private void setupSort(SearchRequest request, ItemFilterDTO param) {
        // 按照文档的分值_score进行排序
        request.source().sort("_score", SortOrder.DESC);
        // 先根据价格或者是评分排序
        if (Objects.equals(param.getSortBy(), "default") || Objects.equals(param.getSortBy(), "sold")) {
            request.source().sort("sold", SortOrder.DESC);
        } else {
            request.source().sort("price", SortOrder.ASC);
        }

    }

    /**
     * 设置查询条件
     *
     * @param param
     * @return
     */
    private QueryBuilder setupQueryConditions(ItemFilterDTO param) {
        // bool查询把所有的子条件组合到一块
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // key关键字
        if (!StringUtils.hasText(param.getKey())) {
            boolQueryBuilder.must(QueryBuilders.matchAllQuery());
        } else {
            boolQueryBuilder.must(QueryBuilders.matchQuery("all", param.getKey()));
        }
        // 过滤条件
        // 品牌，brand
        if (!StringUtils.isEmpty(param.getBrand())) {
            boolQueryBuilder.filter(QueryBuilders.termQuery("brand", param.getBrand()));
        }
        // 品牌 category
        if (!StringUtils.isEmpty(param.getCategory())) {
            boolQueryBuilder.filter(QueryBuilders.termQuery("category", param.getCategory()));
        }

        // 价格, 防御式编程
        if (param.getMinPrice() != null || param.getMaxPrice() != null) {
            RangeQueryBuilder rangeQueryBuilder = QueryBuilders.rangeQuery("price");
            if (param.getMinPrice() != null) {
                rangeQueryBuilder.gt(param.getMinPrice() * 100);
            }
            if (param.getMaxPrice() != null) {
                rangeQueryBuilder.lt(param.getMaxPrice() * 100);
            }
            boolQueryBuilder.filter(rangeQueryBuilder);
        }
        // 算分函数查询
        FunctionScoreQueryBuilder functionScoreQueryBuilder = QueryBuilders.functionScoreQuery(
                        // bool查询作为查询条件
                        boolQueryBuilder,
                        new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{
                                new FunctionScoreQueryBuilder.FilterFunctionBuilder(
                                        // 结果集的过滤条件
                                        QueryBuilders.termQuery("isAD", true),
                                        // 设置分值
                                        ScoreFunctionBuilders.weightFactorFunction(100)
                                )
                        })// 加权模式
                .boostMode(CombineFunction.MULTIPLY);
        return functionScoreQueryBuilder;
    }

    /**
     * 从aggregations中获取对应名字的值
     *
     * @param aggregations
     * @param name
     * @return
     */
    private List<String> getKeyListFromAggs(Aggregations aggregations, String name) {
        // 根据名字获取所有字段
        Terms aggs = aggregations.get(name);
        List<? extends Terms.Bucket> buckets = aggs.getBuckets();
        return buckets.stream().map(each -> each.getKeyAsString()).collect(Collectors.toList());
    }

    /**
     * 解析结果
     *
     * @param response
     * @return
     */
    private PageDTO<ItemDoc> parseResponse(SearchResponse response) {
        PageDTO<ItemDoc> result = new PageDTO<>();
        // 解析响应结果
        SearchHits hits = response.getHits();
        // 从TotalHits中取总记录的条数
        long total = hits.getTotalHits().value;
        result.setTotal(total);
        List<ItemDoc> hotels = new ArrayList<>();
        // 从Hits中取所有的doc
        for (SearchHit hit : hits.getHits()) {
            String json = hit.getSourceAsString();
            ItemDoc itemDoc = JSON.toJavaObject(JSON.parseObject(json), ItemDoc.class);

            hotels.add(itemDoc);
        }
        result.setList(hotels);
        return result;
    }

}