Elasticsearch 核心概念与 Java 客户端实战

2. ES 核心架构解析

2.1 集群架构

节点类型：

• 主节点（Master）：管理集群状态、分片分配
• 数据节点（Data）：存储数据、执行 CRUD
• 协调节点（Coordinating）：路由请求、聚合结果
• 摄取节点（Ingest）：数据预处理

2.2 索引与分片

// 索引创建示例
@Configuration
public class IndexConfig {
    public void createProductIndex(RestHighLevelClient client) throws IOException {
        CreateIndexRequest request = new CreateIndexRequest("products");
        request.settings(Settings.builder()
            .put("index.number_of_shards", 3)
            .put("index.number_of_replicas", 1)
            .put("index.refresh_interval", "1s")
            .put("analysis.analyzer.default.type", "ik_max_word"));
        // Mapping 定义
        XContentBuilder mapping = JsonXContent.contentBuilder()
            .startObject()
            .startObject("properties")
            .startObject("id")
            .field("type", "keyword")
            .endObject()
            .startObject("name")
            .field("type", "text")
            .field("analyzer", "ik_max_word")
            .field("search_analyzer", "ik_smart")
            .endObject()
            .endObject()
            .endObject();
        request.mapping(mapping);
        CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT);
    }
}

分片设计原则：

1. 单个分片不超过 50GB
2. 分片数 = 数据总量 / 50GB
3. 副本数 = 节点数 - 1（至少 1 个）
4. 避免过度分片（每个分片有开销）

3. Java 客户端实战

3.1 客户端选型对比

3.2 RestHighLevelClient 配置

@Configuration
@Slf4j
public class ElasticsearchConfig {
    @Value("${elasticsearch.hosts:localhost:9200}")
    private String hosts;

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        String[] hostArray = hosts.split(",");
        HttpHost[] httpHosts = new HttpHost[hostArray.length];
        for (int i = 0; i < hostArray.length; i++) {
            String[] hostPort = hostArray[i].split(":");
            httpHosts[i] = new HttpHost(hostPort[0].trim(), Integer.parseInt(hostPort[1].trim()), "http");
        }
        RestClientBuilder builder = RestClient.builder(httpHosts)
            .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                .setConnectTimeout(5000)
                .setSocketTimeout(60000))
            .setHttpClientConfigCallback(httpClientBuilder -> {
                httpClientBuilder.setMaxConnTotal(100)
                    .setMaxConnPerRoute(50);
                return httpClientBuilder;
            });
        return new RestHighLevelClient(builder);
    }
}

3.3 Spring Data Elasticsearch 配置

@Configuration
@EnableElasticsearchRepositories(basePackages = "com.example.repository")
public class SpringDataESConfig {
    @Document(indexName = "products", createIndex = false)
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public class Product {
        @Id
        private String id;
        @Field(type = FieldType.Text, analyzer = "ik_max_word")
        private String name;
        @Field(type = FieldType.Double)
        private Double price;
    }

    public interface ProductRepository extends ElasticsearchRepository<Product, String> {
        List<Product> findByName(String name);
        Page<Product> findByCategoryId(Integer categoryId, Pageable pageable);
    }
}

4. 索引设计最佳实践

4.1 索引生命周期管理

@Component
public class IndexLifecycleManager {
    public void rolloverIndex() throws IOException {
        String todayIndex = getDailyIndex("logs");
        String alias = "logs-current";
        // 检查当前索引大小，超过 50GB 或创建超过 7 天，创建新索引
        if (stats.getStoreSizeInBytes() > 50L * 1024 * 1024 * 1024) {
            createIndexWithAlias(todayIndex, alias);
            switchAlias(alias, currentIndex, todayIndex);
        }
    }
}

4.2 映射设计技巧

{
  "mappings": {
    "dynamic": "strict",
    "properties": {
      "id": { "type": "keyword" },
      "title": { "type": "text", "analyzer": "ik_max_word" },
      "price": { "type": "scaled_float", "scaling_factor": 100 },
      "location": { "type": "geo_point" },
      "vector": { "type": "dense_vector", "dims": 128 }
    }
  }
}

5. 查询优化实战

5.1 查询类型对比

