# 创建索引时指定分片数和副本数
PUT /product
{
  "settings": {
    "number_of_shards": 3,
    "number_of_replicas": 1
  },
  "mappings": { ... }
}

# 动态修改副本数（不需要 reindex）
PUT /product/_settings
{
  "number_of_replicas": 2
}

PUT /product
{
  "mappings": {
    "dynamic": "strict",            # 写入未定义字段直接报错
    "properties": {
      "name": { "type": "text", "analyzer": "ik_max_word" },
      "price": { "type": "double" }
    }
  }
}

"name": {
  "type": "text",
  "analyzer": "ik_max_word",        # 索引时：细粒度分词（更多词条=更高召回率）
  "search_analyzer": "ik_smart"     # 搜索时：粗粒度分词（更准匹配=更高精确率）
}

"largeDescription": {
  "type": "text",
  "index": false                    # 不需要搜索，只保留在 _source 中
}

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import jakarta.annotation.PreDestroy;

@Component
public class ProductBulkIngester {

    private final ElasticsearchClient esClient;
    private final BulkIngester<Product> bulkIngester;

    public ProductBulkIngester(ElasticsearchClient esClient) {
        this.esClient = esClient;

        this.bulkIngester = BulkIngester.of(builder -> builder
            .client(esClient)
            .maxOperations(2000)              // 每批最多 2000 条
            .maxSize(10 * 1024 * 1024)        // 每批最多 10MB
            .maxConcurrentRequests(3)         // 最多 3 个并发批量请求
            .flushInterval(5, TimeUnit.SECONDS) // 即使不满 2000 条，每 5 秒也执行一次
            .globalSettings(settings -> settings
                .index("product"))
            .listener(new BulkIngesterListener<Product>() {
                @Override
                public void beforeBulk(long executionId, BulkRequest request,
                                       List<Product> items) {
                    System.out.println("准备写入 " + items.size() + " 条");
                }

                @Override
                public void afterBulk(long executionId, BulkRequest request,
                                      List<Product> items, BulkResponse response) {
                    if (response.errors()) {
                        // 有失败的文档，需要重试
                        response.items().forEach(item -> {
                            if (item.error() != null) {
                                System.err.println("写入失败: " + item.id()
                                    + " | 原因: " + item.error().reason());
                            }
                        });
                    }
                }
            })
            .build());
    }

    public void add(Product product) {
        bulkIngester.add(BulkOperation.of(op -> op
            .index(idx -> idx
                .id(product.getId())
                .document(product))));
    }

    @PreDestroy
    public void close() {
        bulkIngester.close();  // 应用关闭时等待剩余数据写完
    }
}

# 不推荐：所有条件都放 must
{
  "bool": {
    "must": [
      { "match": { "name": "手机" } },
      { "term": { "brand": "华为" } },      # 算分用不到，浪费 CPU
      { "range": { "price": { "gte": 3000 } } }  # 算分用不到，浪费 CPU
    ]
  }
}

# 推荐：算分的放 must，过滤的放 filter
{
  "bool": {
    "must": [
      { "match": { "name": "手机" } }
    ],
    "filter": [
      { "term": { "brand": "华为" } },        # 走 LRU Query Cache
      { "range": { "price": { "gte": 3000 } } } # 走 LRU Query Cache
    ]
  }
}

PUT /product/_settings
{
  "index.max_result_window": 10000
}
# from + size 超过 10000 时直接报错，防止有人翻到第 10001 页拖垮集群

# 第 1 页
GET /product/_search
{
  "query": { "match": { "name": "手机" } },
  "sort": [
    { "soldCount": "desc" },
    { "_id": "asc" }               # 必须有 tiebreaker（保证排序唯一）
  ],
  "size": 10
}
# 记录最后一条的 sort 值：[8000, "abc123"]

# 第 2 页（基于上一页最后一条的 sort 值）
GET /product/_search
{
  "query": { "match": { "name": "手机" } },
  "sort": [
    { "soldCount": "desc" },
    { "_id": "asc" }
  ],
  "size": 10,
  "search_after": [8000, "abc123"]   # 从这条之后开始
}

