第三十五节 ElasticSearch存储订单并检索

博主

亮子

专辑

第三十五节 ElasticSearch存储订单并检索

亮子 2025-03-16 22:10:24

一、需求分析

根据公开的信息，淘宝平台上的商品数量已经超过了数十亿个，每天有成千上万的新商品上架。根据公开的数据，京东平台上的商品种类已经达到数亿个。如何存储和检索商品是一个现实而有严峻的问题。

将商品信息存入 Elasticsearch 的现实需求主要源于现代电商和零售行业对高效搜索、灵活查询和实时数据分析的迫切需求。以下是具体场景和实际问题的描述：

1. 电商平台的高效搜索体验

需求场景：
用户在电商平台搜索商品时，期望输入关键词（如“红色连衣裙”或“防水运动鞋”）后，毫秒级返回相关结果，并支持模糊匹配、拼写纠错、同义词扩展（如搜索“跑步鞋”也能匹配“运动鞋”）。
传统方案的痛点：
关系型数据库（如MySQL）的 LIKE 查询性能低下，无法应对海量商品数据（如百万级SKU）。
难以实现分词、权重排序（如销量高的商品优先展示）或多字段联合检索（如同时匹配标题、描述、标签等）。
Elasticsearch 的解决能力：
倒排索引：支持全文搜索和分词（如中文IK分词），快速定位关键词。
相关性评分：根据关键词匹配度、销量、评分等字段动态排序结果。
模糊查询：支持拼写纠错（fuzzy search）和同义词扩展，提升搜索容错性。

2. 复杂过滤与聚合分析

需求场景：
用户需要根据多个条件筛选商品（如价格区间、品牌、规格、库存状态），并期望平台提供实时数据分析（如“热销商品排行榜”或“用户点击量趋势”）。
传统方案的痛点：
关系型数据库的多字段联合查询（如 WHERE price BETWEEN 100 AND 500 AND brand='A'）在大数据量下性能急剧下降。
无法高效生成聚合统计结果（如按类别统计商品数量、按地区分析销量）。
Elasticsearch 的解决能力：
组合查询（Bool Query）：支持多条件组合过滤（must、should、must_not）。
聚合分析（Aggregation）：实时统计商品价格分布、品类占比、用户行为（如点击/购买趋势）。
范围查询与地理搜索：支持按地理位置筛选附近商家的库存商品（如“附近5公里有货”）。

3. 实时数据同步与更新

需求场景：
商品价格、库存状态、上下架信息需要实时更新，并在搜索结果中立即生效（例如秒杀活动中库存扣减或价格变动）。
传统方案的痛点：
数据库与缓存（如Redis）之间的数据同步延迟可能导致用户看到过期信息（如已售罄商品仍显示可购买）。
Elasticsearch 的解决能力：
近实时（Near Real-Time, NRT）：数据写入后1秒内可被检索，保障信息的及时性。
与业务系统集成：通过Logstash、Kafka或应用层双写，实现商品数据的实时同步。

4. 应对高并发与海量数据

需求场景：
在大促活动（如双11）期间，电商平台可能面临每秒数万次的搜索请求，需保障系统稳定性和低延迟。
传统方案的痛点：
单机数据库或缓存无法承载高并发查询，容易成为性能瓶颈。
分库分表方案复杂，且难以支持跨分片的复杂查询。
Elasticsearch 的解决能力：
分布式架构：数据自动分片（Sharding）和副本（Replica），支持水平扩展，轻松应对高并发。
负载均衡与容灾：节点故障时自动切换，保障服务可用性。

5. 个性化推荐与用户行为分析

需求场景：
根据用户历史行为（如搜索、点击、购买记录）推荐相关商品，提升转化率。
传统方案的痛点：
基于规则或离线批处理的推荐系统延迟高，无法实时响应用户行为。
Elasticsearch 的解决能力：
结合用户画像：存储用户行为日志，通过聚合分析生成实时兴趣标签。
混合搜索（Hybrid Search）：结合关键词搜索和协同过滤（如“购买此商品的人也买了XXX”）。

总结：现实需求的核心驱动

业务层面：
提升用户搜索体验，减少跳出率，促进转化。
支持复杂的运营分析（如实时监控热销商品、优化库存策略）。
技术层面：
解决传统数据库在海量数据、高并发、复杂查询下的性能瓶颈。
简化开发流程（Elasticsearch提供RESTful API，无需复杂SQL或JOIN操作）。

