使用RabbitMQ+Canal实现Es商品更新功能

涉及微服务： 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成功！");
        }
    }
}