使用RabbitMQ+Canal实现Es商品更新功能
2019-08-24 本文已影响0人
月哥说了算
涉及微服务:
service_canal:负责监控数据库,当数据库有改动,使用rabbitq发送消息商品id。
service_search:负责操作Es,接收消息负责修改Es。
(1)引入pom
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
(2)两个工程application.yml加入配置
spring:
rabbitmq:
host: 192.168.200.128
port: 5672
virtual-host: /350
username: gzy
password: gzy
(3)service_canal创建队列
package com.example.canatest.config;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author :gzy
* @date :Created in 2019/8/24
* @description :
* @version: 1.0
*/
@Configuration
public class RabbitCofig {
public static String ADD_ES_QUEUE="add_es_queue";
@Bean
public Queue queue(){
return new Queue(ADD_ES_QUEUE);
}
}
(4)service_canal的Listener 代码
package com.example.canatest.config;
import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.changgou.business.feign.AdFeign;
import com.changgou.pojo.Ad;
import com.xpand.starter.canal.annotation.*;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static java.util.stream.Collectors.toList;
@CanalEventListener
public class MyEventListener {
@Autowired
private AdFeign adFeign;
@Autowired
private StringRedisTemplate redisTemplate;
@Autowired
private RabbitTemplate rabbitTemplate;
// canal名称,数据库名,表名
@UpdateListenPoint(destination = "example",schema = "changgou_goods",table = {"tb_spu"})
public void onEvent1(CanalEntry.RowData rowData) {
System.err.println("UpdateListenPoint");
Map beforeMap=new HashMap<>();
Map afterMap=new HashMap<>();
//遍历得到修改数据库之前的数据
rowData.getBeforeColumnsList().forEach((c)->beforeMap.put(c.getName(),c.getValue()));
//遍历得到修改数据库后的数据
rowData.getAfterColumnsList().forEach((c)->afterMap.put(c.getName(),c.getValue()));
if(null!=beforeMap&&beforeMap.size()>0&&null!=afterMap&&afterMap.size()>0){
//is_marketable 0,未上架,1,上架
Object beforeis_marketable = beforeMap.get("is_marketable");
Object afteris_marketable = afterMap.get("is_marketable");
if(beforeis_marketable.equals("0")&&afteris_marketable.equals("1")){
//发送修改商品信息id消息,简单模式
rabbitTemplate.convertAndSend(RabbitCofig.ADD_ES_QUEUE,afterMap.get("id"));
System.out.println("spu有上架,id为:"+afterMap.get("id"));
}
}
}
}
(5)service_search
监听消息
package com.changgou.search.listenter;
import com.changgou.goods.feign.GoodsFeign;
import com.changgou.search.service.SearchService;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* @author :gzy
* @date :Created in 2019/8/24
* @description :
* @version: 1.0
*/
@Component
@RabbitListener(queues = {"add_es_queue"})
public class RabbitListenter {
@Autowired
private SearchService searchService;
@RabbitHandler
public void update(String id){
//修改Es中商品
boolean b = searchService.insertSkuTo(id);
if(b){
System.out.println("保存到es成功!");
}
}
}