利用 SpringBoot 在 ES 中實現類似連表查詢功能
一、摘要
在上篇文章中,我們詳細的介紹瞭如何在 ES 中精準的實現嵌套json對象查詢?
那麼問題來瞭,我們如何在後端通過技術方式快速的實現 es 中內嵌對象的數據查詢呢?
為瞭方便更容易掌握技術,本文主要以上篇文章中介紹的通過商品找訂單為案例,利用 SpringBoot 整合 ES 實現這個業務需求,向大傢介紹具體的技術實踐方案,存入es中的json數據結構如下:
{ "orderId":"1", "orderNo":"123456", "orderUserName":"張三", "orderItems":[ { "orderItemId":"12234", "orderId":"1", "productName":"火腿腸", "brandName":"雙匯", "sellPrice":"28" }, { "orderItemId":"12235", "orderId":"1", "productName":"果凍", "brandName":"匯源", "sellPrice":"12" } ] }
廢話也不多說瞭,直接上代碼!
二、項目實踐
2.1添加依賴
在SpringBoot項目中,添加rest-high-level-client客戶端,方便與 ES 服務器連接通信,在這裡需要註意一下,推薦客戶端的版本與 ES 服務器的版本號一致,不然會出現接口請求錯誤等異常!
小編本次安裝的ES服務端版本號為6.8.2,因此客戶端也保持6.8.2,與之一致!
<!--elasticsearch--> <dependency> <groupId>org.elasticsearch</groupId> <artifactId>elasticsearch</artifactId> <version>6.8.2</version> </dependency> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-client</artifactId> <version>6.8.2</version> </dependency> <dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>6.8.2</version> </dependency>
2.2配置 es 客戶端
為瞭更佳方便的使用 es,我們可以將其各個配置類進行封裝,方便後續進行維護。
- 在application.properties配置文件中,定義 es 配置連接地址;
# 設置es參數 elasticsearch.scheme=http elasticsearch.address=127.0.0.1:9200 elasticsearch.userName= elasticsearch.userPwd= elasticsearch.socketTimeout=5000 elasticsearch.connectTimeout=5000 elasticsearch.connectionRequestTimeout=5000
- 創建ElasticSearch配置類,方便SpringBoot啟動時註入;
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.elasticsearch.client.RestHighLevelClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import java.util.Arrays; import java.util.Objects; @Configuration public class ElasticSearchConfiguration { private static final Logger log = LoggerFactory.getLogger(ElasticSearchConfiguration.class); private static final int ADDRESS_LENGTH = 2; @Value("${elasticsearch.scheme:http}") private String scheme; @Value("${elasticsearch.address}") private String address; @Value("${elasticsearch.userName}") private String userName; @Value("${elasticsearch.userPwd}") private String userPwd; @Value("${elasticsearch.socketTimeout:5000}") private Integer socketTimeout; @Value("${elasticsearch.connectTimeout:5000}") private Integer connectTimeout; @Value("${elasticsearch.connectionRequestTimeout:5000}") private Integer connectionRequestTimeout; /** * 初始化客戶端 * @return */ @Bean(name = "restHighLevelClient") public RestHighLevelClient restClientBuilder() { HttpHost[] hosts = Arrays.stream(address.split(",")) .map(this::buildHttpHost) .filter(Objects::nonNull) .toArray(HttpHost[]::new); RestClientBuilder restClientBuilder = RestClient.builder(hosts); // 異步參數配置 restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> { httpClientBuilder.setDefaultCredentialsProvider(buildCredentialsProvider()); return httpClientBuilder; }); // 異步連接延時配置 restClientBuilder.setRequestConfigCallback(requestConfigBuilder -> { requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeout); requestConfigBuilder.setSocketTimeout(socketTimeout); requestConfigBuilder.setConnectTimeout(connectTimeout); return requestConfigBuilder; }); return new RestHighLevelClient(restClientBuilder); } /** * 根據配置創建HttpHost * @param s * @return */ private HttpHost buildHttpHost(String s) { String[] address = s.split(":"); if (address.length == ADDRESS_LENGTH) { String ip = address[0]; int port = Integer.parseInt(address[1]); return new HttpHost(ip, port, scheme); } else { return null; } } /** * 構建認證服務 * @return */ private CredentialsProvider buildCredentialsProvider(){ final CredentialsProvider credentialsProvider = new BasicCredentialsProvider(); credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(userName, userPwd)); return credentialsProvider; } }
- 封裝ElasticSearch客戶端服務類,方便公共調用處理
import com.fasterxml.jackson.databind.ObjectMapper; import org.example.es.exception.CommonException; import org.apache.commons.lang3.StringUtils; import org.elasticsearch.action.admin.indices.alias.get.GetAliasesRequest; import org.elasticsearch.action.admin.indices.create.CreateIndexRequest; import org.elasticsearch.action.admin.indices.create.CreateIndexResponse; import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest; import org.elasticsearch.action.admin.indices.get.GetIndexRequest; import org.elasticsearch.action.admin.indices.get.GetIndexResponse; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest; import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse; import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest; import org.elasticsearch.action.delete.DeleteRequest; import org.elasticsearch.action.delete.DeleteResponse; import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; import org.elasticsearch.action.index.IndexRequest; import org.elasticsearch.action.index.IndexResponse; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.action.support.master.AcknowledgedResponse; import org.elasticsearch.action.update.UpdateRequest; import org.elasticsearch.action.update.UpdateResponse; import org.elasticsearch.client.GetAliasesResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.RestHighLevelClient; import org.elasticsearch.common.settings.Settings; import org.elasticsearch.common.xcontent.XContentType; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; import java.io.IOException; import java.util.Collections; import java.util.Map; import java.util.Set; @Component public class ElasticSearchClient { private static final Logger log = LoggerFactory.getLogger(ElasticSearchClient.class); private static final ObjectMapper objectMapper = new ObjectMapper(); @Autowired private RestHighLevelClient client; /** * 查詢全部索引 * @return */ public Set<String> getAlias(){ try { GetAliasesRequest request = new GetAliasesRequest(); GetAliasesResponse response = client.indices().getAlias(request, RequestOptions.DEFAULT); return response.getAliases().keySet(); } catch (IOException e) { log.error("向es發起查詢全部索引信息請求失敗", e); } return Collections.emptySet(); } /** * 檢查索引是否存在 * @param indexName * @return */ public boolean existsIndex(String indexName){ try { // 創建請求 GetIndexRequest request = new GetIndexRequest().indices(indexName); // 執行請求,獲取響應 boolean response = client.indices().exists(request, RequestOptions.DEFAULT); return response; } catch (Exception e) { log.error("向es發起查詢索引是否存在請求失敗,請求參數:" + indexName, e); } return false; } /** * 查詢索引 * @param indexName * @return */ public String getIndex(String indexName){ try { // 創建請求 GetIndexRequest request = new GetIndexRequest().indices(indexName); // 執行請求,獲取響應 GetIndexResponse response = client.indices().get(request, RequestOptions.DEFAULT); return response.toString(); } catch (Exception e) { log.error("向es發起查詢索引請求失敗,請求參數:" + indexName, e); } return StringUtils.EMPTY; } /** * 創建索引 * @param indexName * @param mapping * @return */ public void createIndex(String indexName, Map<String, Object> mapping){ try { CreateIndexRequest request = new CreateIndexRequest(); //索引名稱 request.index(indexName); //索引配置 Settings settings = Settings.builder() .put("index.number_of_shards", 3) .put("index.number_of_replicas", 1) .put("index.max_inner_result_window", 5000) .build(); request.settings(settings); //索引結構 request.mapping("_doc",mapping); //執行請求,獲取響應 CreateIndexResponse response = client.indices().create(request, RequestOptions.DEFAULT); if(!response.isAcknowledged()){ throw new CommonException("向es發起創建索引請求失敗"); } log.info("向es發起創建索引請求成功,返回參數:{}", response.index()); } catch (Exception e) { log.error("向es發起創建索引請求失敗,請求參數:" + indexName, e); throw new CommonException("向es發起創建索引請求失敗"); } } /** * 刪除索引 * @param indexName * @return */ public void deleteIndex(String indexName){ try { DeleteIndexRequest request = new DeleteIndexRequest(indexName); AcknowledgedResponse response = client.indices().delete(request, RequestOptions.DEFAULT); if(!response.isAcknowledged()){ throw new CommonException("向es發起刪除索引請求失敗"); } log.info("向es發起刪除索引請求成功,請求參數:{}", indexName); } catch (Exception e) { log.error("向es發起刪除索引請求失敗,請求參數:" + indexName, e); throw new CommonException("向es發起刪除索引請求失敗"); } } /** * 查詢索引映射字段 * @param indexName * @return */ public String getMapping(String indexName){ try { GetMappingsRequest request = new GetMappingsRequest().indices(indexName).types("_doc"); GetMappingsResponse response = client.indices().getMapping(request, RequestOptions.DEFAULT); return response.toString(); } catch (Exception e) { log.error("向es發起查詢索引映射字段請求失敗,請求參數:" + indexName, e); } return StringUtils.EMPTY; } /** * 添加索引映射字段 * @param indexName * @return */ public void addMapping(String indexName, Map<String, Object> mapping){ try { PutMappingRequest request = new PutMappingRequest(); request.indices(indexName); request.type("_doc"); //添加字段 request.source(mapping); AcknowledgedResponse response = client.indices().putMapping(request, RequestOptions.DEFAULT); if(!response.isAcknowledged()){ throw new CommonException("向es發起添加索引映射字段請求失敗"); } log.info("向es發起添加索引映射字段請求成功,請求參數:{}", toJson(request)); } catch (Exception e) { log.error("向es發起添加索引映射字段請求失敗,請求參數:" + indexName, e); throw new CommonException("向es發起添加索引映射字段請求失敗"); } } /** * 向索引中添加文檔 * @param indexName * @param id * @param obj */ public void addDocument(String indexName, String id, Object obj){ try { //向索引中添加文檔 IndexRequest request = new IndexRequest(); // 外層參數 request.id(id); request.index(indexName); request.type("_doc"); // 存入對象 request.source(toJson(obj), XContentType.JSON); // 發送請求 IndexResponse response = client.index(request, RequestOptions.DEFAULT); if(response.status().getStatus() >= 400){ log.warn("向es發起添加文檔數據請求失敗,請求參數:{},返回參數:{}", request.toString(), response.toString()); throw new CommonException("向es發起添加文檔數據請求失敗"); } } catch (Exception e) { log.error("向es發起添加文檔數據請求失敗,請求參數:" + indexName, e); throw new CommonException("向es發起添加文檔數據請求失敗"); } } /** * 修改索引中的文檔數據 * @param indexName * @param id * @param obj */ public void updateDocument(String indexName, String id, Map<String,Object> obj){ try { //修改索引中的文檔數據 UpdateRequest request = new UpdateRequest(); // 外層參數 request.id(id); request.index(indexName); request.type("_doc"); // 存入對象 request.doc(obj); request.doc(toJson(obj), XContentType.JSON); // 發送請求 UpdateResponse response = client.update(request, RequestOptions.DEFAULT); if(response.status().getStatus() >= 400){ log.warn("向es發起修改文檔數據請求失敗,請求參數:{},返回參數:{}", request.toString(), response.toString()); throw new CommonException("向es發起修改文檔數據請求失敗"); } } catch (Exception e) { log.error("向es發起修改文檔數據請求失敗,請求參數:" + indexName, e); throw new CommonException("向es發起修改文檔數據請求失敗"); } } /** * 刪除索引中的文檔數據 * @param indexName * @param id */ public void deleteDocument(String indexName, String id){ try { //刪除索引中的文檔數據 DeleteRequest request = new DeleteRequest(); // 外層參數 request.id(id); request.index(indexName); request.type("_doc"); // 發送請求 DeleteResponse response = client.delete(request, RequestOptions.DEFAULT); if(response.status().getStatus() >= 400){ log.warn("向es發起刪除文檔數據請求失敗,請求參數:{},返回參數:{}", request.toString(), response.toString()); throw new CommonException("向es發起刪除文檔數據請求失敗"); } } catch (Exception e) { log.error("向es發起刪除文檔數據請求失敗,請求參數:" + indexName, e); throw new CommonException("向es發起刪除文檔數據請求失敗"); } } /** * 查詢索引中的文檔數據 * @param indexName * @param id */ public String getDocumentById(String indexName, String id){ try { GetRequest request = new GetRequest(); // 外層參數 request.id(id); request.index(indexName); request.type("_doc"); // 發送請求 GetResponse response = client.get(request, RequestOptions.DEFAULT); response.getSourceAsString(); } catch (Exception e) { log.error("向es發起查詢文檔數據請求失敗,請求參數:" + indexName, e); } return StringUtils.EMPTY; } /** * 索引高級查詢 * @param indexName * @param source * @return */ public SearchResponse searchDocument(String indexName, SearchSourceBuilder source){ //搜索 SearchRequest searchRequest = new SearchRequest(); searchRequest.indices(indexName); searchRequest.source(source); try { // 執行請求 SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT); return response; } catch (Exception e) { log.warn("向es發起查詢文檔數據請求失敗,請求參數:" + searchRequest.toString(), e); } return null; } /** * 將對象格式化成json,並保持原字段類型輸出 * @param object * @return */ private String toJson(Object object) { try { return objectMapper.writeValueAsString(object); } catch (Exception e) { throw new CommonException(e); } } }
2.3初始化索引結構
在使用 es 對訂單進行查詢搜索時,我們需要先定義好對應的訂單索引結構,內容如下:
@ActiveProfiles("dev") @RunWith(SpringRunner.class) @SpringBootTest public class OrderIndexServiceJunit { @Autowired private ElasticSearchClient elasticSearchClient; /** * 初始化索引結構 * * @return */ @Test public void initIndex(){ String indexName = "orderIndex-2022-07"; // 創建請求 boolean existsIndex = elasticSearchClient.existsIndex(indexName); if (!existsIndex) { Map<String, Object> properties = buildMapping(); elasticSearchClient.createIndex(indexName, properties); } } /** * 構建索引結構 * * @return */ private Map<String, Object> buildMapping() { Map<String, Object> properties = new HashMap(); //訂單id 唯一鍵ID properties.put("orderId", ImmutableBiMap.of("type", "keyword")); //訂單號 properties.put("orderNo", ImmutableBiMap.of("type", "keyword")); //客戶姓名 properties.put("orderUserName", ImmutableBiMap.of("type", "text")); //訂單項 Map<String, Object> orderItems = new HashMap(); //訂單項ID orderItems.put("orderItemId", ImmutableBiMap.of("type", "keyword")); //產品名稱 orderItems.put("productName", ImmutableBiMap.of("type", "text")); //品牌名稱 orderItems.put("brandName", ImmutableBiMap.of("type", "text")); //銷售金額,單位分*100 orderItems.put("sellPrice", ImmutableBiMap.of("type", "integer")); properties.put("orderItems", ImmutableBiMap.of("type", "nested", "properties", orderItems)); //文檔結構映射 Map<String, Object> mapping = new HashMap(); mapping.put("properties", properties); return mapping; } }
2.4向 es 中同步文檔數據
索引結構創建好之後,我們需要將支持 es 搜索的訂單數據同步進去。
將指定的訂單 ID 從數據庫查詢出來,並封裝成 es 訂單數據結構,保存到 es 中!
@ActiveProfiles("dev") @RunWith(SpringRunner.class) @SpringBootTest public class OrderIndexServiceJunit { @Autowired private ElasticSearchClient elasticSearchClient; /** * 保存訂單到ES中 * @param request */ @Test public void saveDocument(){ String indexName = "orderIndex-2022-07"; //從數據庫查詢最新訂單數據,並封裝成對應的es訂單結構 String orderId = "202202020202"; OrderIndexDocDTO indexDocDTO = buildOrderIndexDocDTO(orderId); //保存數據到ES中 elasticSearchClient.addDocument(indexName, indexDocDTO.getOrderId(), indexDocDTO); } }
2.5內嵌對象查詢
內嵌對象查詢分兩種形式,比如,第一種通過商品、品牌、價格等條件,分頁查詢訂單數據;第二種是通過訂單ID、商品、品牌、價格等,分頁查詢訂單項數據。具體的實踐,請看下文。
通過商品、品牌、價格等條件,分頁查詢訂單數據;
@ActiveProfiles("dev") @RunWith(SpringRunner.class) @SpringBootTest public class OrderIndexServiceJunit { @Autowired private ElasticSearchClient elasticSearchClient; /** * 通過商品、品牌、價格等條件,分頁查詢訂單數據 * @param request */ @Test public void search1(){ //查詢索引,支持通配符 String indexName = "orderIndex-*"; String orderUserName = "張三"; String productName = "薯條"; // 條件搜索 SearchSourceBuilder builder = new SearchSourceBuilder(); //組合搜索 BoolQueryBuilder mainBoolQuery = new BoolQueryBuilder(); mainBoolQuery.must(QueryBuilders.matchQuery("orderUserName", orderUserName)); //訂單項相關信息搜索 BoolQueryBuilder nestedBoolQuery = new BoolQueryBuilder(); nestedBoolQuery.must(QueryBuilders.matchQuery("orderItems.productName", productName)); //內嵌對象搜索,需要指定path NestedQueryBuilder nestedQueryBuilder = QueryBuilders.nestedQuery("orderItems",nestedBoolQuery, ScoreMode.None); //子表查詢 mainBoolQuery.must(nestedQueryBuilder); //封裝查詢參數 builder.query(mainBoolQuery); //返回參數 builder.fetchSource(new String[]{}, new String[]{}); //結果集合分頁,從第一頁開始,返回最多四條數據 builder.from(0).size(4); //排序 builder.sort("orderId", SortOrder.DESC); log.info("dsl:{}", builder.toString()); // 執行請求 SearchResponse response = elasticSearchClient.searchDocument(indexName, builder); // 當前返回的總行數 long count = response.getHits().getTotalHits(); // 返回的具體行數 SearchHit[] searchHits = response.getHits().getHits(); log.info("response:{}", response.toString()); } }
通過訂單ID、商品、品牌、價格等,分頁查詢訂單項數據;
@ActiveProfiles("dev") @RunWith(SpringRunner.class) @SpringBootTest public class OrderIndexServiceJunit { @Autowired private ElasticSearchClient elasticSearchClient; /** * 通過訂單ID、商品、品牌、價格等,分頁查詢訂單項數據 * @param request */ @Test public void search2(){ //查詢索引,支持通配符 String indexName = "orderIndex-*"; String orderId = "202202020202"; String productName = "薯條"; // 條件搜索 SearchSourceBuilder builder = new SearchSourceBuilder(); //組合搜索 BoolQueryBuilder mainBoolQuery = new BoolQueryBuilder(); mainBoolQuery.must(QueryBuilders.termQuery("_id", orderId)); //訂單項相關信息搜索 BoolQueryBuilder nestedBoolQuery = new BoolQueryBuilder(); nestedBoolQuery.must(QueryBuilders.matchQuery("orderItems.productName", productName)); //內嵌對象搜索,需要指定path NestedQueryBuilder nestedQueryBuilder = QueryBuilders.nestedQuery("orderItems",nestedBoolQuery, ScoreMode.None); //內嵌對象分頁查詢 InnerHitBuilder innerHitBuilder = new InnerHitBuilder(); //結果集合分頁,從第一頁開始,返回最多四條數據 innerHitBuilder.setFrom(0).setSize(4); //隻返回訂單項id innerHitBuilder.setFetchSourceContext(new FetchSourceContext(true, new String[]{"orderItems.orderItemId"}, new String[]{})); innerHitBuilder.addSort(SortBuilders.fieldSort("orderItems.orderItemId").order(SortOrder.DESC)); nestedQueryBuilder.innerHit(innerHitBuilder); //子表查詢 mainBoolQuery.must(nestedQueryBuilder); //封裝查詢參數 builder.query(mainBoolQuery); //返回參數 builder.fetchSource(new String[]{}, new String[]{}); //結果集合分頁,從第一頁開始,返回最多四條數據 builder.from(0).size(4); //排序 builder.sort("orderId", SortOrder.DESC); log.info("dsl:{}", builder.toString()); // 執行請求 SearchResponse response = elasticSearchClient.searchDocument(indexName, builder); // 當前返回的訂單主表總行數 long count = response.getHits().getTotalHits(); // 返回的訂單主表數據 SearchHit[] searchHits = response.getHits().getHits(); // 返回查詢的的訂單項分頁數據 Map<String, SearchHits> = searchHit[0].getInnerHits(); log.info("response:{}", response.toString()); } }
三、小結
本文主要以通過商品名稱查詢訂單數據為案例,介紹利用 SpringBoot 整合 es 實現數據的高效搜索,內容如果難免有些遺漏,歡迎網友指出!
推薦閱讀:
- .Net Api 中使用Elasticsearch存儲文檔的方法
- Java操作Elasticsearch rest-high-level-client 的基本使用
- SpringBoot+Elasticsearch實現數據搜索的方法詳解
- SpringBoot 整合 Elasticsearch 實現海量級數據搜索功能
- 解決RestTemplate 請求url中包含百分號 會被轉義成25的問題