@Service
public class ProductSearchService {
    public SearchResponse searchByMatch(String keyword) throws IOException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchQuery("name", keyword)
            .analyzer("ik_smart")
            .operator(Operator.AND));
        sourceBuilder.size(20);
        return client.search(new SearchRequest("products").source(sourceBuilder), RequestOptions.DEFAULT);
    }

    public SearchResponse searchByBool(String keyword, Double minPrice, Double maxPrice) throws IOException {
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        if (StringUtils.isNotBlank(keyword)) {
            boolQuery.must(QueryBuilders.matchQuery("name", keyword));
        }
        if (minPrice != null || maxPrice != null) {
            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery("price");
            if (minPrice != null) rangeQuery.gte(minPrice);
            if (maxPrice != null) rangeQuery.lte(maxPrice);
            boolQuery.filter(rangeQuery);
        }
        return executeSearch(boolQuery);
    }
}

5.2 性能优化技巧

@Component
public class QueryOptimizer {
    public SearchResponse searchWithPagination(String keyword, int page, int size) throws IOException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 推荐：Search After 分页
        if (page > 1) {
            Object[] searchAfter = getSearchAfterValues(page - 1, size);
            sourceBuilder.searchAfter(searchAfter);
        }
        sourceBuilder.sort(SortBuilders.fieldSort("_id"));
        return executeSearch(sourceBuilder);
    }

    public SearchResponse optimizedSearch(String keyword) throws IOException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        // 避免 wildcard 查询，使用 ngram 或 prefix
        QueryBuilders.matchQuery("name.ngram", keyword);
        return executeSearch(sourceBuilder);
    }
}

6. 批量操作与实时性

6.1 Bulk 批量操作

@Component
public class BulkOperationService {
    public void bulkIndexProducts(List<Product> products) throws IOException {
        BulkRequest request = new BulkRequest();
        for (Product product : products) {
            IndexRequest indexRequest = new IndexRequest("products")
                .id(product.getId())
                .source(JsonUtils.toJson(product), XContentType.JSON);
            request.add(indexRequest);
        }
        BulkResponse response = client.bulk(request, RequestOptions.DEFAULT);
    }

    public void bulkWithListener(List<Product> products) {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                if (response.hasFailures()) {
                    log.error("Bulk 执行失败：{}", response.buildFailureMessage());
                }
            }
        };
        BulkProcessor processor = BulkProcessor.builder(
            (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener),
            listener
        ).setBulkActions(1000).build();
        for (Product product : products) {
            processor.add(new IndexRequest("products").id(product.getId()).source(JsonUtils.toJson(product), XContentType.JSON));
        }
        processor.awaitClose(30, TimeUnit.SECONDS);
    }
}

6.2 实时性控制

@Component
public class RealtimeControlService {
    public void indexWithRefresh(Product product) throws IOException {
        IndexRequest request = new IndexRequest("products")
            .id(product.getId())
            .source(JsonUtils.toJson(product), XContentType.JSON)
            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
        client.index(request, RequestOptions.DEFAULT);
    }

    public void updateRefreshInterval(int seconds) throws IOException {
        UpdateSettingsRequest request = new UpdateSettingsRequest("products");
        Settings settings = Settings.builder()
            .put("index.refresh_interval", seconds + "s")
            .build();
        client.indices().putSettings(request, request.settings(settings), RequestOptions.DEFAULT);
    }
}

7. 企业级实战案例

7.1 电商商品搜索系统

@Service
public class ECommerceSearchService {
    public SearchResult<Product> searchProducts(ProductSearchRequest request) throws IOException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        
        if (StringUtils.isNotBlank(request.getKeyword())) {
            boolQuery.should(QueryBuilders.matchPhraseQuery("name", request.getKeyword()).boost(3.0f));
            boolQuery.should(QueryBuilders.matchQuery("name.pinyin", request.getKeyword()).boost(2.0f));
        }
        
        if (request.getMinPrice() != null) {
            boolQuery.filter(QueryBuilders.rangeQuery("price").gte(request.getMinPrice()));
        }
        
        sourceBuilder.query(boolQuery);
        sourceBuilder.from((request.getPage() - 1) * request.getSize())
                     .size(request.getSize());
        
