06-22 23:45 中山大学后端发布于广东

关注

商品中心—10.商品B端搜索系统的说明文档

大纲

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 {
    //配置内容对象
    @Autowired
    private 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 {
    @Autowired
    private MessageHandlerManager messageHandlerManager;

    @Override
    public 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都会被注入到这里
    @Autowired
    private 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 {
    @Autowired
    private ProductSearchRepository productSearchRepository;
    
    @Override
    public String tableName() {
        return "sku_info";
    }
    
    @Override
    public void handleMessage(TableDataChangeDTO tableDataChangeDTO) throws IOException {
        String skuId = String.valueOf(tableDataChangeDTO.getKeyId());
        //到数据库查询索引相关的信息
        ProductSearchDO productSearchDO = productSearchRepository.queryProductSearchInfo(skuId);
        //保存索引数据到ES
        productSearchRepository.saveProductSearchInfos(Collections.singletonList(productSearchDO));
    }
}

@Repository
public class ProductSearchRepository {
    private static final String SKU_INFO_INDEX = "sku_info_index";
    
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    
    @Autowired
    private 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 {
    @Autowired
    private ProductSearchRepository productSearchRepository;
    
    @Override
    public String tableName() {
        return "item_info";
    }
    
    @Override
    public 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";
    
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    
    @Autowired
    private 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 {
    @Resource
    private RedisCache redisCache;
    
    @Resource
    private ProductSearchRepository productSearchRepository;
    
    //保存历史搜索词接口
    @Override
    public 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 {
    //用户id
    private Long userId;
    //新的历史搜索词
    private String historySearchWord;
}

(2)商品B端查询历史搜索词的接⼝

使用场景：展示商家的搜索历史记录的时候
接口说明：从Redis列表中查询商家的历史搜索词

//商品搜索服务
@DubboService(version = "1.0.0", interfaceClass = ProductSearchApi.class, retries = 0)
public class ProductSearchApiImpl implements ProductSearchApi {
    @Resource
    private RedisCache redisCache;
    
    @Resource
    private ProductSearchRepository productSearchRepository;
    
    //查询历史搜索词接口
    @Override
    public 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 {
    //用户id
    private 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 {
    @Autowired
    private CompletionSearchWordService completionSearchWordService;
    
    //保存搜索补全词接口
    @Override
    public 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 {
    @Autowired
    private CompletionSearchWordRepository completionSearchWordRepository;
    
    //保存搜索补全词
    @Override
    public 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 {
    @Autowired
    private 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 {
    @Autowired
    private CompletionSearchWordService completionSearchWordService;
    
    //查询补全词接口
    @Override
    public 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 {
    @Autowired
    private CompletionSearchWordRepository completionSearchWordRepository;
    
    //查询搜索补全词
    @Override
    public 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 {
    @Autowired
    private 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 {
    @Resource
    private RedisCache redisCache;
    
    @Resource
    private ProductSearchRepository productSearchRepository;
    ...
    
    //搜索词补全接口
    @Override
    public 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";
    
    @Resource
    private 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.查询elasticsearch
        SearchResponse 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 {
    ...
    //商品搜索查询接口
    @Override
    public 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";
    
    @Resource
    private 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.查询elasticsearch
        SearchResponse 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 {
    ...
    //商品结构化查询接口
    @Override
    public 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 DSL
    private Map<String, Object> queryDsl;
}

//商品搜索
@Repository
public class ProductSearchRepository {
    private static final String MY_SUGGEST = "my_suggest";
    
    @Resource
    private RestHighLevelClient restHighLevelClient;
    ...
    
    //商品结构化查询
    public PorductSearchDTO structuredSearchProducts(ProductStructuredQuery request) throws IOException {
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.trackTotalHits(true);
  
        //1.解析queryDSL
        String 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.查询elasticsearch
        SearchResponse 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")

后端技术栈的基础修养文章被收录于专栏

详细介绍后端技术栈的基础内容，包括但不限于：MySQL原理和优化、Redis原理和应用、JVM和G1原理和优化、RocketMQ原理应用及源码、Kafka原理应用及源码、ElasticSearch原理应用及源码、JUC源码、Netty源码、zk源码、Dubbo源码、Spring源码、Spring Boot源码、SCA源码、分布式锁源码、分布式事务、分库分表和TiDB、大型商品系统、大型订单系统等