// Java 代码：search_after
public List<Product> searchAfter(String keyword, int size, Object[] lastSortValues) {
    NativeQuery query = NativeQuery.builder()
        .withQuery(QueryBuilders.match().field("name").query(keyword).build())
        .withSort(Sort.by(
            new Sort.Order(Sort.Direction.DESC, "soldCount"),
            new Sort.Order(Sort.Direction.ASC, "_id")))
        .withPage(Pageable.ofSize(size))   // 不传 page 索引，只传 size
        .build();

    // 如果有上一页的最后一条 sort 值
    if (lastSortValues != null) {
        query.setSearchAfter(List.of(lastSortValues));
    }

    SearchHits<Product> hits = restTemplate.search(query, Product.class);
    List<Product> products = hits.stream()
        .map(SearchHit::getContent)
        .toList();

    // 记录最后一条的 sort 值，给下一页用
    if (!hits.isEmpty()) {
        SearchHit<Product> lastHit = hits.getSearchHits().get(hits.size() - 1);
        // lastHit.getSortValues() 就是这页最后一条的 sort 值
    }
    return products;
}

# 初始化 scroll（保持 2 分钟有效）
POST /product/_search?scroll=2m
{
  "query": { "match": { "name": "手机" } },
  "size": 1000
}
# 返回：{ "_scroll_id": "DXF1ZXJ5QW...", "hits": {...} }

# 后续翻页（使用 scroll_id）
POST /_search/scroll
{
  "scroll": "2m",
  "scroll_id": "DXF1ZXJ5QW..."
}

# 用完后清理 scroll
DELETE /_search/scroll
{
  "scroll_id": "DXF1ZXJ5QW..."
}

# 创建 PIT（保持 5 分钟）
POST /product/_pit?keep_alive=5m
# 返回：{ "id": "pit_id_xxx" }

# 使用 PIT 搜索
GET /_search
{
  "size": 100,
  "pit": {
    "id": "pit_id_xxx",
    "keep_alive": "5m"
  },
  "sort": [{ "_shard_doc": "asc" }],    # PIT 需要 _shard_doc 排序
  "search_after": [0]
}

# 删除 PIT
DELETE /_pit
{ "id": "pit_id_xxx" }

GET /product/_search
{
  "profile": true,                  # 开启性能分析
  "query": {
    "bool": {
      "must": [{ "match": { "name": "手机" } }],
      "filter": [{ "term": { "brand": "华为" } }]
    }
  }
}
# 返回结果会详细列出每个查询阶段（query / rewrite / advance / match 等）的耗时和遍历文档数

# 在 ES 配置中开启慢查询日志（也可以在单个 Index 级别设置）
PUT /product/_settings
{
  "index.search.slowlog.threshold.query.warn": "200ms",   # 超过 200ms 输出 WARN
  "index.search.slowlog.threshold.query.info": "100ms",   # 超过 100ms 输出 INFO
  "index.search.slowlog.threshold.fetch.warn": "100ms",   # fetch 阶段超过 100ms 输出 WARN
  "index.indexing.slowlog.threshold.index.warn": "200ms"  # 写入超过 200ms 输出 WARN
}

# 查看索引状态
GET /_cat/indices?v&s=store.size:desc
# 输出：index | docs.count | store.size | pri.store.size

# 查看分片分布（看有没有不均匀的）
GET /_cat/shards?v&s=store:desc
# 输出：index | shard | prirep | state | docs | store | node

# 查看节点级别搜索统计
GET /_nodes/stats/indices/search
# 输出：query_total / query_time_in_millis / fetch_total / fetch_time_in_millis

GET /product/_search
{
  "_source": ["name", "price", "soldCount"],   # 只返回需要的字段
  "query": { "match": { "name": "手机" } }
}

# 写入时带上 routing
POST /product/_doc/1?routing=tenant_001
{ "name": "...", "tenantId": "tenant_001" }

# 搜索时也可以带 routing（只查 tenant_001 的分片）
GET /product/_search?routing=tenant_001
{
  "query": { "match": { "name": "手机" } }
}

