更改商品信息消息队列通知ES更新与静态页生成

（1）创建配置类

package com.example.canatest.config;

import org.springframework.amqp.core.*;
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 {
    //负责更新ES
    public static String ADD_ES_QUEUE="add_es_queue";
    //负责生成商品详情静态页
    public static String CREATE_PAGE_QUEUE="create_page_queue";
    //交换机
    public static String ES_PAGE_EXCHANGE="es_page_exchange";

    @Bean
    public Queue queue(){
        return new Queue(ADD_ES_QUEUE);
    }
    @Bean
    public Queue queue2(){return new Queue(CREATE_PAGE_QUEUE);}
    @Bean
    public TopicExchange topicExchange(){
        return new TopicExchange(ES_PAGE_EXCHANGE);
    }
    //绑定交换机
    @Bean
    public Binding topicExchageBingQueue(){
    return BindingBuilder.bind(queue()).to(topicExchange()).with("it.*");

    }

    //绑定交换机
    @Bean
    public Binding topicExchageBingQueue2(){

      return BindingBuilder.bind(queue2()).to(topicExchange()).with("it.#");
    }
}

(2)商品更改canal发送消息

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;

/**
 * @author chen.qian
 * @date 2018/3/19
 */
@CanalEventListener
public class MyEventListener {
    @Autowired
    private AdFeign  adFeign;
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RabbitTemplate rabbitTemplate;
        @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){
            Object beforeis_marketable = beforeMap.get("is_marketable");
            Object afteris_marketable = afterMap.get("is_marketable");
            if(beforeis_marketable.equals("0")&&afteris_marketable.equals("1")){
                rabbitTemplate.convertAndSend(RabbitCofig.ES_PAGE_EXCHANGE,"it.haha",afterMap.get("id"));
                System.out.println("spu有上架，id为："+afterMap.get("id"));
            }
        }
    }
//    @InsertListenPoint
//    public void onEvent(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
//        rowData.getAfterColumnsList().forEach((c) -> System.err.println("By--Annotation: " + c.getName() + " ::   " + c.getValue()));
//    }
//
//    @UpdateListenPoint
//    public void onEvent1(CanalEntry.RowData rowData) {
//        System.err.println("UpdateListenPoint");
//        rowData.getAfterColumnsList().forEach((c) -> System.err.println("By--Annotation: " + c.getName() + " ::   " + c.getValue()));
//    }
//
//    @DeleteListenPoint
//    public void onEvent3(CanalEntry.EventType eventType) {
//        System.err.println("DeleteListenPoint");
//    }

    @ListenPoint(destination = "example", schema = "changgou_business", table = {"tb_ad"}, eventType = {CanalEntry.EventType.UPDATE,CanalEntry.EventType.DELETE,CanalEntry.EventType.INSERT})
    public void onEvent4(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        System.err.println("广告跟新了");
//        rowData.getAfterColumnsList().forEach((c) -> System.err.println("By--Annotation: " + c.getName() + " ::   " + c.getValue()));
        String position = rowData
                .getAfterColumnsList()
                .stream()
                .filter(column -> column.getName().equals("positon"))
                .limit(1)
                .collect(toList())
                .get(0)
                .getValue();

        System.out.println("position:" + position);

        List<Ad> adByPosition = adFeign.findAdByPosition(position);
        System.out.println(adByPosition);
        String string = JSON.toJSONString(adByPosition);
        redisTemplate.boundValueOps("ad_"+position).set(string);
    }
}

(3)监听消息队列
生成静态页

package com.changgou.web.item.listenter;

import com.changgou.web.item.service.ItemService;
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/26
 * @description ：
 * @version: 1.0
 */
@Component
@RabbitListener(queues = {"create_page_queue"})
public class Mqlistenter {
    @Autowired
    private ItemService itemService;
    @RabbitHandler
    public void getPage(String spuid){
        System.out.println("生成静态页："+spuid);
        itemService.staticPage(spuid);
    }
}

更新ES

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){
        boolean b = searchService.insertSkuTo(id);
        if(b){
            System.out.println("保存到es成功！");
        }
    }
}