        HighlightBuilder highlightBuilder = new HighlightBuilder();
        highlightBuilder.field("name").preTags("<em>").postTags("</em>");
        sourceBuilder.highlighter(highlightBuilder);
        
        return processSearchResult(client.search(new SearchRequest("products").source(sourceBuilder), RequestOptions.DEFAULT), request);
    }
}

7.2 日志分析系统

@Service
public class LogAnalysisService {
    public LogSearchResult searchLogs(LogSearchRequest request) throws IOException {
        String index = getLogIndex(request.getStartTime(), request.getEndTime());
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        
        RangeQueryBuilder timeRange = QueryBuilders.rangeQuery("@timestamp")
            .gte(request.getStartTime())
            .lte(request.getEndTime())
            .format("yyyy-MM-dd HH:mm:ss");
        
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery()
            .filter(timeRange)
            .must(QueryBuilders.queryStringQuery(request.getKeyword())
                .field("message")
                .defaultOperator(Operator.AND));
        
        sourceBuilder.query(boolQuery);
        sourceBuilder.sort(SortBuilders.fieldSort("@timestamp").order(SortOrder.DESC));
        
        return executeSearch(index, sourceBuilder);
    }
}

8. 性能优化与监控

8.1 性能调优

@Component
public class PerformanceTuner {
    public Settings getOptimizedSettings() {
        return Settings.builder()
            .put("index.number_of_shards", 3)
            .put("index.number_of_replicas", 1)
            .put("index.refresh_interval", "30s")
            .put("index.translog.durability", "async")
            .put("index.merge.scheduler.max_thread_count", 1)
            .build();
    }

    public SearchSourceBuilder tuneSearch(SearchSourceBuilder builder) {
        builder.query(QueryBuilders.boolQuery()
            .must(QueryBuilders.matchQuery("name", "手机"))
            .filter(QueryBuilders.termQuery("status", "active")))
            .requestCache(true);
        builder.timeout(TimeValue.timeValueSeconds(5));
        builder.fetchSource(new String[]{"id", "name", "price"}, null);
        return builder;
    }
}

8.2 监控告警

# prometheus 配置
scrape_configs:
  - job_name: 'elasticsearch'
    static_configs:
      - targets: ['localhost:9200']
    metrics_path: '/_prometheus/metrics'

alerting_rules:
  - alert: ClusterHealthRed
    expr: elasticsearch_cluster_health_status{color="red"} == 1
    labels:
      severity: critical

9. 故障排查指南

9.1 常见问题排查

@Component
public class TroubleshootingGuide {
    public void diagnoseSlowSearch(String index, String query) throws IOException {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.query(QueryBuilders.matchQuery("name", query));
        sourceBuilder.profile(true);
        SearchRequest request = new SearchRequest(index);
        request.source(sourceBuilder);
        SearchResponse response = client.search(request, RequestOptions.DEFAULT);
        // 分析 profile 结果
        Map<String, ProfileShardResult> profileResults = response.getProfileResults();
    }

    public void diagnoseUnassignedShards() throws IOException {
        ClusterAllocationExplainRequest request = new ClusterAllocationExplainRequest();
        ClusterAllocationExplainResponse response = client.cluster().allocationExplain(request, RequestOptions.DEFAULT);
        log.info("分片分配解释：{}", response.getExplanation());
    }
}

10. 选型与总结

10.1 ES vs 其他方案对比

10.2 最佳实践

1. 分片设计要合理：单个分片不超过 50GB
2. 映射设计要严谨：禁用动态映射，明确字段类型
3. 查询要优化：避免 wildcard，善用 filter
4. 监控要全面：集群健康、性能指标、业务指标
5. 容量要规划：提前规划扩容，设置水位线
6. 备份要定期：定期快照，测试恢复

参考资料

1. Elasticsearch 官方文档 - https://www.elastic.co/guide/en/elasticsearch/reference/current/index.html
2. Java 客户端文档 - https://www.elastic.co/guide/en/elasticsearch/client/java-rest/current/index.html
3. Elasticsearch 源码 - https://github.com/elastic/elasticsearch
4. Lucene 源码 - https://github.com/apache/lucene
5. Prometheus 监控 - https://prometheus.io/