大綱

1.商品B端搜索系統的運行流程 + 緩存和索引設計

2.商品B端搜索系統監聽數據變更與寫入ES索引

3.商品B端搜索系統的歷史搜索詞的實現

4.商品B端搜索系統的搜索詞補全的實現

5.商品B端搜索系統的搜索接口實現

6.索引重建

1.商品B端搜索系統的運行流程 + 緩存和索引設計

(1)運行流程

(2)Redis緩存設計

(3)索引設計

(1)運行流程

(2)Redis緩存設計

使用Redis緩存用戶的搜索詞記錄，比如保存最近的10個搜索詞記錄，使?的數據結構：list。

key的格式：history_search_words:{userId}
value的格式：["?奶", "雞蛋", "??鳳?", "正?蒸餃"]

(3)索引設計

一.商品索引

二.索引字段說明

三.數據示例

四.搜索補全索引

一.商品索引

PUT /sku_info_index
{"settings": {"number_of_shards": 3,"number_of_replicas": 1},"mappings": {"properties": {"skuName": {"type": "text","analyzer": "ik_max_word","search_analyzer": "ik_smart"},"basePrice": {"type": "integer"},"vipPrice": {"type": "integer"},"brandId": {"type": "keyword"},"brandName": {"type": "keyword"},"saleCount": {"type": "integer"},"createTime": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"},"updateTime": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"}}}
}

二.索引字段說明

三.數據示例

{"_index": "sku_info_index","_type": "_doc","_id": "8000177337","_score": 1.0,"_source": {"skuName": "Apple iPhone 13 Pro Max 256GB 蒼嶺綠? ?持移動聯通電信5G 雙卡雙待?機","brandName": "蘋果","createTime": "2022-03-12 08:24:57","brandId": 4,"vipPrice": 9799,"updateTime": "2022-03-12 08:24:57","basePrice": 9999}
}

四.搜索補全索引

put /completion_word_index
{"settings": {"number_of_shards": 3,"number_of_replicas": 1,"analysis": {"analyzer": {"ik_and_pinyin_analyzer": {"type": "custom","tokenizer": "ik_smart","filter": "my_pinyin"}},"filter": {"my_pinyin": {"type": "pinyin","keep_first_letter": true,"keep_full_pinyin": true,"keep_original": true,"remove_duplicated_term": true}}}},"mappings": {"properties": {"completion_word": {"type": "completion","analyzer": "ik_and_pinyin_analyzer"}}}
}

2.商品B端搜索系統監聽數據變更與寫入ES索引

(1)消息處理系統添加數據監聽配置

(2)商品B端搜索系統下的數據變更消息消費者

(3)sku表變更消息處理器

(4)item表變更消息處理器

(1)消息處理系統添加數據監聽配置

一.data_change_listen_config表

INSERT INTO data_change_listen_config (id, table_name, key_column, filter_flag, del_flag, create_user, create_time, update_user, update_time) 
VALUES (1, 'sku_info', 'sku_id', 1, 1, 0, '2022-02-25 13:42:28', 0, '2022-02-25 13:42:28');
INSERT INTO data_change_listen_config (id, table_name, key_column, filter_flag, del_flag, create_user, create_time, update_user, update_time) 
VALUES (2, 'item_info', 'item_id', 1, 1, 0, '2022-02-25 13:42:28', 0, '2022-02-25 13:42:28');

二.data_change_column_config表