PUT /product
{
  "settings": {
    "number_of_shards": 3,              # 500 万 × 2KB ≈ 10GB → 3 个分片足够
    "number_of_replicas": 1,            # 1 个副本保证高可用
    "refresh_interval": "5s",           # 调大 refresh 间隔，降低写入压力
    "index.search.slowlog.threshold.query.warn": "200ms",
    "index.search.slowlog.threshold.fetch.info": "100ms"
  },
  "mappings": {
    "dynamic": "strict",                # 生产环境 strict
    "properties": {
      "name": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart",
        "fields": {
          "keyword": { "type": "keyword" }          # 精确匹配和排序用
        }
      },
      "brand":      { "type": "keyword" },
      "category":   { "type": "keyword" },
      "price":      { "type": "double" },
      "stock":      { "type": "integer" },
      "soldCount":  { "type": "integer" },
      "score":      { "type": "float" },
      "status":     { "type": "keyword" },          # "上架"/"下架"
      "createTime": { "type": "date", "format": "yyyy-MM-dd HH:mm:ss" },
      "description": {
        "type": "text",
        "analyzer": "ik_max_word",
        "search_analyzer": "ik_smart"
      },
      "detailHtml": { "type": "text", "index": false },  # 不需要搜索
      "imageUrls":  { "type": "keyword", "index": false }  # 不需要搜索

      # 自动补全专用字段
      "suggestName": {
        "type": "completion"
      }
    }
  }
}

@Component
public class ProductSyncService {

    // Canal 监听 MySQL binlog → RocketMQ → 这里消费
    @RocketMQMessageListener(topic = "product_change", consumerGroup = "es_sync")
    public class ProductChangeConsumer implements RocketMQListener<ProductChangeEvent> {

        @Autowired
        private ElasticsearchRestTemplate restTemplate;
        @Autowired
        private ProductBulkIngester bulkIngester;

        @Override
        public void onMessage(ProductChangeEvent event) {
            switch (event.getType()) {
                case "INSERT":
                case "UPDATE":
                    Product product = convertToProduct(event);
                    bulkIngester.add(product);          // 走 BulkIngester 异步批量写入
                    break;
                case "DELETE":
                    restTemplate.delete(event.getProductId(), Product.class);
                    break;
            }
        }
    }
}

@Service
public class ProductSearchService {

    private static final double FUNCTION_SCORE_FACTOR = 0.001;
    private static final double SCORE_FACTOR = 0.1;
    private static final int DEFAULT_PAGE_SIZE = 20;

    @Autowired
    private ElasticsearchRestTemplate restTemplate;

