【云岚到家】-day04-索引同步-搜索接口

【云岚到家】-day04-索引同步-搜索接口

1.索引同步

1.1 编写同步程序

前面通过配置Canal+MQ的数据同步环境实现了Canal从数据库读取binlog并且将数据写入MQ

下边编写同步程序监听MQ，收到消息后向ES创建索引

1.1.1 创建索引结构

启动ES和kibana：

docker start elasticsearch7.17.7 
docker start kibana7.17.7

浏览器搜索http://192.168.101.68:5601/，点击dev_tools客户端

下边创建索引serve_aggregation，serve_aggregation索引的结构与jzo2o-foundations数据库的serve_sync表结构对应

GET /_cat/indices?v 

如果需要修改索引结构需要删除重新创建：

DELETE 索引名

查询索引结构

GET /索引名/_mapping

创建serve_aggregation索引 （已经存在无法重复创建）

PUT /serve_aggregation
{
  "mappings" : {
     "properties" : {
       "city_code" : {
         "type" : "keyword" 
       },
       "detail_img" : {
         "type" : "text",
         "index" : false
       },
       "hot_time_stamp" : {
         "type" : "long"
       },
       "id" : {
         "type" : "keyword"
       },
       "is_hot" : {
         "type" : "short"
       },
       "price" : {
         "type" : "double"
       },
       "serve_item_icon" : {
         "type" : "text",
         "index" : false
       },
       "serve_item_id" : {
         "type" : "keyword"
       },
       "serve_item_img" : {
         "type" : "text",
         "index" : false
       },
       "serve_item_name" : {
         "type" : "text",
         "analyzer": "ik_max_word",
         "search_analyzer":"ik_smart"
         
       },
       "serve_item_sort_num" : {
         "type" : "short"
       },
       "serve_type_icon" : {
         "type" : "text",
         "index" : false
       },
       "serve_type_id" : {
         "type" : "keyword"
       },
       "serve_type_img" : {
         "type" : "text",
         "index" : false
       },
       "serve_type_name" : {
         "type" : "text",
         "analyzer": "ik_max_word",
         "search_analyzer":"ik_smart"
       },
       "serve_type_sort_num" : {
         "type" : "short"
       }
     }
   }
}

1.1.2 编写同步程序

1.添加依赖

首先在foundations工程添加下边的依赖

<dependency>
    <groupId>com.jzo2o</groupId>
    <artifactId>jzo2o-canal-sync</artifactId>  <!--专门做数据同步的 -->
</dependency>
<dependency>
    <groupId>com.jzo2o</groupId>
    <artifactId>jzo2o-es</artifactId>
</dependency>

2.配置连接 ES

修改foundations的配置文件：

修改nacos中es的配置文件

修改nacos中rabbitmq的配置文件

3.编写同步程序

同步程序继承AbstractCanalRabbitMqMsgListener类，泛型中指定同步表对应的类型。

根据数据同步环境去配置监听MQ：

package com.jzo2o.foundations.handler;

/**
 * 服务信息同步程序，实现canal同步数据的
 **/
@Component
@Slf4j
public class ServeCanalDataSyncHandler extends AbstractCanalRabbitMqMsgListener<ServeSync> {

    @Resource
    private ElasticSearchTemplate elasticSearchTemplate;


    //监听mq
    //@RabbitListener(queues = "canal-mq-jzo2o-foundations", concurrency = "1")
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "canal-mq-jzo2o-foundations"),//队列名称
            exchange = @Exchange(name = "exchange.canal-jzo2o", type = ExchangeTypes.TOPIC),//监听exchange.canal-jzo2o这个交换机
            key = "canal-mq-jzo2o-foundations"),
            concurrency = "1" //指定消费线程数为1
    )
    public void onMessage(Message message) throws Exception {
        //调用抽象类中的方法
        parseMsg(message);
    }

    /**
     * 向es中保存数据，解析到binlog中的新增、更新信息都执行此方法
     */
    @Override
    public void batchSave(List<ServeSync> data) {
        //向es中保存索引，有索引就保存，没有就添加
        Boolean aBoolean = elasticSearchTemplate.opsForDoc().batchInsert(IndexConstants.SERVE, data);
        //如果执行失败，就要抛出异常，给mq回nack
        if(!aBoolean){
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步失败");
        }
    }

    /**
     * 解析到binlog中的删除信息，将es中的指定的文档进行删除
     */
    @Override
    public void batchDelete(List<Long> ids) {
        Boolean aBoolean = elasticSearchTemplate.opsForDoc().batchDelete(IndexConstants.SERVE, ids);
        if(!aBoolean){
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("同步失败");
        }
    }
}

concurrency = “1”：表示消费线程数为1。

在同步程序中需要根据业务需求编写同步方法，当服务下架时会删除索引需要重写抽象类中的batchDelete(List ids)方法，此方法是当删除Serve_sync表的记录时 对索引执行删除操作。

当服务上架后需要添加索引，当服务信息修改时需要修改索引，需要重写抽象类中的batchSave(List data)方法，此方法是当向Serve_sync表新增或修改记录时对索引执行添加及修改操作

