🔍 Elasticsearch 从零到生产

<!-- Spring Boot 3.x + ES 8.x --> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-elasticsearch</artifactId> </dependency>

# application.yml spring: elasticsearch: uris: http://localhost:9200 # 多节点集群 # uris: http://es-node1:9200,http://es-node2:9200,http://es-node3:9200 username: elastic password: ${ES_PASSWORD} connection-timeout: 5s socket-timeout: 30s

import org.springframework.data.annotation.Id; import org.springframework.data.elasticsearch.annotations.*; import java.time.LocalDateTime; import java.util.List; @Document(indexName = "products") @Setting(shards = 3, replicas = 1) public class Product { @Id private String id; @Field(type = FieldType.Text, analyzer = "ik_max_word", searchAnalyzer = "ik_smart") private String name; @Field(type = FieldType.Keyword) private String category; @Field(type = FieldType.Double) private Double price; @Field(type = FieldType.Boolean) private Boolean inStock; @Field(type = FieldType.Date, format = DateFormat.date_hour_minute_second) private LocalDateTime createdAt; @Field(type = FieldType.Keyword) private List<String> tags; // getter/setter... }

import org.springframework.data.elasticsearch.client.elc.NativeQuery; import org.springframework.data.elasticsearch.core.ElasticsearchOperations; import org.springframework.data.elasticsearch.core.SearchHit; import org.springframework.data.elasticsearch.core.SearchHits; import org.springframework.stereotype.Service; import co.elastic.clients.elasticsearch._types.query_dsl.*; import java.util.List; import java.util.stream.Collectors; @Service public class ProductSearchService { private final ElasticsearchOperations elasticsearchOperations; public ProductSearchService(ElasticsearchOperations ops) { this.elasticsearchOperations = ops; } // ===== 全文搜索 ===== public List<Product> searchByName(String keyword) { NativeQuery query = NativeQuery.builder() .withQuery(q -> q .match(m -> m .field("name") .query(keyword) ) ) .build(); SearchHits<Product> hits = elasticsearchOperations.search(query, Product.class); return hits.getSearchHits().stream() .map(SearchHit::getContent) .collect(Collectors.toList()); } // ===== 组合查询（bool） ===== public List<Product> searchProducts(String keyword, Double minPrice, Double maxPrice, String category) { NativeQuery query = NativeQuery.builder() .withQuery(q -> q .bool(b -> { // must: 全文搜索 if (keyword != null && !keyword.isEmpty()) { b.must(m -> m.match(mq -> mq.field("name").query(keyword))); } // filter: 不影响打分的条件 if (minPrice != null || maxPrice != null) { b.filter(f -> f.range(r -> { r.field("price"); if (minPrice != null) r.gte(minPrice.toString()); if (maxPrice != null) r.lte(maxPrice.toString()); return r; })); } if (category != null) { b.filter(f -> f.term(t -> t.field("category").value(category))); } return b; }) ) .withHighlightQuery(new HighlightQuery( new Highlight(List.of(new HighlightField("name"))), Product.class )) .build(); SearchHits<Product> hits = elasticsearchOperations.search(query, Product.class); return hits.getSearchHits().stream() .map(SearchHit::getContent) .collect(Collectors.toList()); } // ===== 聚合查询 ===== public List<CategoryStats> getCategoryStats() { NativeQuery query = NativeQuery.builder() .withQuery(q -> q.matchAll(m -> m)) .withAggregation("by_category", Aggregation.of(a -> a .terms(t -> t.field("category").size(20)) .aggregations("avg_price", sa -> sa .avg(av -> av.field("price")) ) )) .withMaxResults(0) // 只要聚合，不要文档 .build(); SearchHits<Product> hits = elasticsearchOperations.search(query, Product.class); // 解析聚合结果... return parseCategoryAggregation(hits); } }

public interface ProductRepository extends ElasticsearchRepository<Product, String> { // 方法名自动生成查询 List<Product> findByCategory(String category); List<Product> findByCategoryAndPriceBetween(String category, Double min, Double max); List<Product> findByNameOrDescription(String name, String desc); // 自定义查询（@Query） @Query("{\"bool\": {\"must\": [{\"match\": {\"name\": \"?0\"}}], \"filter\": [{\"range\": {\"price\": {\"gte\": \"?1\", \"lte\": \"?2\"}}}]}}") List<Product> searchByNameAndPriceRange(String keyword, Double minPrice, Double maxPrice); }

// 方案2 示例：MQ 异步同步 @Service public class ProductSyncService { private final ProductRepository productRepository; // MySQL private final ElasticsearchOperations esOps; // ES private final RabbitTemplate rabbitTemplate; // 写入 MySQL 后发 MQ 消息 @Transactional public Product createProduct(ProductDTO dto) { Product product = productRepository.save(dto.toEntity()); // 发送同步消息 rabbitTemplate.convertAndSend("es.sync.exchange", "product.sync", product.getId()); return product; } // 消费者：收到消息后同步到 ES @RabbitListener(queues = "es.sync.queue") public void syncToES(Long productId) { Product product = productRepository.findById(productId) .orElseThrow(); esOps.save(product); // 自动同步到 ES } }