Elasticsearch 核心命脉：倒排索引、分片机制与全链路高性能调优实战

PUT /product_index
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1,
    "refresh_interval": "1s"
  },
  "mappings": {
    "properties": {
      "product_id": {
        "type": "keyword"
      },
      "product_name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "fields": {
          "keyword": {
            "type": "keyword",
            "ignore_above": 256
          }
        }
      },
      "category_id": {
        "type": "long"
      },
      "price": {
        "type": "double"
      },
      "stock": {
        "type": "integer"
      },
      "create_time": {
        "type": "date",
        "format": "yyyy-MM-dd HH:mm:ss||epoch_millis"
      }
    }
  }
}

PUT /product_index/_doc/1
{
  "product_id": "10001",
  "product_name": "华为Mate60 Pro 5G手机",
  "category_id": 1001,
  "price": 5999.00,
  "stock": 1000,
  "create_time": "2024-01-01 10:00:00"
}

POST /product_index/_analyze
{
  "analyzer": "ik_max_word",
  "text": "华为Mate60 Pro 5G手机"
}

GET /product_index/_doc/1/_termvectors?fields=product_name

shard = hash(routing) % number_of_primary_shards

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>3.3.0</version>
        <relativePath/>
    </parent>
    <groupId>com.jam.demo</groupId>
    <artifactId>es-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>es-demo</name>
    <properties>
        <java.version>17</java.version>
        <elasticsearch.version>8.15.0</elasticsearch.version>
        <mybatis-plus.version>3.5.7</mybatis-plus.version>
        <fastjson2.version>2.0.52</fastjson2.version>
        <lombok.version>1.18.30</lombok.version>
        <springdoc.version>2.5.0</springdoc.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>co.elastic.clients</groupId>
            <artifactId>elasticsearch-java</artifactId>
            <version>${elasticsearch.version}</version>
        </dependency>
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-elasticsearch</artifactId>
        </dependency>
        <dependency>
            <groupId>com.baomidou</groupId>
            <artifactId>mybatis-plus-boot-starter</artifactId>
            <version>${mybatis-plus.version}</version>
        </dependency>
        <dependency>
            <groupId>com.mysql</groupId>
            <artifactId>mysql-connector-j</artifactId>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>com.alibaba.fastjson2</groupId>
            <artifactId>fastjson2</artifactId>
            <version>${fastjson2.version}</version>
        </dependency>
        <dependency>
            <groupId>org.springdoc</groupId>
            <artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
            <version>${springdoc.version}</version>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>${lombok.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>com.google.guava</groupId>
            <artifactId>guava</artifactId>
            <version>33.0.0-jre</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

spring:
  application:
    name: es-demo
  datasource:
    url: jdbc:mysql://localhost:3306/test_db?useUnicode=true&characterEncoding=utf-8&serverTimezone=Asia/Shanghai&useSSL=false
    username: root
    password: root
    driver-class-name: com.mysql.cj.jdbc.Driver
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: Asia/Shanghai

elasticsearch:
  host: localhost
  port: 9200
  username: elastic
  password: elastic123

mybatis-plus:
  mapper-locations: classpath*:/mapper/**/*.xml
  type-aliases-package: com.jam.demo.entity
  configuration:
    map-underscore-to-camel-case: true
    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl

springdoc:
  swagger-ui:
    path: /swagger-ui.html
    enabled: true
  api-docs:
    enabled: true
    path: /v3/api-docs

CREATE TABLE `t_product` (
  `product_id` varchar(64) NOT NULL COMMENT '商品ID',
  `product_name` varchar(255) NOT NULL COMMENT '商品名称',
  `category_id` bigint NOT NULL COMMENT '分类ID',
  `price` decimal(10,2) NOT NULL COMMENT '商品价格',
  `stock` int NOT NULL DEFAULT '0' COMMENT '库存数量',
  `create_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  PRIMARY KEY (`product_id`),
  KEY `idx_category_id` (`category_id`),
  KEY `idx_create_time` (`create_time`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='商品表';

package com.jam.demo.config;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.client.RestClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Elasticsearch客户端配置类
 *
 * @author ken
 */
@Configuration
public class ElasticsearchConfig {

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

    @Value("${elasticsearch.port}")
    private int port;

    @Value("${elasticsearch.username}")
    private String username;

    @Value("${elasticsearch.password}")
    private String password;

    /**
     * 构建ElasticsearchClient实例
     *
     * @return ElasticsearchClient客户端实例
     */
    @Bean
    public ElasticsearchClient elasticsearchClient() {
        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        credentialsProvider.setCredentials(AuthScope.ANY,
                new UsernamePasswordCredentials(username, password));

        RestClient restClient = RestClient.builder(new HttpHost(host, port))
                .setHttpClientConfigCallback(httpClientBuilder -> httpClientBuilder
                        .setDefaultCredentialsProvider(credentialsProvider))
                .build();

        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());

        return new ElasticsearchClient(transport);
    }
}

package com.jam.demo.entity;

import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;

import java.io.Serializable;
import java.math.BigDecimal;
import java.time.LocalDateTime;

/**
 * 商品实体类
 *
 * @author ken
 */
@Data
@TableName("t_product")
@Schema(description = "商品实体")
public class Product implements Serializable {

    private static final long serialVersionUID = 1L;

    @TableId(type = IdType.ASSIGN_ID)
    @Schema(description = "商品ID", example = "10001")
    private String productId;

    @Schema(description = "商品名称", example = "华为Mate60 Pro 5G手机")
    private String productName;

    @Schema(description = "分类ID", example = "1001")
    private Long categoryId;

    @Schema(description = "商品价格", example = "5999.00")
    private BigDecimal price;

    @Schema(description = "库存数量", example = "1000")
    private Integer stock;

    @Schema(description = "创建时间", example = "2024-01-01 00:00:00")
    private LocalDateTime createTime;
}

package com.jam.demo.mapper;

import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.jam.demo.entity.Product;
import org.apache.ibatis.annotations.Mapper;

/**
 * 商品Mapper接口
 *
 * @author ken
 */
@Mapper
public interface ProductMapper extends BaseMapper<Product> {
}

package com.jam.demo.service;

import com.jam.demo.entity.Product;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;

/**
 * 商品ES服务接口
 *
 * @author ken
 */
public interface ProductEsService {

    void batchSaveProduct(List<Product> productList);

    List<Product> queryProductByFilter(Long categoryId, BigDecimal minPrice, BigDecimal maxPrice);

    List<Product> queryProductByRouting(Long categoryId, String keyword);

    List<Product> queryProductBySearchAfter(Long categoryId, Integer pageSize, LocalDateTime lastCreateTime, String lastProductId);
}

package com.jam.demo.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.BulkResponse;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.google.common.collect.Lists;
import com.jam.demo.entity.Product;
import com.jam.demo.service.ProductEsService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

/**
 * 商品ES服务实现类
 *
 * @author ken
 */
@Slf4j
@Service
public class ProductEsServiceImpl implements ProductEsService {

    private static final String INDEX_NAME = "product_index";
    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private final ElasticsearchClient elasticsearchClient;

    public ProductEsServiceImpl(ElasticsearchClient elasticsearchClient) {
        this.elasticsearchClient = elasticsearchClient;
    }

    @Override
    public void batchSaveProduct(List<Product> productList) {
        if (CollectionUtils.isEmpty(productList)) {
            return;
        }

        List<BulkOperation> bulkOperations = Lists.newArrayListWithCapacity(productList.size());
        for (Product product : productList) {
            if (ObjectUtils.isEmpty(product) || !StringUtils.hasText(product.getProductId())) {
                continue;
            }
            bulkOperations.add(BulkOperation.of(op -> op
                    .index(idx -> idx
                            .index(INDEX_NAME)
                            .id(product.getProductId())
                            .routing(product.getCategoryId().toString())
                            .document(product)
                    )
            ));
        }

        BulkRequest bulkRequest = BulkRequest.of(req -> req.operations(bulkOperations));
        try {
            BulkResponse bulkResponse = elasticsearchClient.bulk(bulkRequest);
            if (bulkResponse.errors()) {
                log.error("ES批量写入数据存在错误，错误信息：{}", bulkResponse.items().stream()
                        .filter(item -> !ObjectUtils.isEmpty(item.error()))
                        .map(item -> item.error().reason())
                        .toList()
                );
            }
        } catch (IOException e) {
            log.error("ES批量写入数据异常", e);
            throw new RuntimeException("ES批量写入数据异常", e);
        }
    }

    @Override
    public List<Product> queryProductByFilter(Long categoryId, BigDecimal minPrice, BigDecimal maxPrice) {
        if (ObjectUtils.isEmpty(categoryId)) {
            return Lists.newArrayList();
        }

        try {
            SearchResponse<Product> searchResponse = elasticsearchClient.search(req -> req
                            .index(INDEX_NAME)
                            .query(query -> query
                                    .bool(bool -> bool
                                            .filter(filter -> filter
                                                    .term(term -> term
                                                            .field("category_id")
                                                            .value(categoryId)
                                                    )
                                            )
                                            .filter(filter -> {
                                                if (ObjectUtils.isEmpty(minPrice) && ObjectUtils.isEmpty(maxPrice)) {
                                                    return filter.matchAll(m -> m);
                                                }
                                                return filter.range(range -> range
                                                        .field("price")
                                                        .gte(!ObjectUtils.isEmpty(minPrice) ? minPrice : null)
                                                        .lte(!ObjectUtils.isEmpty(maxPrice) ? maxPrice : null)
                                                );
                                            })
                                    )
                            ),
                    Product.class
            );

            return parseSearchResponse(searchResponse);
        } catch (IOException e) {
            log.error("ES过滤查询数据异常", e);
            throw new RuntimeException("ES过滤查询数据异常", e);
        }
    }

    @Override
    public List<Product> queryProductByRouting(Long categoryId, String keyword) {
        if (ObjectUtils.isEmpty(categoryId) || !StringUtils.hasText(keyword)) {
            return Lists.newArrayList();
        }

        try {
            SearchResponse<Product> searchResponse = elasticsearchClient.search(req -> req
                            .index(INDEX_NAME)
                            .routing(categoryId.toString())
                            .query(query -> query
                                    .match(match -> match
                                            .field("product_name")
                                            .query(keyword)
                                    )
                            ),
                    Product.class
            );

            return parseSearchResponse(searchResponse);
        } catch (IOException e) {
            log.error("ES路由查询数据异常", e);
            throw new RuntimeException("ES路由查询数据异常", e);
        }
    }

    @Override
    public List<Product> queryProductBySearchAfter(Long categoryId, Integer pageSize, LocalDateTime lastCreateTime, String lastProductId) {
        if (ObjectUtils.isEmpty(categoryId) || ObjectUtils.isEmpty(pageSize) || pageSize <= 0) {
            return Lists.newArrayList();
        }

        try {
            co.elastic.clients.elasticsearch.core.SearchRequest.Builder requestBuilder = new co.elastic.clients.elasticsearch.core.SearchRequest.Builder()
                    .index(INDEX_NAME)
                    .size(pageSize)
                    .sort(sort -> sort.field(f -> f.field("create_time").order(co.elastic.clients.elasticsearch.core.sort.SortOrder.Desc)))
                    .sort(sort -> sort.field(f -> f.field("product_id").order(co.elastic.clients.elasticsearch.core.sort.SortOrder.Asc)))
                    .query(query -> query
                            .term(term -> term
                                    .field("category_id")
                                    .value(categoryId)
                            )
                    );

            if (!ObjectUtils.isEmpty(lastCreateTime) && StringUtils.hasText(lastProductId)) {
                requestBuilder.searchAfter(
                        lastCreateTime.format(DATE_TIME_FORMATTER),
                        lastProductId
                );
            }

            SearchResponse<Product> searchResponse = elasticsearchClient.search(requestBuilder.build(), Product.class);
            return parseSearchResponse(searchResponse);
        } catch (IOException e) {
            log.error("ES分页查询数据异常", e);
            throw new RuntimeException("ES分页查询数据异常", e);
        }
    }

    /**
     * 解析ES查询响应结果
     *
     * @param searchResponse ES查询响应
     * @return 商品列表
     */
    private List<Product> parseSearchResponse(SearchResponse<Product> searchResponse) {
        List<Hit<Product>> hitList = searchResponse.hits().hits();
        if (CollectionUtils.isEmpty(hitList)) {
            return Lists.newArrayList();
        }

        List<Product> productList = Lists.newArrayListWithCapacity(hitList.size());
        for (Hit<Product> hit : hitList) {
            Product product = hit.source();
            if (!ObjectUtils.isEmpty(product)) {
                productList.add(product);
            }
        }
        return productList;
    }
}

package com.jam.demo.controller;

import com.jam.demo.entity.Product;
import com.jam.demo.mapper.ProductMapper;
import com.jam.demo.service.ProductEsService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.Parameter;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.util.CollectionUtils;
import org.springframework.web.bind.annotation.*;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.List;

/**
 * 商品ES操作Controller
 *
 * @author ken
 */
@RestController
@RequestMapping("/product/es")
@Tag(name = "商品ES操作接口", description = "商品ES数据写入与查询相关接口")
public class ProductEsController {

    private final ProductEsService productEsService;
    private final ProductMapper productMapper;

    public ProductEsController(ProductEsService productEsService, ProductMapper productMapper) {
        this.productEsService = productEsService;
        this.productMapper = productMapper;
    }

    @PostMapping("/batch/sync")
    @Operation(summary = "全量同步商品数据到ES", description = "从MySQL查询全量商品数据，批量写入到ES")
    public String syncAllProductToEs() {
        List<Product> productList = productMapper.selectList(null);
        if (CollectionUtils.isEmpty(productList)) {
            return "没有需要同步的商品数据";
        }
        productEsService.batchSaveProduct(productList);
        return "商品数据同步完成，同步数量：" + productList.size();
    }

    @GetMapping("/filter/query")
    @Operation(summary = "过滤条件查询商品", description = "基于分类ID和价格区间过滤查询商品")
    public List<Product> queryProductByFilter(
            @Parameter(description = "分类ID", required = true) @RequestParam Long categoryId,
            @Parameter(description = "最低价格") @RequestParam(required = false) BigDecimal minPrice,
            @Parameter(description = "最高价格") @RequestParam(required = false) BigDecimal maxPrice
    ) {
        return productEsService.queryProductByFilter(categoryId, minPrice, maxPrice);
    }

    @GetMapping("/routing/query")
    @Operation(summary = "路由查询商品", description = "基于分类ID作为路由，精准查询对应分片的商品数据")
    public List<Product> queryProductByRouting(
            @Parameter(description = "分类ID", required = true) @RequestParam Long categoryId,
            @Parameter(description = "搜索关键词", required = true) @RequestParam String keyword
    ) {
        return productEsService.queryProductByRouting(categoryId, keyword);
    }

    @GetMapping("/page/searchAfter")
    @Operation(summary = "search_after分页查询", description = "基于search_after实现高性能深度分页查询")
    public List<Product> queryProductBySearchAfter(
            @Parameter(description = "分类ID", required = true) @RequestParam Long categoryId,
            @Parameter(description = "每页条数", required = true) @RequestParam Integer pageSize,
            @Parameter(description = "上一页最后一条数据的创建时间") @RequestParam(required = false) LocalDateTime lastCreateTime,
            @Parameter(description = "上一页最后一条数据的商品ID") @RequestParam(required = false) String lastProductId
    ) {
        return productEsService.queryProductBySearchAfter(categoryId, pageSize, lastCreateTime, lastProductId);
    }
}