1.1.3 测试

启动jzo2o-foundations服务

启动成功,jzo2o-foundations服务作为MQ的消费者和MQ建立通道，进入canal-mq-jzo2o-foundations队列的管理界面，查看是否建立 了监听通道

监听通道建立 成功，下边在同步程序打断点：

手动修改jzo2o-foundations数据库的serve_sync表的记录，这里修改了服务项的名称

正常执行同步程序：

放行继续执行到batchSave方法：

ES服务正常，同步方法执行成功后进入Kibana查看

执行命令：

GET /serve_aggregation/_search
{
}

查询服务信息与数据库serve_sync表中1686352662791016449记录的信息一致

下边再将服务项名称恢复

再进入Kibana查看索引的内容与数据库一致

小结

编写同步程序的步骤：

1.根据数据库同步表的结构，创建索引结构

2.同步程序监听MQ的同步队列

3.同步程序收到数据同步消息写入Elasticsearch，写的失败抛出异常，消息回到MQ

1.2 管理同步表

通过测试Canal+MQ同步流程，只有当serve_sync表变化时才会触发同步，serve_sync表什么时候变化 ？

当服务信息变更时需要同时修改serve_sync表，下边先分析serve_sync的变化需求，再进行代码实现。

1.2.1 管理同步表需求

现在如何去维护serve_sync这张表呢？

根据serve_sync表的结构分析：

添加：区域服务上架向serve_sync表添加记录，同步程序新增索引记录

删除：区域服务下架从serve_sync表删除记录，同步程序删除索引记录

修改：

修改服务项修改serve_sync的记录

修改服务分类修改serve_sync的记录

修改服务价格修改serve_sync的记录

设置热门/取消热门修改serve_sync的记录

1.2.2 代码实现

1.区域服务上架向serve_sync表添加记录

在ServeServiceImpl增加私有方法，如下：

/**
 * 新增服务同步数据
 *
 * @param serveId 服务id
 */
private void addServeSync(Long serveId) {
    //服务信息
    Serve serve = baseMapper.selectById(serveId);
    //区域信息
    Region region = regionMapper.selectById(serve.getRegionId());
    //服务项信息
    ServeItem serveItem = serveItemMapper.selectById(serve.getServeItemId());
    //服务类型
    ServeType serveType = serveTypeMapper.selectById(serveItem.getServeTypeId());

    ServeSync serveSync = new ServeSync();
    //下面把数据放到serveSync
    serveSync.setServeTypeId(serveType.getId());
    serveSync.setServeTypeName(serveType.getName());
    serveSync.setServeTypeIcon(serveType.getServeTypeIcon());
    serveSync.setServeTypeImg(serveType.getImg());
    serveSync.setServeTypeSortNum(serveType.getSortNum());

    serveSync.setServeItemId(serveItem.getId());
    serveSync.setServeItemIcon(serveItem.getServeItemIcon());
    serveSync.setServeItemName(serveItem.getName());
    serveSync.setServeItemImg(serveItem.getImg());
    serveSync.setServeItemSortNum(serveItem.getSortNum());
    serveSync.setUnit(serveItem.getUnit());
    serveSync.setDetailImg(serveItem.getDetailImg());
    serveSync.setPrice(serve.getPrice());

    serveSync.setCityCode(region.getCityCode());
    serveSync.setId(serve.getId());
    serveSync.setIsHot(serve.getIsHot());
    serveSyncMapper.insert(serveSync);
}

修改服务上架的方法：

@Override
@Transactional
@CachePut(value = RedisConstants.CacheName.SERVE, key = "#id",  cacheManager = RedisConstants.CacheManager.ONE_DAY)
public Serve onSale(Long id){
    ...
    //添加同步表，向serve_sync表写记录
    addServeSync(id);
    return baseMapper.selectById(id);

}

2.区域服务下架从serve_sync表删除记录

3.修改服务项修改serve_sync的记录

4.修改服务分类修改serve_sync的记录

1.2.3 测试

测试服务上架添加索引

测试服务下架删除索引

测试修改服务价格修改索引

测试修改服务项名称修改索引

测试修改修改服务分类名称修改索引

2.搜索接口

目标：开发搜索接口

2.1 定义接口

参数内容：区域编码，服务类型id、关键字

区域编码：用户定位成功前端记录区域编码（city_code），搜索时根据city_code搜索该区域的服务

服务类型id：在全部服务界面选择一个服务类型查询其它下的服务列表

关键字：输入关键字搜索服务项名称、服务类型名称

接口名称：服务搜索接口

接口路径：GET/foundations/customer/serve/search

编写controller方法：