INSERT INTO data_change_column_config (id, listen_id, listen_column, del_flag, create_user, create_time, update_user, update_time)
VALUES (1, 1, 'sku_name', 1, 0, '2022-02-25 13:43:28', 0, '2022-02-25 13:43:28');
INSERT INTO data_change_column_config (id, listen_id, listen_column, del_flag, create_user, create_time, update_user, update_time)
VALUES (2, 1, 'channel', 1, 0, '2022-02-25 13:43:28', 0, '2022-02-25 13:43:28');
INSERT INTO data_change_column_config (id, listen_id, listen_column, del_flag, create_user, create_time, update_user, update_time)
VALUES (3, 1, 'features', 1, 0, '2022-02-25 13:43:28', 0, '2022-02-25 13:43:28');
INSERT INTO data_change_column_config (id, listen_id, listen_column, del_flag, create_user, create_time, update_user, update_time)
VALUES (4, 1, 'vip_price', 1, 0, '2022-02-25 13:43:28', 0, '2022-02-25 13:43:28');
INSERT INTO data_change_column_config (id, listen_id, listen_column, del_flag, create_user, create_time, update_user, update_time)
VALUES (5, 1, 'base_price', 1, 0, '2022-02-25 13:43:28', 0, '2022-02-25 13:43:28');
INSERT INTO data_change_column_config (id, listen_id, listen_column, del_flag, create_user, create_time, update_user, update_time)
VALUES (6, 2, 'brand_id', 1, 0, '2022-02-25 13:43:28', 0, '2022-02-25 13:43:28');

三.data_change_message_config表

INSERT INTO data_change_message_config (id, listen_id, notify_column, message_topic, delay_level, message_type, del_flag, create_user, create_time, update_user, update_time)
VALUES (1, 1, 'id,sku_id', 'product_update_topic', 3, 1, 1, 0, '2022-02-25 13:45:24', 0, '2022-02-25 13:45:24');
INSERT INTO data_change_message_config (id, listen_id, notify_column, message_topic, delay_level, message_type, del_flag, create_user, create_time, update_user, update_time)
VALUES (3, 2, 'id,item_id', 'product_update_topic', 3, 1, 1, 0, '2022-02-25 13:45:24', 0, '2022-02-25 13:45:24');

(2)商品B端搜索系統下的數據變更消息消費者