通过将商品信息存入Elasticsearch，企业能够在竞争激烈的市场中快速响应用户需求，同时降低技术维护成本，实现数据驱动的业务增长。

二、SpringBoot集成ElasticSearch

1、添加依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
</dependency>

2、配置 Elasticsearch 连接

在 application.properties 或 application.yml 中配置 Elasticsearch 的连接信息：

# application.properties
spring.elasticsearch.rest.uris=http://localhost:9200
spring.data.elasticsearch.repositories.enabled=true

或使用 YAML 格式：

# application.yml
spring:
  elasticsearch:
    rest:
      uris: http://localhost:9200
  data:
    elasticsearch:
      repositories:
        enabled: true

3、创建ES的实体类

package com.bwie.es;

import com.fasterxml.jackson.annotation.JsonFormat;
import lombok.Data;
import org.springframework.data.annotation.Id;
import org.springframework.data.elasticsearch.annotations.DateFormat;
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;

/**
 * @author 军哥
 * @version 1.0
 * @description: TODO
 * @date 2025/3/1 9:27
 */

@Data
@Document(indexName = "es_prod", shards = 1, replicas = 1)
public class EsProd implements Serializable {

    @Id
    @Field(type = FieldType.Keyword)
    private Integer prodId;

    @Field(type = FieldType.Text, analyzer = "ik_max_word")
    private String prodName;

    @Field(type = FieldType.Double)
    private Double prodPrice;

    @Field(type = FieldType.Integer)
    private Integer prodStock;

    @Field(type = FieldType.Integer)
    private Integer prodType;

    @Field(type = FieldType.Integer)
    private Integer userId;

    @Field(type = FieldType.Integer)
    private Integer prodStatus;


    @Field(type = FieldType.Keyword, index = false)
    private String nickName;
    @Field(type = FieldType.Keyword, index = false)
    private String typeName;
    @Field(type = FieldType.Keyword, index = false)
    private String statusName;

    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8", shape = JsonFormat.Shape.STRING)
    @Field(type = FieldType.Date, format = DateFormat.custom, pattern = "yyyy-MM-dd HH:mm:ss")
    private Date createTime;

    @Field(type = FieldType.Keyword)
    private String prodCode;
}

4、创建索引

创建索引：

@Override
public R createIndex() {
    //-- 检查索引是否存在，如果存在，删除索引
    IndexOperations indexOps = elasticsearchRestTemplate.indexOps(EsProd.class);
    boolean exists = indexOps.exists();
    if(exists) {
        System.out.println("索引存在，删除索引");
        indexOps.delete();
    }

    //-- 创建索引
    indexOps.create();

    //-- 创建映射
    Document mapping = indexOps.createMapping();
    indexOps.putMapping(mapping);

    return R.OK();
}

从数据库同步到ElasticSearch中：

package com.bwie.task;

import cn.hutool.core.bean.BeanUtil;
import com.bwie.es.EsProd;
import com.bwie.mapper.TbProdMapper;
import com.bwie.pojo.TbProd;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.IndexOperations;
import org.springframework.data.elasticsearch.core.document.Document;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author 军哥
 * @version 1.0
 * @description: TODO
 * @date 2025/3/1 10:19
 */

@Component
@Slf4j
public class SyncProdTask {

    @Autowired
    TbProdMapper tbProdMapper;

    @Autowired
    ElasticsearchRestTemplate elasticsearchRestTemplate;

    /**
     * @description 同步商品到ES中,每分钟执行一次
     * @params []
     * @return void
     * @author 军哥
     * @date 2025/3/1 10:20
     */
//    @Scheduled(cron = "0 0/1 * * * ?")
    public void syncProdToEs() {
        log.info("同步商品到ES中...");

        //-- 判断索引是否存在，如果不存在，创建索引
        IndexOperations indexOps = elasticsearchRestTemplate.indexOps(EsProd.class);
        boolean exists = indexOps.exists();
        if(!exists) {
            System.out.println("索引不存在，创建索引");
            indexOps.create();

            //-- 创建映射
            Document mapping = indexOps.createMapping();
            indexOps.putMapping(mapping);
        }


        //-- 获取商品列表
        List<TbProd> tbProds = tbProdMapper.getProdList();

        //-- 循环商品列表,并保存到ES中
        for (TbProd tbProd : tbProds) {
            EsProd esProd = BeanUtil.toBean(tbProd, EsProd.class);
            elasticsearchRestTemplate.save(esProd);
        }

        log.info("同步商品到ES中完成...");
    }

}

