利用Canal实现数据库改动就更新redis功能

此功能设计到三个微服务：
service_business: 负责查询数据库数据。
service_business_api：存feigin远程调用接口。
service_canal:负责canal功能。

service_business

controller

@RestController
@CrossOrigin
@RequestMapping("/ad")
public class AdController {


    @Autowired
    private AdService adService;
    @GetMapping("/findAdByPosition")
   public List<Ad> findAdByPosition(String position){
       List<Ad> adList=adService.findAdByPosition(position);
       return adList;
    }

这里注意返回类型，这里一般返回Resoult，但这里是集合。

接口及实现类省略。。。

service_business_api

feign接口：

/**
 * @author ：gzy
 * @date ：Created in 2019/8/19
 * @description ：
 * @version: 1.0
 */
@FeignClient(name = "business")
@RequestMapping("/ad")
public interface AdFeign {
    @GetMapping("/findAdByPosition")
    List<Ad> findAdByPosition(@RequestParam(name = "position") String position);
}

service_canal

首先：启动类添加feign客户端

@SpringBootApplication
@EnableCanalClient
//找到feign的接口
@EnableFeignClients(basePackages = "com.changgou.business.feign")
public class CanaTestApplication {

    public static void main(String[] args) {
        SpringApplication.run(CanaTestApplication.class, args);
    }
}

然后引入service_business_api

<!--用于feign-->
        <dependency>
            <groupId>com.gzy</groupId>
            <artifactId>service_business_api</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

因为后面要是用Redis,所以在yml配置文件加入redis的IP配置。

最后使用：

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.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundValueOperations;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.util.List;
import java.util.stream.Collectors;

import static java.util.stream.Collectors.toList;

/**
 * @author gzy
 * @date 2019/8/19
 */
@CanalEventListener
public class MyEventListener {
    @Autowired
    private AdFeign  adFeign;
    @Autowired
    private StringRedisTemplate redisTemplate;
//    @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);
    }
}

只要数据库有变动canal就会在bin-log日志中监控到，然后就会自动更新redis了。