@Configuration
public class ConsumerBeanConfig {//配置內容對象@Autowiredprivate RocketMQProperties rocketMQProperties;//監聽商品修改的MQ消息@Bean("productUpdateTopic")public DefaultMQPushConsumer productUpdateTopic(ProductUpdateListener productUpdateListener) throws MQClientException {DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(RocketMqConstant.PRODUCT_UPDATE_CONSUMER_GROUP);consumer.setNamesrvAddr(rocketMQProperties.getNameServer());consumer.subscribe(RocketMqConstant.PRODUCT_UPDATE_TOPIC, "*");consumer.registerMessageListener(productUpdateListener);consumer.start();return consumer;}
}//搜索模塊在商品變更的時候更新商品索引
@Component
public class ProductUpdateListener implements MessageListenerConcurrently {@Autowiredprivate MessageHandlerManager messageHandlerManager;@Overridepublic ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {try {for (MessageExt messageExt : list) {String msg = new String(messageExt.getBody());log.info("執行商品索引數據更新邏輯，消息內容：{}", msg);TableDataChangeDTO tableDataChangeDTO = JsonUtil.json2Object(msg, TableDataChangeDTO.class);//處理消息messageHandlerManager.handleMessage(tableDataChangeDTO);}} catch (Exception e){log.error("consume error, 商品索引數據更新失敗", e);//本次消費失敗，下次重新消費return ConsumeConcurrentlyStatus.RECONSUME_LATER;}return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;}
}@Component
public class MessageHandlerManager {//繼承了MessageHandler的ItemInfoTableMessageHandler和SkuInfoTableMessageHandler都會被注入到這里@Autowiredprivate List<MessageHandler> messageHandlers;public void handleMessage(TableDataChangeDTO tableDataChangeDTO) throws IOException {MessageHandler messageHandlerToUse = messageHandlers.stream().filter(e -> StringUtils.equals(e.tableName(), tableDataChangeDTO.getTableName())).findFirst().orElse(null);if (messageHandlerToUse == null) {return;}messageHandlerToUse.handleMessage(tableDataChangeDTO);}
}

(3)sku表變更消息處理器

@Component
public class SkuInfoTableMessageHandler implements MessageHandler {@Autowiredprivate ProductSearchRepository productSearchRepository;@Overridepublic String tableName() {return "sku_info";}@Overridepublic void handleMessage(TableDataChangeDTO tableDataChangeDTO) throws IOException {String skuId = String.valueOf(tableDataChangeDTO.getKeyId());//到數據庫查詢索引相關的信息ProductSearchDO productSearchDO = productSearchRepository.queryProductSearchInfo(skuId);//保存索引數據到ESproductSearchRepository.saveProductSearchInfos(Collections.singletonList(productSearchDO));}
}@Repository
public class ProductSearchRepository {private static final String SKU_INFO_INDEX = "sku_info_index";@Autowiredprivate RestHighLevelClient restHighLevelClient;@Autowiredprivate SkuInfoMapper skuInfoMapper;//根據skuId查詢和商品索引相關的信息public ProductSearchDO queryProductSearchInfo(String skuId) {return skuInfoMapper.queryProductSearchInfo(skuId);}//批量保存商品索引數據public void saveProductSearchInfos(List<ProductSearchDO> productSearchDOS) throws IOException {BulkRequest bulkRequest = new BulkRequest();for (ProductSearchDO productSearchDO : productSearchDOS) {Map<String, Object> jsonMap = new HashMap<>();jsonMap.put("skuName", productSearchDO.getSkuName());jsonMap.put("basePrice", productSearchDO.getBasePrice());jsonMap.put("vipPrice", productSearchDO.getVipPrice());jsonMap.put("brandId", productSearchDO.getBrandId());jsonMap.put("brandName", productSearchDO.getBrandName());jsonMap.put("createTime", new Date());jsonMap.put("updateTime", new Date());IndexRequest indexRequest = new IndexRequest(SKU_INFO_INDEX).id(productSearchDO.getSkuId()).source(jsonMap);bulkRequest.add(indexRequest);}restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);}...
}

(4)item表變更消息處理器

@Component
public class ItemInfoTableMessageHandler implements MessageHandler {@Autowiredprivate ProductSearchRepository productSearchRepository;@Overridepublic String tableName() {return "item_info";}@Overridepublic void handleMessage(TableDataChangeDTO tableDataChangeDTO) throws IOException {String itemId = String.valueOf(tableDataChangeDTO.getKeyId());List<ProductSearchDO> productSearchDOS = productSearchRepository.queryProductSearchInfos(itemId);productSearchRepository.saveProductSearchInfos(productSearchDOS);}
}@Repository
public class ProductSearchRepository {private static final String SKU_INFO_INDEX = "sku_info_index";@Autowiredprivate RestHighLevelClient restHighLevelClient;@Autowiredprivate SkuInfoMapper skuInfoMapper;//根據itemId查詢和商品索引相關的信息public List<ProductSearchDO> queryProductSearchInfos(String itemId) {return skuInfoMapper.queryProductSearchInfos(itemId);}//批量保存商品索引數據public void saveProductSearchInfos(List<ProductSearchDO> productSearchDOS) throws IOException {BulkRequest bulkRequest = new BulkRequest();for (ProductSearchDO productSearchDO : productSearchDOS) {Map<String, Object> jsonMap = new HashMap<>();jsonMap.put("skuName", productSearchDO.getSkuName());jsonMap.put("basePrice", productSearchDO.getBasePrice());jsonMap.put("vipPrice", productSearchDO.getVipPrice());jsonMap.put("brandId", productSearchDO.getBrandId());jsonMap.put("brandName", productSearchDO.getBrandName());jsonMap.put("createTime", new Date());jsonMap.put("updateTime", new Date());IndexRequest indexRequest = new IndexRequest(SKU_INFO_INDEX).id(productSearchDO.getSkuId()).source(jsonMap);bulkRequest.add(indexRequest);}restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);}...
}

3.商品B端搜索系統的歷史搜索詞的實現

(1)商品B端保存歷史搜索詞的接?

(2)商品B端查詢歷史搜索詞的接?

(1)商品B端保存歷史搜索詞的接?

使用場景：商家輸入搜索詞搜索商品的時候
接口說明：把商家搜索過的詞保存到Redis的List數據結構中

//商品搜索服務
@DubboService(version = "1.0.0", interfaceClass = ProductSearchApi.class, retries = 0)
public class ProductSearchApiImpl implements ProductSearchApi {@Resourceprivate RedisCache redisCache;@Resourceprivate ProductSearchRepository productSearchRepository;//保存歷史搜索詞接口@Overridepublic JsonResult<HistorySearchWordResultDTO> saveHistorySearchWord(HistorySearchWordRequest request) {//在隊列頭部添加新的歷史搜索詞redisCache.lpush(HistorySearchWordConstants.getKey(request.getUserId()), request.getHistorySearchWord());//修改隊列只保存固定數量的搜索詞redisCache.ltrim(HistorySearchWordConstants.getKey(request.getUserId()), 0, HistorySearchWordConstants.HISTORY_WORD_COUNT_PER_USER - 1);return JsonResult.buildSuccess(new HistorySearchWordResultDTO(true));}...
}//保存用戶歷史搜索詞請求
@Data
public class HistorySearchWordRequest implements Serializable {//用戶idprivate Long userId;//新的歷史搜索詞private String historySearchWord;
}

(2)商品B端查詢歷史搜索詞的接?

使用場景：展示商家的搜索歷史記錄的時候
接口說明：從Redis列表中查詢商家的歷史搜索詞

//商品搜索服務
@DubboService(version = "1.0.0", interfaceClass = ProductSearchApi.class, retries = 0)
public class ProductSearchApiImpl implements ProductSearchApi {@Resourceprivate RedisCache redisCache;@Resourceprivate ProductSearchRepository productSearchRepository;//查詢歷史搜索詞接口@Overridepublic JsonResult<HistorySearchWordDTO> listHistorySearchWords(HistorySearchWordQuery request) {List<String> result = redisCache.lrange(HistorySearchWordConstants.getKey(request.getUserId()), 0, HistorySearchWordConstants.HISTORY_WORD_COUNT_PER_USER - 1);return JsonResult.buildSuccess(new HistorySearchWordDTO(result));}...
}//查詢商家歷史搜索詞請求
@Data
public class HistorySearchWordQuery implements Serializable {//用戶idprivate Long userId;
}

4.商品B端搜索系統的搜索詞補全的實現

(1)商品B端搜索系統的添加搜索補全詞的接?

(2)商品B端搜索系統查詢搜索補全詞的接口

(3)商品B端搜索詞補全的接口

(1)商品B端搜索系統的添加搜索補全詞的接?

使用場景：運營人員添加搜索補全詞的時候
接口說明：把搜索補全詞保存到ES的搜索補全詞索引中

//搜索詞
@DubboService(version = "1.0.0", interfaceClass = CompletionSearchWordApi.class, retries = 0)
public class CompletionSearchWordApiImpl implements CompletionSearchWordApi {@Autowiredprivate CompletionSearchWordService completionSearchWordService;//保存搜索補全詞接口@Overridepublic JsonResult<CompletionSearchWordResultDTO> saveCompletionSearchWord(CompletionSearchWordRequest request) {try {CompletionSearchWordResultDTO resultDTO = completionSearchWordService.saveCompletionSearchWord(request);return JsonResult.buildSuccess(resultDTO);} catch (ProductBizException e) {log.error("biz error: request={}", JSON.toJSONString(request), e);return JsonResult.buildError(e.getErrorCode(), e.getErrorMsg());} catch (Exception e) {log.error("system error: request={}", JSON.toJSONString(request), e);return JsonResult.buildError(e.getMessage());}}...
}@Service
public class CompletionSearchWordServiceImpl implements CompletionSearchWordService {@Autowiredprivate CompletionSearchWordRepository completionSearchWordRepository;//保存搜索補全詞@Overridepublic CompletionSearchWordResultDTO saveCompletionSearchWord(CompletionSearchWordRequest request) throws IOException {return completionSearchWordRepository.saveCompletionSearchWord(request);}...
}//運營添加搜索補全詞請求
@Data
public class CompletionSearchWordRequest implements Serializable {//索引名稱private String indexName;//字段名稱private String fieldName;//要添加的補全詞private List<String> completionSearchWords;
}@Repository
public class CompletionSearchWordRepository {@Autowiredprivate RestHighLevelClient restHighLevelClient;//保存搜索補全詞public CompletionSearchWordResultDTO saveCompletionSearchWord(CompletionSearchWordRequest request) throws IOException {BulkRequest bulkRequest = new BulkRequest(request.getIndexName());List<String> completionSearchWords = request.getCompletionSearchWords();for (String completionSearchWord : completionSearchWords) {bulkRequest.add(new IndexRequest().source(XContentType.JSON, request.getFieldName(), completionSearchWord));}restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);return new CompletionSearchWordResultDTO(true);}...
}

(2)商品B端搜索系統查詢搜索補全詞的接口

使用場景：后臺展示搜索補全詞列表的時候
接口說明：從ES的搜索補全詞索引中分頁查詢數據

//搜索詞
@DubboService(version = "1.0.0", interfaceClass = CompletionSearchWordApi.class, retries = 0)
public class CompletionSearchWordApiImpl implements CompletionSearchWordApi {@Autowiredprivate CompletionSearchWordService completionSearchWordService;//查詢補全詞接口@Overridepublic JsonResult<PageResult<CompletionSearchWordDTO>> listCompletionSearchWordPage(QueryCompletionSearchWordPageRequest request) {try {PageResult<CompletionSearchWordDTO> resultDTO = completionSearchWordService.listCompletionSearchWordPage(request);return JsonResult.buildSuccess(resultDTO);} catch (ProductBizException e) {log.error("biz error: request={}", JSON.toJSONString(request), e);return JsonResult.buildError(e.getErrorCode(), e.getErrorMsg());} catch (Exception e) {log.error("system error: request={}", JSON.toJSONString(request), e);return JsonResult.buildError(e.getMessage());}}...
}@Service
public class CompletionSearchWordServiceImpl implements CompletionSearchWordService {@Autowiredprivate CompletionSearchWordRepository completionSearchWordRepository;//查詢搜索補全詞@Overridepublic PageResult<CompletionSearchWordDTO> listCompletionSearchWordPage(QueryCompletionSearchWordPageRequest request) throws IOException {return completionSearchWordRepository.listCompletionSearchWordPage(request);}...
}//后臺查詢搜索詞列表請求
@Data
public class QueryCompletionSearchWordPageRequest extends PageRequest {//索引名稱private String indexName;//字段名稱private String fieldName;//補全詞private String completionSearchWord;
}@Repository
public class CompletionSearchWordRepository {@Autowiredprivate RestHighLevelClient restHighLevelClient;...//查詢搜索補全詞public PageResult<CompletionSearchWordDTO> listCompletionSearchWordPage(QueryCompletionSearchWordPageRequest request) throws IOException {SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();if (StringUtils.isNotBlank(request.getCompletionSearchWord())) {searchSourceBuilder.query(QueryBuilders.matchQuery(request.getFieldName(), request.getCompletionSearchWord()));}int from = (request.getPageNum() - 1) * request.getPageSize();searchSourceBuilder.from(from);searchSourceBuilder.size(request.getPageSize());SearchRequest searchRequest = new SearchRequest(request.getIndexName());searchRequest.source(searchSourceBuilder);SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);PageResult<CompletionSearchWordDTO> pageResult = new PageResult<>();List<CompletionSearchWordDTO> pageContent = new ArrayList<>();SearchHit[] hits = searchResponse.getHits().getHits();for (SearchHit hit : hits) {pageContent.add(new CompletionSearchWordDTO(((String) hit.getSourceAsMap().get(request.getFieldName()))));}pageResult.setContent(pageContent);pageResult.setTotalElements(searchResponse.getHits().getTotalHits().value);pageResult.setSize(request.getPageSize());pageResult.setNumber(request.getPageNum());return pageResult;}...
}

(3)商品B端搜索詞補全的接口

使用場景：商家在搜索框輸入搜索詞的時候
接口說明：根據輸入的搜索詞從ES的搜索補全詞索引中查詢對應的詞

//商品搜索
@DubboService(version = "1.0.0", interfaceClass = ProductSearchApi.class, retries = 0)
public class ProductSearchApiImpl implements ProductSearchApi {@Resourceprivate RedisCache redisCache;@Resourceprivate ProductSearchRepository productSearchRepository;...//搜索詞補全接口@Overridepublic JsonResult<CompletionSearchWordsDTO> listCompletionSearchWords(CompletionSearchWordQuery request) {try {CompletionSearchWordsDTO result = productSearchRepository.listCompletionSearchWords(request);return JsonResult.buildSuccess(result);} catch (Exception e) {e.printStackTrace();return JsonResult.buildError(e.getMessage());}}...
}//補全用戶搜索詞請求
@Data
public class CompletionSearchWordQuery {//索引名稱private String indexName;//字段名稱private String fieldName;//需要補全的詞(用戶輸入的內容)private String text;//返回多少個補全后的詞private int count;
}//商品搜索
@Repository
public class ProductSearchRepository {private static final String MY_SUGGEST = "my_suggest";@Resourceprivate RestHighLevelClient restHighLevelClient;//搜索詞補全public CompletionSearchWordsDTO listCompletionSearchWords(CompletionSearchWordQuery request) throws IOException {//1.構建CompletionSuggestion條件CompletionSuggestionBuilder completionSuggestionBuilder = SuggestBuilders.completionSuggestion(request.getFieldName());completionSuggestionBuilder.prefix(request.getText());completionSuggestionBuilder.skipDuplicates(true);completionSuggestionBuilder.size(request.getCount());SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();searchSourceBuilder.sort(new ScoreSortBuilder().order(SortOrder.DESC));searchSourceBuilder.suggest(new SuggestBuilder().addSuggestion(MY_SUGGEST, completionSuggestionBuilder));//2.封裝搜索請求SearchRequest searchRequest = new SearchRequest();searchRequest.indices(request.getIndexName());searchRequest.source(searchSourceBuilder);//3.查詢elasticsearchSearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);//4.獲取響應中的補全的詞的列表CompletionSuggestion completionSuggestion = searchResponse.getSuggest().getSuggestion(MY_SUGGEST);List<CompletionSuggestion.Entry.Option> options = completionSuggestion.getEntries().get(0).getOptions();List<String> result = new ArrayList<>();for (CompletionSuggestion.Entry.Option option : options) {result.add(option.getText().string());}return new CompletionSearchWordsDTO(result);}...
}

5.商品B端搜索系統的搜索接口實現

(1)商品B端的搜索查詢接口

(2)商品B端的結構化查詢接口

(1)商品B端的搜索查詢接口

使用場景：商家搜索商品的時候
接口說明：根據輸入的搜索詞從商品索引中查詢skuId列表

//商品搜索
@DubboService(version = "1.0.0", interfaceClass = ProductSearchApi.class, retries = 0)
public class ProductSearchApiImpl implements ProductSearchApi {...//商品搜索查詢接口@Overridepublic JsonResult<PorductSearchDTO> searchProducts(ProductSearchQuery request) {try {PorductSearchDTO result = productSearchRepository.searchProducts(request);return JsonResult.buildSuccess(result);} catch (Exception e) {e.printStackTrace();return JsonResult.buildError(e.getMessage());}}...
}//商品搜索請求
@Data
public class ProductSearchQuery extends PageQuery {//索引名字private String indexName;//查詢參數private Map<String, String> queryTexts;//高亮字段private String highLightField;
}//商品搜索
@Repository
public class ProductSearchRepository {private static final String MY_SUGGEST = "my_suggest";@Resourceprivate RestHighLevelClient restHighLevelClient;...//商品搜索查詢接口public PorductSearchDTO searchProducts(ProductSearchQuery request) throws IOException {SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();searchSourceBuilder.trackTotalHits(true);//1.構建match條件request.getQueryTexts().forEach((field, text) -> {searchSourceBuilder.query(QueryBuilders.matchQuery(field, text));});//2.設置搜索高亮配置HighlightBuilder highlightBuilder = new HighlightBuilder();highlightBuilder.field(request.getHighLightField());highlightBuilder.preTags("<span stype=color:red>");highlightBuilder.postTags("</span>");highlightBuilder.numOfFragments(0);searchSourceBuilder.highlighter(highlightBuilder);//3.設置搜索分頁參數int from = (request.getPageNum() - 1) * request.getPageSize();searchSourceBuilder.from(from);searchSourceBuilder.size(request.getPageSize());//4.封裝搜索請求SearchRequest searchRequest = new SearchRequest(request.getIndexName());searchRequest.source(searchSourceBuilder);//5.查詢elasticsearchSearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);//6.對結果進行高亮處理SearchHits hits = searchResponse.getHits();for (SearchHit hit : hits) {HighlightField highlightField = hit.getHighlightFields().get(request.getHighLightField());Map<String, Object> sourceAsMap = hit.getSourceAsMap();Text[] fragments = highlightField.fragments();StringBuilder builder = new StringBuilder();for (Text fragment : fragments) {builder.append(fragment.string());}sourceAsMap.put(request.getHighLightField(), builder.toString());}//7.封裝返回結果return buildPorductSearchDTO(hits, request.getPageNum(), request.getPageSize());}...
}

(2)商品B端的結構化查詢接口

使用場景：商家對搜索結果過濾和排序的時候
接口說明：根據用戶輸入的過濾和排序條件從商品索引中查詢skuId列表

//商品搜索
@DubboService(version = "1.0.0", interfaceClass = ProductSearchApi.class, retries = 0)
public class ProductSearchApiImpl implements ProductSearchApi {...//商品結構化查詢接口@Overridepublic JsonResult<PorductSearchDTO> structuredSearchProducts(ProductStructuredQuery request) {try {PorductSearchDTO result = productSearchRepository.structuredSearchProducts(request);return JsonResult.buildSuccess(result);} catch (Exception e) {e.printStackTrace();return JsonResult.buildError(e.getMessage());}}...
}//商品結構化查詢請求
@Data
public class ProductStructuredQuery extends PageQuery {//索引名字private String indexName;//Query DSLprivate Map<String, Object> queryDsl;
}//商品搜索
@Repository
public class ProductSearchRepository {private static final String MY_SUGGEST = "my_suggest";@Resourceprivate RestHighLevelClient restHighLevelClient;...//商品結構化查詢public PorductSearchDTO structuredSearchProducts(ProductStructuredQuery request) throws IOException {SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();searchSourceBuilder.trackTotalHits(true);//1.解析queryDSLString queryDsl = JSON.toJSONString(request.getQueryDsl());SearchModule searchModule = new SearchModule(Settings.EMPTY, false, Collections.emptyList());NamedXContentRegistry namedXContentRegistry = new NamedXContentRegistry(searchModule.getNamedXContents());XContent xContent = XContentFactory.xContent(XContentType.JSON);XContentParser xContentParser = xContent.createParser(namedXContentRegistry, LoggingDeprecationHandler.INSTANCE, queryDsl);searchSourceBuilder.parseXContent(xContentParser);//2.設置搜索分頁參數int from = (request.getPageNum() - 1) * request.getPageSize();searchSourceBuilder.from(from);searchSourceBuilder.size(request.getPageSize());//3.封裝搜索請求SearchRequest searchRequest = new SearchRequest(request.getIndexName());searchRequest.source(searchSourceBuilder);//4.查詢elasticsearchSearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);//5.封裝返回結果return buildPorductSearchDTO(searchResponse.getHits(), request.getPageNum(), request.getPageSize());}...
}

6.索引重建

(1)問題分析

(2)解決方案

(3)操作演示

(4)其他說明

(1)問題分析

在實際中可能會遇到正在使?的索引需要變化字段類型、增減字段等，這時可能就需要創建新的mappings。

因為索引正在被應?使?，在進?操作時就要考慮怎么降低對應?的影響，以及如何把?前的數據遷移到新的索引中。

(2)解決方案

可以使?ES的索引別名功能來降低對應?的影響，實現不停機重建索引。可以使?ES的Scroll API + Bulk API，實現把?前的數據遷移到新的索引中。

(3)操作演示

一.假設目前正在被使用的商品索引為sku_info_index
二.首先給sku_info_index索引起別名sku_info_index_alias
三.然后需要新建一個索引sku_info_index_v2
四.接著使用Scroll API + Bulk API遷移數據
五.最后把sku_info_index_alias索引別名指向sku_info_index_v2索引

一.目前正在被使用的商品索引

sku_info_index現在正在被業務使?：

PUT /sku_info_index
{"settings": {"number_of_shards": 3,"number_of_replicas": 1},"mappings":{"properties": {"skuName": {"type": "text","analyzer": "ik_max_word","search_analyzer": "ik_smart"},"basePrice": {"type": "integer"},"vipPrice": {"type": "integer"},"brandId": {"type": "keyword"},"brandName": {"type": "keyword"},"saleCount": {"type": "integer"},"createTime": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"},"updateTime": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"}}}
}

二.給sku_info_index索引起別名

讓應?使?sku_info_index_alias別名來操作數據：

PUT /sku_info_index/_alias/sku_info_index_alias

三.然后需要新建一個索引sku_info_index_v2

新建一個sku_info_index_v2索引：

PUT /sku_info_index
{"settings": {"number_of_shards": 3,"number_of_replicas": 1},"mappings": {"properties": {"skuName": {"type": "text","analyzer": "ik_max_word","search_analyzer": "ik_smart"},"basePrice": {"type": "integer"},"vipPrice": {"type": "integer"},"brandId": {"type": "keyword"},"brandName": {"type": "keyword"},"saleCount": {"type": "integer"},"label": {"type": "integer"},"createTime": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"},"updateTime": {"type": "date","format": "yyyy-MM-dd HH:mm:ss"}}}
}

四.接著使用Scroll API + Bulk API遷移數據

#https://www.elastic.co/guide/en/elasticsearch/reference/7.6/search-request-body.html#request-body-search-scroll
POST /sku_info_index/_search?scroll=1m
{"size": 3,"query": {"match_all": { }}
}POST /_bulk
{"index": {"_index": "sku_info_index_v2","_id": "8000177337"}
}
{"skuName": "Apple iPhone 13 Pro Max 256GB 蒼嶺綠? ?持移動聯通電信5G 雙卡雙待 ?機","brandName": "蘋果","createTime": "2022-03-12 08:24:57","brandId": 4,"vipPrice": 9799,"updateTime": "2022-03-12 08:24:57","basePrice": 9999,"label": "新品"
}
{"index": {"_index": "sku_info_index_v2","_id": "8000177338"}
}
{"skuName": "Apple iPhone 13 (A2634)128GB 綠? ?持移動聯通電信5G 雙卡雙待? 機","brandName": "蘋果","createTime": "2022-03-12 08:24:57","brandId": 4,"vipPrice": 5798,"updateTime": "2022-03-12 08:24:57","basePrice": 5999,"label": "爆品"
}
{"index": {"_index": "sku_info_index_v2","_id": "8000177339"}
}
{"skuName": "蘋果13mini Apple iphone 13 mini 5G新品?機 粉? 128GB","brandName": "蘋果","createTime": "2022-03-12 08:24:57","brandId": 4,"vipPrice": 4900,"updateTime": "2022-03-12 08:24:57","basePrice": 5100,"label": "超值特惠"
}

五.最后把sku_info_index_alias索引別名指向sku_info_index_v2索引

POST /_aliases
{"actions": [{"remove": {"index": "sku_info_index","alias": "sku_info_index_alias"}}, {"add": {"index": "sku_info_index_v2","alias": "sku_info_index_alias"}}]
}

(4)其他說明

一.如果在上ES前，就預計索引結構可能會發?變化。可以?開始就通過索引別名來操作數據，這樣當索引結構需要變更時可按上?的?案及演示實現不停機重建索引。

二.當使?索引別名時，ES Java API的代碼無需任何變化，?如下?是SearchRequest的構造法?法：

public SearchRequest(String... indices) {this(indices, new SearchSourceBuilder());
}

如果直接索引名是example_index，那么創建的SearchRequest對象如下，其中example_index為索引名：

new SearchRequest("example_index")

如果直接索引example_index對應的索引別名是example_index_alias，那么創建的SearchRequest對象就是：

new SearchRequest("example_index_alias")