三、ElasticSearch数据操作

1、保存文档

@Override
public R addProdToEs(Integer prodId) {

    TbProd tbProd = tbProdMapper.selectById(prodId);
    if(tbProd == null) {
        return R.ERROR(404, "商品不存在");
    }

    //-- 将商品信息添加到ES中
    EsProd esProd = BeanUtil.toBean(tbProd, EsProd.class);
    elasticsearchRestTemplate.save(esProd);

    return R.OK();
}

2、根据ID读取文档

@Override
public R getProdFromEs(Integer prodId) {
    EsProd esProd = elasticsearchRestTemplate.get(String.valueOf(prodId), EsProd.class);
    return R.OK(esProd);
}

3、修改文档

@Override
public R updateProdToEs(Integer prodId) {
    TbProd tbProd = tbProdMapper.selectById(prodId);
    if(tbProd == null) {
        return R.ERROR(404, "商品不存在");
    }

    //-- 将商品信息添加到ES中
    // save 根据ID来判断，如果ID存在，修改，如果不存在，添加
    EsProd esProd = BeanUtil.toBean(tbProd, EsProd.class);
    elasticsearchRestTemplate.save(esProd);

    return R.OK();
}

4、删除文档

@Override
public R deleteProdFromEs(Integer prodId) {
    elasticsearchRestTemplate.delete(""+prodId, EsProd.class);
    return R.OK();
}

四、ElasticSearch 难点

1、多条件+排序+高亮+分词

@Override
public R getProdListFromEs(MyPageVo myPageVo) {
    //-- 构造分页参数（ES页面是从0开始的）
    Integer pageNum = 0;
    if(myPageVo.getPageNum() > 0) {
        pageNum = myPageVo.getPageNum() - 1;
    }
    PageRequest pageRequest = PageRequest.of(pageNum, myPageVo.getPageSize());

    //-- 构造查询条件
    BoolQueryBuilder boolQueryBuilder = new BoolQueryBuilder();

    // 根据商品名称搜索:must=and,should=or
    if(!StringUtils.isEmpty(myPageVo.getProdName())) {
        boolQueryBuilder.must(QueryBuilders.matchQuery("prodName", myPageVo.getProdName()));
    }

    // 根据商品类型搜索:matchQuery 分词查找，termQuery 精确查找
    if(myPageVo.getProdType() != null) {
        boolQueryBuilder.must(QueryBuilders.termQuery("prodType", myPageVo.getProdType()));
    }

    // 根据商品状态搜索
    if(myPageVo.getProdStatus() != null) {
        boolQueryBuilder.must(QueryBuilders.termQuery("prodStatus", myPageVo.getProdStatus()));
    }

    // 根据价格区间搜索
    if(myPageVo.getMinPrice() != null && myPageVo.getMaxPrice() != null) {
        boolQueryBuilder.must(
                QueryBuilders.rangeQuery("prodPrice")
                        .gte(myPageVo.getMinPrice())
                        .lte(myPageVo.getMaxPrice())
        );
    }

    // 根据时间区间搜索
    if(!StringUtils.isEmpty(myPageVo.getBeginTime()) && !StringUtils.isEmpty(myPageVo.getEndTime())) {
        boolQueryBuilder.must(
                QueryBuilders.rangeQuery("createTime")
                        .format("yyyy-MM-dd HH:mm:ss")
                        .timeZone("GMT+8")
                        .gte(myPageVo.getBeginTime())
                        .lte(myPageVo.getEndTime())
        );
    }

    // 高亮查询
    HighlightBuilder highlightBuilder = new HighlightBuilder();
    highlightBuilder.field("prodName").preTags("<font color='red'>").postTags("</font>");

    // 价格排序
    SortBuilder sortBuilder = SortBuilders.fieldSort("prodPrice").order(SortOrder.DESC);


    //-- 执行查询
    NativeSearchQueryBuilder nativeSearchQueryBuilder = new NativeSearchQueryBuilder();
    nativeSearchQueryBuilder.withQuery(boolQueryBuilder)
            .withPageable(pageRequest)
            .withHighlightBuilder(highlightBuilder)
            .withSort(sortBuilder);
    SearchHits<EsProd> searchHits = elasticsearchRestTemplate.search(nativeSearchQueryBuilder.build(), EsProd.class);

    //-- 解析查询结果
    List<EsProd> records = new ArrayList<>();
    Long total = searchHits.getTotalHits();

    for (SearchHit<EsProd> searchHit : searchHits) {
        EsProd esProd = searchHit.getContent();

        //-- 获取高亮字段
        List<String> list = searchHit.getHighlightField("prodName");
        if(list != null && list.size() > 0) {
            esProd.setProdName(list.get(0));
        }

        records.add(esProd);
    }

    HashMap<String, Object> map = new HashMap<>();
    map.put("records", records);
    map.put("total", total);
    return R.OK(map);
}

