电商项目-数据同步解决方案（四）商品下架同步更新ES索引库数据

商品下架索引库删除数据

一、 需求分析和业务逻辑

商品下架后将商品从索引库中移除。

主要应用技术有： 消息队列-RabbitMQ ，分布式搜索引擎-ElasticSearch，Eureka，Canal，Feign远程调用

（1）在数据监控微服务中监控tb_spu表的数据，当tb_spu发生更改且is_marketable为0时，表示商品下架，将spu的id发送到rabbitmq。

（2）在rabbitmq管理后台创建商品下架交换器（fanout）。使用分列模式的交换器是考虑商品下架会有很多种逻辑需要处理，索引库删除数据只是其中一项，另外还有删除商品详细页等操作。

（3）搜索微服务从rabbitmq的的队列中提取spu的id，通过调用elasticsearch的高级restAPI 将相关的sku列表从索引库删除。

二、 代码实现

商品下架删除ES索引库数据的功能实现：

主要应用技术有： 消息队列-RabbitMQ ，分布式搜索引擎-ElasticSearch，Eureka，Canal，Feign远程调用

2.1 创建交换器与队列

完成商品下架交换器的创建，队列的创建与绑定，将spuId发送消息到mq

商品下架交换器：goods_down_exchange

队列名称： search_delete_queue

绑定 search_delete_queue到goods_down_exchange

@Configuration
public class RabbitMQConfig {
​//定义交换机名称public static final String GOODS_UP_EXCHANGE="goods_up_exchange";public static final String GOODS_DOWN_EXCHANGE="goods_down_exchange";
​//定义队列名称public static final String AD_UPDATE_QUEUE="ad_update_queue";public static final String SEARCH_ADD_QUEUE="search_add_queue";public static final String SEARCH_DEL_QUEUE="search_del_queue";
​//声明队列@Beanpublic Queue queue(){return new Queue(AD_UPDATE_QUEUE);}@Bean(SEARCH_ADD_QUEUE)public Queue SEARCH_ADD_QUEUE(){return new Queue(SEARCH_ADD_QUEUE);}@Bean(SEARCH_DEL_QUEUE)public Queue SEARCH_DEL_QUEUE(){return new Queue(SEARCH_DEL_QUEUE);}
​//声明交换机@Bean(GOODS_UP_EXCHANGE)public Exchange GOODS_UP_EXCHANGE(){return ExchangeBuilder.fanoutExchange(GOODS_UP_EXCHANGE).durable(true).build();}@Bean(GOODS_DOWN_EXCHANGE)public Exchange GOODS_DOWN_EXCHANGE(){return ExchangeBuilder.fanoutExchange(GOODS_DOWN_EXCHANGE).durable(true).build();}
​
​//队列与交换机的绑定@Beanpublic Binding GOODS_UP_EXCHANGE_BINDING(@Qualifier(SEARCH_ADD_QUEUE)Queue queue,@Qualifier(GOODS_UP_EXCHANGE)Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with("").noargs();}@Beanpublic Binding GOODS_DOWN_EXCHANGE_BINDING(@Qualifier(SEARCH_DEL_QUEUE)Queue queue,@Qualifier(GOODS_DOWN_EXCHANGE)Exchange exchange){return BindingBuilder.bind(queue).to(exchange).with("").noargs();}
​
}

2.2 canal监听下架

修改changgou_canal的SpuListener的spuUpdate方法，添加以下代码

//获取最新下架的商品 1->0
if ("1".equals(oldData.get("is_marketable")) && "0".equals(newData.get("is_marketable"))){//将商品的spuid发送到mqrabbitTemplate.convertAndSend(RabbitMQConfig.GOODS_DOWN_EXCHANGE,"",newData.get("id"));
}

2.3 根据spuId删除索引数据

编写业务逻辑，实现根据spuId删除索引库数据的方法。

（1）ESManagerService新增方法定义

//根据souid删除es索引库中相关的sku数据
void delDataBySpuId(String spuId);

（2）ESManagerServiceImpl实现方法

@Override
public void delDataBySpuId(String spuId) {List<Sku> skuList = skuFeign.findSkuListBySpuId(spuId);if (skuList == null || skuList.size()<=0){throw new RuntimeException("当前没有数据被查询到,无法导入索引库");}for (Sku sku : skuList) {esManagerMapper.deleteById(Long.parseLong(sku.getId()));}
}

2.4 接收mq消息，执行索引库删除

从rabbitmq中提取消息，调动根据spuId删除索引库数据的方法 changgou_service_search新增监听类

@Component
public class GoodsDelListener {
​@Autowiredprivate ESManagerService esManagerService;
​@RabbitListener(queues = RabbitMQConfig.SEARCH_DEL_QUEUE)public void receiveMessage(String spuId){System.out.println("删除索引库监听类,接收到的spuId:  "+spuId);
​//调用业务层完成索引库数据删除esManagerService.delDataBySpuId(spuId);}
}