@RestController("consumerServeController")
@RequestMapping("/customer/serve")
@Api(tags = "用户端 - 首页服务查询接口")
public class FirstPageServeController {
...
@GetMapping("/search")
@ApiOperation("首页服务搜索")
@ApiImplicitParams({
        @ApiImplicitParam(name = "cityCode", value = "城市编码", required = true, dataTypeClass = String.class),
        @ApiImplicitParam(name = "serveTypeId", value = "服务类型id", dataTypeClass = Long.class),
        @ApiImplicitParam(name = "keyword", value = "关键词", dataTypeClass = String.class)
})
public List<ServeSimpleResDTO> findServeList(@RequestParam("cityCode") String cityCode,
 @RequestParam(value = "serveTypeId", required = false) Long serveTypeId,
@RequestParam(value = "keyword", required = false) String keyword) {
    List<ServeSimpleResDTO> serveList = serveAggregationService.findServeList(cityCode, serveTypeId, keyword);
    return serveList;
}

ServeSimpleResDTO

package com.jzo2o.foundations.model.dto.response;

/**
 * 服务简略响应信息
 **/
@Data
@ApiModel("服务简略响应信息")
public class ServeSimpleResDTO {

    /**
     * 服务id
     */
    @ApiModelProperty("服务id")
    private Long id;

    /**
     * 服务项id
     */
    @ApiModelProperty("服务项id")
    private Long serveItemId;

    /**
     * 服务项名称
     */
    @ApiModelProperty("服务项名称")
    private String serveItemName;

    /**
     * 服务项图标
     */
    @ApiModelProperty("服务项图标")
    private String serveItemIcon;

    /**
     * 服务项排序字段
     */
    @ApiModelProperty("服务项排序字段")
    private Integer serveItemSortNum;
}

2.2 搜索方法

首先通过ES的查询语言进行查询，如下：

GET /serve_aggregation/_search
{
   "query" : {
      "bool" : {
         "must" : [
            {
               "term" : {
                  "city_code" : {
                     "value" : "010"
                  }
               }
            },
            {
               "multi_match" : {
                  "fields" : [ "serve_item_name", "serve_type_name" ],
                  "query" : "保洁"
               }
            }
         ]
      }
   },
   "sort" : [
      {
         "serve_item_sort_num" : {
            "order" : "asc"
         }
      }
   ]
}

2.3 service方法

下边按照ES查询语句编写service方法：

定义service接口：

package com.jzo2o.foundations.service;

public interface ServeAggregationService {

    /**
     * 查询服务列表
     *
     * @param cityCode    城市编码
     * @param serveTypeId 服务类型id
     * @param keyword     关键词
     * @return 服务列表
     */
    List<ServeSimpleResDTO> findServeList(String cityCode, Long serveTypeId, String keyword);
}

service实现类

package com.jzo2o.foundations.service.impl;

/**
 * 服务相关
 **/
@Slf4j
@Service
public class ServeAggregationServiceImpl implements ServeAggregationService {

    @Resource
    private ElasticSearchTemplate elasticSearchTemplate;

    /**
     * 查询服务列表
     * @param cityCode    城市编码
     * @param serveTypeId 服务类型id
     * @param keyword     关键词
     * @return 服务列表
     */
    @Override
public List<ServeSimpleResDTO> findServeList(String cityCode, Long serveTypeId, String keyword) {
    // 构造查询条件
    SearchRequest.Builder builder = new SearchRequest.Builder();

    builder.query(query->query.bool(bool->{
        //匹配citycode
        bool.must(must->
                must.term(term->
                        term.field("city_code").value(cityCode)));
        //todo 匹配服务类型

        //匹配关键字
        if(ObjectUtils.isNotEmpty(keyword)){
            bool.must(must->
                must.multiMatch(multiMatch->
                    multiMatch.fields("serve_item_name","serve_type_name").query(keyword)));
        }
        return bool;
    }));
    // 排序 按服务项的serveItemSortNum排序(升序)
    List<SortOptions> sortOptions = new ArrayList<>();
    sortOptions.add(SortOptions.of(sortOption -> sortOption.field(field->field.field("serve_item_sort_num").order(SortOrder.Asc))));
    builder.sort(sortOptions);
    //指定索引
    builder.index("serve_aggregation");
    //请求对象
    SearchRequest searchRequest = builder.build();
    // 检索数据
    SearchResponse<ServeAggregation> searchResponse = elasticSearchTemplate.opsForDoc().search(searchRequest, ServeAggregation.class);
    //如果搜索成功返回结果集
    if (SearchResponseUtils.isSuccess(searchResponse)) {
        List<ServeAggregation> collect = searchResponse.hits().hits()
                .stream().map(hit -> {
                    ServeAggregation serve = hit.source();
                    return serve;
                })
                .collect(Collectors.toList());
        List<ServeSimpleResDTO> serveSimpleResDTOS = BeanUtil.copyToList(collect, ServeSimpleResDTO.class);
        return serveSimpleResDTOS;
    }
    return  Collections.emptyList();

}

}

2.5 测试

进入小程序，进入搜索界面

跟踪Network

查看response

{"code":200,
 "msg":"OK",
 "data": [
    {"id":"1693815624114970626","serveItemId":"1685894105234755585","serveItemName":"日常保洁","serveItemIcon":"https://yjy-xzbjzfw-oss.oss-cn-hangzhou.aliyuncs.com/072e4709-9be3-4df2-96ad-c3e2d5790556.png","serveItemSortNum":1}
 ]
}

原文地址：https://blog.csdn.net/weixin_69595694/article/details/145261356

免责声明：本站文章内容转载自网络资源，如侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！