Elasticsearch 实战指南：从入门到生产

Elasticsearch 实战指南从入门到生产ES 不是只有搜索那么简单。用好它日志分析、指标监控、推荐系统都能搞定。用不好集群黄了、查询慢了、数据丢了全是坑。这篇文章总结了我在多个项目里踩过的坑和验证过的做法。一、基础使用1.1 核心概念别搞混Index数据集合类似数据库的表MySQL 的 tableDocument一条数据记录MySQL 的一行Mapping字段类型定义表结构Shard数据分片数据分区Replica副本分片主从备份关键理解Index 是逻辑概念Shard 是物理概念。一个 Index 默认 1 个主分片 1 个副本分片。1.2 快速启动Docker 版# 单节点开发环境 docker run -d \ --name elasticsearch \ -p 9200:9200 \ -e discovery.typesingle-node \ -e xpack.security.enabledfalse \ -e ES_JAVA_OPTS-Xms512m -Xmx512m \ docker.elastic.co/elasticsearch/elasticsearch:8.11.0 # 验证启动 curl http://localhost:92001.3 基础 CRUD创建索引显式定义 mappingcurl -X PUT localhost:9200/products -H Content-Type: application/json -d{ mappings: { properties: { name: { type: text, analyzer: ik_max_word }, price: { type: float }, category: { type: keyword }, created_at: { type: date } } }, settings: { number_of_shards: 3, number_of_replicas: 1 } }插入文档curl -X POST localhost:9200/products/_doc -H Content-Type: application/json -d{ name: iPhone 15 Pro, price: 7999.00, category: 手机, created_at: 2024-01-15 }搜索curl -X GET localhost:9200/products/_search -H Content-Type: application/json -d{ query: { match: { name: iPhone } }, sort: [ { price: asc } ] }1.4 常用查询类型全文搜索分词匹配{ query: { match: { title: 机器学习 }} }精确匹配不分词{ query: { term: { status: published }} }范围查询{ query: { range: { price: { gte: 100, lte: 500 } } } }多条件组合{ query: { bool: { must: [ { match: { title: 教程 }}, { range: { price: { lte: 100 }}} ], must_not: [ { term: { status: deleted }} ] } } }聚合统计{ aggs: { by_category: { terms: { field: category } }, avg_price: { avg: { field: price } } } }二、实际接入项目Java 版2.1 项目结构以电商搜索为例project/ ├── src/main/java/com/example/es/ │ ├── config/ │ │ └── ElasticsearchConfig.java # ES 连接配置 │ ├── model/ │ │ └── ProductDocument.java # 文档实体类 │ ├── repository/ │ │ └── ProductRepository.java # 数据访问层 │ ├── service/ │ │ ├── ProductIndexer.java # 数据同步服务 │ │ └── ProductSearchService.java # 搜索业务逻辑 │ └── controller/ │ └── SearchController.java # API 接口 └── src/main/resources/ └── application.yml2.2 Maven 依赖dependencies !-- Elasticsearch Java API Client -- dependency groupIdco.elastic.clients/groupId artifactIdelasticsearch-java/artifactId version8.11.0/version /dependency !-- Jackson JSON 处理器 -- dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId version2.15.2/version /dependency !-- Jakarta JSON API -- dependency groupIdjakarta.json/groupId artifactIdjakarta.json-api/artifactId version2.1.2/version /dependency /dependencies2.3 配置类application.ymlelasticsearch: host: localhost port: 9200 username: # 如有认证 password: connection-timeout: 5000 socket-timeout: 30000ElasticsearchConfig.javapackage com.example.es.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.elasticsearch.client.RestClientBuilder; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; Configuration public class ElasticsearchConfig { Value(${elasticsearch.host:localhost}) private String host; Value(${elasticsearch.port:9200}) private int port; Value(${elasticsearch.username:}) private String username; Value(${elasticsearch.password:}) private String password; Bean public ElasticsearchClient elasticsearchClient() { RestClientBuilder builder RestClient.builder( new HttpHost(host, port) ); // 配置认证如有 if (!username.isEmpty()) { CredentialsProvider credentialsProvider new BasicCredentialsProvider(); credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password)); builder.setHttpClientConfigCallback(httpClientBuilder - httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider) ); } ElasticsearchTransport transport new RestClientTransport( builder.build(), new JacksonJsonpMapper() ); return new ElasticsearchClient(transport); } }2.4 文档实体类ProductDocument.javapackage com.example.es.model; import com.fasterxml.jackson.annotation.JsonFormat; import java.time.LocalDateTime; public class ProductDocument { private String skuId; private String name; private String categoryPath; private Float price; private Integer stock; private Integer salesCount; private String shopId; private String shopName; private Object attributes; JsonFormat(pattern yyyy-MM-ddTHH:mm:ss) private LocalDateTime createdAt; // Getters and Setters public String getSkuId() { return skuId; } public void setSkuId(String skuId) { this.skuId skuId; } public String getName() { return name; } public void setName(String name) { this.name name; } public String getCategoryPath() { return categoryPath; } public void setCategoryPath(String categoryPath) { this.categoryPath categoryPath; } public Float getPrice() { return price; } public void setPrice(Float price) { this.price price; } public Integer getStock() { return stock; } public void setStock(Integer stock) { this.stock stock; } public Integer getSalesCount() { return salesCount; } public void setSalesCount(Integer salesCount) { this.salesCount salesCount; } public String getShopId() { return shopId; } public void setShopId(String shopId) { this.shopId shopId; } public String getShopName() { return shopName; } public void setShopName(String shopName) { this.shopName shopName; } public Object getAttributes() { return attributes; } public void setAttributes(Object attributes) { this.attributes attributes; } public LocalDateTime getCreatedAt() { return createdAt; } public void setCreatedAt(LocalDateTime createdAt) { this.createdAt createdAt; } }2.5 索引管理ProductIndexService.javapackage com.example.es.service; import co.elastic.clients.elasticsearch.ElasticsearchClient; import co.elastic.clients.elasticsearch._types.mapping.Property; import co.elastic.clients.elasticsearch.indices.CreateIndexRequest; import co.elastic.clients.elasticsearch.indices.CreateIndexResponse; import co.elastic.clients.elasticsearch.indices.ExistsRequest; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.HashMap; import java.util.Map; Service public class ProductIndexService { Autowired private ElasticsearchClient client; private static final String INDEX_NAME products_v1; public boolean createIndex() throws IOException { // 检查索引是否存在 boolean exists client.indices().exists( ExistsRequest.of(e - e.index(INDEX_NAME)) ).value(); if (exists) { return false; } // 定义 mapping MapString, Property properties new HashMap(); properties.put(skuId, Property.of(p - p.keyword(k - k))); properties.put(name, Property.of(p - p.text(t - t .analyzer(ik_max_word) .searchAnalyzer(ik_smart) ))); properties.put(categoryPath, Property.of(p - p.keyword(k - k))); properties.put(price, Property.of(p - p.float_(f - f))); properties.put(stock, Property.of(p - p.integer(i - i))); properties.put(salesCount, Property.of(p - p.integer(i - i))); properties.put(shopId, Property.of(p - p.keyword(k - k))); properties.put(shopName, Property.of(p - p.keyword(k - k))); properties.put(attributes, Property.of(p - p.object(o - o))); properties.put(createdAt, Property.of(p - p.date(d - d.format(yyyy-MM-ddTHH:mm:ss)))); // 创建索引 CreateIndexRequest request CreateIndexRequest.of(c - c .index(INDEX_NAME) .mappings(m - m.properties(properties)) .settings(s - s .numberOfShards(5) .numberOfReplicas(1) .refreshInterval(5s) ) ); CreateIndexResponse response client.indices().create(request); return response.acknowledged(); } public void deleteIndex() throws IOException { client.indices().delete(d - d.index(INDEX_NAME)); } }2.6 商品搜索服务ProductSearchService.javapackage com.example.es.service; import co.elastic.clients.elasticsearch.ElasticsearchClient; import co.elastic.clients.elasticsearch._types.SortOrder; import co.elastic.clients.elasticsearch._types.mapping.FieldMapping; import co.elastic.clients.elasticsearch.core.*; import co.elastic.clients.elasticsearch.core.search.Hit; import co.elastic.clients.elasticsearch.core.search.HighlightField; import com.example.es.model.ProductDocument; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.IOException; import java.util.*; import java.util.stream.Collectors; Service public class ProductSearchService { Autowired private ElasticsearchClient client; private static final String INDEX_NAME products_v1; /** * 商品搜索 */ public SearchResult search(String keyword, MapString, Object filters, int page, int size) throws IOException { int from (page - 1) * size; // 构建查询 SearchRequest.Builder searchBuilder new SearchRequest.Builder() .index(INDEX_NAME) .from(from) .size(size); // 构建 bool 查询 Listco.elastic.clients.elasticsearch._types.query_dsl.Query mustQueries new ArrayList(); Listco.elastic.clients.elasticsearch._types.query_dsl.Query filterQueries new ArrayList(); // 关键词搜索多字段 if (keyword ! null !keyword.isEmpty()) { mustQueries.add(co.elastic.clients.elasticsearch._types.query_dsl.Query.of(q - q .multiMatch(m - m .query(keyword) .fields(name^3, category_path, shop_name) ) )); } // 过滤条件 if (filters ! null) { if (filters.containsKey(category)) { filterQueries.add(co.elastic.clients.elasticsearch._types.query_dsl.Query.of(q - q .term(t - t.field(categoryPath).value(v - v.stringValue((String) filters.get(category)))) )); } if (filters.containsKey(priceMin) || filters.containsKey(priceMax)) { filterQueries.add(co.elastic.clients.elasticsearch._types.query_dsl.Query.of(q - q .range(r - { r.field(price); if (filters.containsKey(priceMin)) { r.gte(JsonData.of(filters.get(priceMin))); } if (filters.containsKey(priceMax)) { r.lte(JsonData.of(filters.get(priceMax))); } return r; }) )); } } // 组合 bool 查询 searchBuilder.query(q - q .bool(b - { b.must(mustQueries); b.filter(filterQueries); return b; }) ); // 排序 searchBuilder.sort(s - s .score(sc - sc.order(SortOrder.Desc)) ); searchBuilder.sort(s - s .field(f - f.field(salesCount).order(SortOrder.Desc)) ); // 高亮 searchBuilder.highlight(h - h .fields(name, HighlightField.of(hf - hf)) ); // 执行搜索 SearchResponseProductDocument response client.search( searchBuilder.build(), ProductDocument.class ); // 解析结果 ListProductDocument items new ArrayList(); MapString, MapString, ListString highlights new HashMap(); for (HitProductDocument hit : response.hits().hits()) { ProductDocument doc hit.source(); items.add(doc); if (hit.highlight() ! null) { MapString, ListString docHighlights new HashMap(); hit.highlight().forEach((field, values) - docHighlights.put(field, values) ); highlights.put(hit.id(), docHighlights); } } return new SearchResult( (int) response.hits().total().value(), items, highlights ); } /** * 根据ID获取商品 */ public ProductDocument getById(String id) throws IOException { GetResponseProductDocument response client.get(g - g .index(INDEX_NAME) .id(id), ProductDocument.class ); return response.found() ? response.source() : null; } /** * 索引商品新增或更新 */ public void indexProduct(ProductDocument product) throws IOException { client.index(i - i .index(INDEX_NAME) .id(product.getSkuId()) .document(product) ); } /** * 批量索引 */ public void bulkIndex(ListProductDocument products) throws IOException { BulkRequest.Builder bulkBuilder new BulkRequest.Builder(); for (ProductDocument product : products) { bulkBuilder.operations(op - op .index(idx - idx .index(INDEX_NAME) .id(product.getSkuId()) .document(product) ) ); } client.bulk(bulkBuilder.build()); } /** * 删除商品 */ public void deleteProduct(String id) throws IOException { client.delete(d - d.index(INDEX_NAME).id(id)); } // 搜索结果封装类 public static class SearchResult { private final int total; private final ListProductDocument items; private final MapString, MapString, ListString highlights; public SearchResult(int total, ListProductDocument items, MapString, MapString, ListString highlights) { this.total total; this.items items; this.highlights highlights; } public int getTotal() { return total; } public ListProductDocument getItems() { return items; } public MapString, MapString, ListString getHighlights() { return highlights; } } }2.7 数据同步服务监听 BinlogProductSyncService.javapackage com.example.es.service; import com.example.es.model.ProductDocument; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.stereotype.Service; import java.io.IOException; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; Service public class ProductSyncService { Autowired private ProductSearchService searchService; private final ObjectMapper objectMapper new ObjectMapper(); KafkaListener(topics db_product, groupId es-sync-group) public void consume(ConsumerRecordString, String record) throws IOException { JsonNode data objectMapper.readTree(record.value()); String type data.get(type).asText(); JsonNode payload data.get(data); switch (type) { case INSERT: case UPDATE: ProductDocument doc convertToDocument(payload); searchService.indexProduct(doc); break; case DELETE: String id payload.get(id).asText(); searchService.deleteProduct(id); break; } } private ProductDocument convertToDocument(JsonNode payload) { ProductDocument doc new ProductDocument(); doc.setSkuId(payload.get(id).asText()); doc.setName(payload.get(name).asText()); doc.setCategoryPath(payload.get(category_path).asText()); doc.setPrice(payload.get(price).floatValue()); doc.setStock(payload.get(stock).asInt()); doc.setSalesCount(payload.get(sales_count).asInt()); doc.setShopId(payload.get(shop_id).asText()); doc.setShopName(payload.get(shop_name).asText()); String dateStr payload.get(created_at).asText(); doc.setCreatedAt(LocalDateTime.parse(dateStr, DateTimeFormatter.ofPattern(yyyy-MM-dd HH:mm:ss))); return doc; } }2.8 数据同步方案对比方案适用场景延迟复杂度双写数据量小实时性高实时低监听 Binlog已有 MySQL不想改代码秒级中定时同步离线分析允许延迟分钟级低CDC (Debezium)大规模多数据源秒级高推荐Binlog 监听方案Canal/Maxwell Kafka三、适配的业务场景3.1 站内搜索最常用电商商品搜索文档/知识库检索内容社区搜索特点需要分词、相关性排序、facets 过滤3.2 日志分析ELK 栈Filebeat 配置filebeat.inputs: - type: log enabled: true paths: - /var/log/app/*.log multiline.pattern: ^[ multiline.negate: true multiline.match: after fields: service: order-service env: production output.elasticsearch: hosts: [es-node1:9200, es-node2:9200] index: app-logs-%{yyyy.MM.dd}典型查询GET /app-logs-*/_search { query: { bool: { must: [ { match: { level: ERROR }}, { range: { timestamp: { gte: now-1h, lte: now } }} ] } }, aggs: { error_by_service: { terms: { field: service.keyword } } } }