    /**
     * 商品搜索核心方法
     *
     * @param keyword        搜索关键词
     * @param brand          品牌筛选（可选）
     * @param category       分类筛选（可选）
     * @param minPrice       最低价（可选）
     * @param maxPrice       最高价（可选）
     * @param sortBy         排序方式（score/soldCount/price）
     * @param sortOrder      升序/降序
     * @param page           页码（0 开始）
     * @param lastSortValues 上一页最后一条的 sort 值（search_after 用）
     */
    public SearchResult search(String keyword, String brand, String category,
                                Double minPrice, Double maxPrice,
                                String sortBy, Sort.Direction sortOrder,
                                int page, Object[] lastSortValues) {

        // 1. 构建 bool 查询：must 只有搜索词，其他全放 filter
        BoolQuery.Builder boolQuery = QueryBuilders.bool();

        if (keyword != null && !keyword.isEmpty()) {
            boolQuery.must(QueryBuilders.multiMatch()
                .query(keyword)
                .fields(Map.of(
                    "name", 3.0f,        // 商品名权重最高
                    "brand", 1.5f,       // 品牌名次之
                    "description", 1.0f  // 描述权重最低
                ))
                .build());
        }

        if (brand != null) {
            boolQuery.filter(QueryBuilders.term()
                .field("brand").value(brand).build());
        }
        if (category != null) {
            boolQuery.filter(QueryBuilders.term()
                .field("category").value(category).build());
        }
        if (minPrice != null || maxPrice != null) {
            boolQuery.filter(QueryBuilders.range()
                .field("price")
                .gte(minPrice != null ? minPrice : 0.0)
                .lte(maxPrice != null ? maxPrice : Double.MAX_VALUE)
                .build());
        }
        // 只搜上架商品
        boolQuery.filter(QueryBuilders.term()
            .field("status").value("上架").build());

        // 2. function_score：综合销量和评分
        FunctionScoreQuery fsQuery = QueryBuilders.functionScore()
            .query(boolQuery.build())
            .functions(
                new FunctionScore.Builder()
                    .fieldValueFactor(fvf -> fvf
                        .field("soldCount")
                        .factor(FUNCTION_SCORE_FACTOR)
                        .modifier(FieldValueFactorModifier.Log1p))
                    .build(),
                new FunctionScore.Builder()
                    .fieldValueFactor(fvf -> fvf
                        .field("score")
                        .factor(SCORE_FACTOR)
                        .modifier(FieldValueFactorModifier.None))
                    .build()
            )
            .boostMode(FunctionBoostMode.Multiply)
            .scoreMode(FunctionScoreMode.Sum)
            .build();

        // 3. 排序
        Sort sort;
        if (keyword != null && ("score".equals(sortBy) || sortBy == null)) {
            // 有搜索词时默认按相关性排序（search_after 不能用 _score 做唯一排序）
            sort = Sort.by(
                new Sort.Order(Sort.Direction.DESC, "_score"),
                new Sort.Order(Sort.Direction.DESC, "_id"));
        } else {
            // 指定了业务字段排序
            Sort.Direction dir = sortOrder != null ? sortOrder
                : Sort.Direction.DESC;
            sort = Sort.by(
                new Sort.Order(dir, sortBy != null ? sortBy : "soldCount"),
                new Sort.Order(Sort.Direction.ASC, "_id"));
        }

        // 4. 构建查询
        NativeQuery query = NativeQuery.builder()
            .withQuery(fsQuery)
            .withAggregation("brand_agg",
                AggregationBuilders.terms().field("brand").build())
            .withAggregation("category_agg",
                AggregationBuilders.terms().field("category").build())
            .withHighlightQuery(new HighlightQuery(
                new Highlight(new HighlightParameters.Builder()
                    .withPreTags("<strong>")
                    .withPostTags("</strong>")
                    .build()),
                List.of(new HighlightField("name",
                    HighlightFieldParameters.builder()
                        .withFragmentSize(50)
                        .withNumberOfFragments(1)
                        .build()))))
            .withSort(sort)
            .withSourceFilter(new FetchSourceFilter(
                new String[]{"name", "brand", "category", "price",
                             "soldCount", "score", "createTime"},
                new String[]{}))
            .withPage(Pageable.ofSize(DEFAULT_PAGE_SIZE))
            .build();

        if (lastSortValues != null) {
            query.setSearchAfter(List.of(lastSortValues));
        }

        // 5. 执行搜索
        SearchHits<Product> hits = restTemplate.search(query, Product.class);

        // 6. 组装结果
        SearchResult result = new SearchResult();
        result.setTotal(hits.getTotalHits());

        List<ProductVO> products = new ArrayList<>();
        for (SearchHit<Product> hit : hits.getSearchHits()) {
            ProductVO vo = ProductVO.from(hit.getContent());
            List<String> hl = hit.getHighlightField("name");
            if (hl != null && !hl.isEmpty()) {
                vo.setHighlightName(hl.get(0));
            }
            // 记录 sort 值，给 search_after 用
            vo.setSortValues(hit.getSortValues().toArray());
            products.add(vo);
        }
        result.setProducts(products);

        // 7. 解析聚合结果（品牌、分类）—— 省略，在前面的聚合章节已讲过
        // parseAggregation("brand_agg", agg -> result.addBrand(agg.getKey(), agg.getDocCount()));
        // parseAggregation("category_agg", agg -> result.addCategory(agg.getKey(), agg.getDocCount()));

        return result;
    }
}