五、拓展

1、什么是倒排索引？

ES分词器通过扫描文章中的每一个词，对每一个词建立一个索引，指明该词在文章中出现的次数和位置，当用户查询时，检索程序就根据事先建立的索引进行查找，并将查找的结果反馈给用户的检索方式。这种建立索引的方式叫倒排索引。
当数据写入 ES 时，数据将会通过分词被切分为不同的term，ES 将term 与其对应的文档列表建立一种映射关系，这种结构就是倒排索引。如下图所示：

图片alt

参考文章

3、ES脑裂是怎么回事？（扩展）

脑裂问题，就是同一个集群中的不同节点，对于集群的状态，有了不一样的理解。

由于并发访问量的提高，导致了我们两个节点的集群（分片数默认为5，副本为1，没有固定的master，都是集群中的节点又做data又做master）状态变成了red，出现了大量的坏片，并且坏掉的都是主分片及其副本。分析发现，是ES集群出现了脑裂问题（俗称精神分裂），即**集群中不同的节点对于master的选择出现了分歧，出现了多个master竞争，导致主分片和副本的识别也发生了分歧，对一些分歧中的分片标识为了坏片。**

理解思路

原因1：网络问题，导致由多个主
原因2：主节点负载过大

4、“脑裂”问题是怎么形成的？（扩展）

1.网络问题：集群间的网络延迟导致一些节点访问不到master，认为master挂掉了从而选举出新的master，并对master上的分片和副本标红，分配新的主分片

2.节点负载：主节点的角色既为master又为data，访问量较大时可能会导致ES停止响应造成大面积延迟，此时其他节点得不到主节点的响应认为主节点挂掉了，会重新选取主节点。

3.内存回收：data节点上的ES进程占用的内存较大，引发JVM的大规模内存回收，造成ES进程失去响应。

5、脑裂问题如何解决？（扩展）

1、减少误判。discovery.zen.ping_timeout节点状态的响应时间，默认为3s，可以适当调大，减少误判。
2、选举触发。discovery.zen.minimum_master_nodes:1，该参数是用于控制选举行为发生的最小集群主节点数量。增大该参数，这样参与选举的节点增多，减少选举。
3、角色分离：即master节点与data节点分离，限制角色。
主节点配置为：
node.master: true node.data: false
从节点配置为：
node.master: false node.data: true

6、ES如何添加自定义词库、热更新词库？（扩展）

我们项目中使用中文分词器IK分词器。
1、在ik插件对应的配置文件目录下创建一个自定义词库文件 my.dic。直接在文件中添加词语即可，每一个词语一行。
2、修改ik的IKAnalyzer.cfg.xml配置文件。
3、将修改好的IK配置文件复制到集群中的所有节点中。
4、重启ES验证一下自定义词库的分词效果。

参考文章

7、ES如何分页？

默认情况下，不加from，size的话，ES会返回前10条记录。加上from，size就会查询指定的条数。其中from代表起始行号，size代表查询行数。

1、深度分页（from+size）
2、快照查询（scroll）
3、Search After

博主

标签

专辑

第三十五节 ElasticSearch存储订单并检索

一、需求分析

1. 电商平台的高效搜索体验

2. 复杂过滤与聚合分析

3. 实时数据同步与更新

4. 应对高并发与海量数据

5. 个性化推荐与用户行为分析

总结：现实需求的核心驱动

二、SpringBoot集成ElasticSearch

1、添加依赖

2、配置 Elasticsearch 连接

3、创建ES的实体类

4、创建索引

三、ElasticSearch数据操作

1、保存文档

2、根据ID读取文档

3、修改文档

4、删除文档

四、ElasticSearch 难点

1、多条件+排序+高亮+分词

五、拓展

1、什么是倒排索引？

3、ES脑裂是怎么回事？（扩展）

理解思路

4、“脑裂”问题是怎么形成的？（扩展）

5、脑裂问题如何解决？（扩展）

6、ES如何添加自定义词库、热更新词库？（扩展）

7、ES如何分页？