Canal 数据监控同步缓存数据

刺骨的言语ヽ痛彻心扉 2022-09-02 11:55 257阅读 0赞

canal: 阿里旗下的开源的,java语言的开发

## 作用: ##

用来监控数据库数据的变化,从而获取新增数据或者修改的数据;

## 原理: ##

Canal伪装成mysql的备份机,当mysql数据发生过改变,就会把binary log日志发送给从数据库来保证数据的一致性,然后Canal就可以监控数据的改变

## 实现: ##

mysql需要开启binlog模式  
数据监控微服务:  
1.搭建maven工程  
导入:starter-canal  
spring-rabbit 依赖

<dependency>
            <groupId>com.xpand</groupId>
            <artifactId>starter-canal</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.amqp</groupId>
            <artifactId>spring-rabbit</artifactId>
        </dependency>

2.启动类  
@EnableCanalClient//声明开启Canal的客户端

@SpringBootApplication
    @EnableCanalClient//声明开启Canal的客户端
    public class CanalApplication {
        public static void main(String[] args) {
            SpringApplication.run(CanalApplication.class,args);
        }
    }

3.配置文件

canal.client.instances.example.host=192.168.200.128
    canal.client.instances.example.port=11111
    canal.client.instances.example.batchSize=1000
    spring.rabbitmq.host=192.168.200.128
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=guest
    spring.rabbitmq.password=guest
    spring.rabbitmq.virtual-host=/

4.创建监听类

@CanalEventListener//声明当前类是一个Canal的监听类
    public class MyListener {
    //监听的是哪个数据库,和那张表
          @ListenPoint(schema = "changgou_business" ,table = "tb_ad")
        public void adup(CanalEntry.EntryType entryType,CanalEntry.RowData rowData){
            System.out.println("数据发生变化");
      //获取之前的数据
            rowData.getBeforeColumnsList().forEach((i)->{
                System.out.println("之前的数据"+i.getMysqlType()+i.getName()+i.getValue());
                                 // 获取该字段的类型, 该字段的名称 , 该字段的值
            });
            //获取改变之后的数据
            rowData.getAfterColumnsList().forEach((a)->System.out.println("之后的数据"+a.getName()+a.getValue()));*/
            }
        }
    }

@CanalEventListener//声明当前的类是canal的监听类  
定义监听方法  
两个参数CanalEntry.EventType(当前操作数据库的类型) 和CanalEntry.RowData(当前操作数据库的数据)  
在方法上添加@ListenPoint(schema=“监听哪一个库” , table=“哪一张表”)  
rowData.getBeforeColumnsList()//获取改变之前的数据.forEach(©->system.out.println(“输出”))// 遍历输出  
rowData.getAfterColumnsList()//获取改变之后的数据  
当数据库数据发生改变,canal监控到,就发送消息

## 实现缓存数据和数据库数据保持一致 ##

### 原理: ###

当数据库数据发生变化,Canal监控到,就向rabbitMQ发送消息,将更新的数据给mq,监听mq的类监听到mq的新数据,就同步到缓存中  
实现步骤:  
1.之前有导入spring-rabbit的依赖

2.创建rabbitMQ的config类,配置交换机和队列

@Configuration//声明当前类是个配置类
    public class RabbitMQConfig {
        public static final String AD_QUEUE="ad_queue";
        public static final String AD_EXCHANGE="ad_exchange";
        public static final String KEY="ad_changgou";
        @Bean(AD_QUEUE)//创建队列
        public Queue getQueue(){
            return QueueBuilder.durable(AD_QUEUE).build();
        }
        @Bean(AD_EXCHANGE)//创建交换机
        public Exchange getExchange(){
            //
            return ExchangeBuilder.topicExchange(AD_EXCHANGE).durable(true).build();
    
        }
        @Bean //创建绑定关系
        public Binding getBinding(@Qualifier(AD_EXCHANGE) Exchange exchange,@Qualifier(AD_QUEUE) Queue queue){
            return BindingBuilder.bind(queue).to(exchange).with(KEY).noargs();
        }
    }

当数据库数据发送变化后,

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NTQyMzQ0_size_16_color_FFFFFF_t_70]

队列中就会有消息;  
消费者消费消息,监听这个队列  
1.导入依赖

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
            </dependency>
            
                   <dependency>
                <groupId>org.apache.httpcomponents</groupId>
                <artifactId>httpclient</artifactId>
            </dependency>

2.创建启动类  
3.配置文件  
4.监听类

@Component
    public class Listener {
        @RabbitListener(queues = "ad_queue")
        public void receive(String message){
            System.out.println(message);
            //监听到消息了,要发送请求给linux执行lua脚本,更新数据
            //java代码发送请求
            //创建httpClient对象
            HttpClient client = HttpClients.createDefault();
            //创建一个请求对象(GET/POST/PUT/DELETE...)
            HttpGet httpGet = new HttpGet("http://192.168.200.128/ad_update?position=web_index_lb");
            try {
                HttpResponse response = client.execute(httpGet);
                String s = EntityUtils.toString(response.getEntity());
                System.out.println(s);
            } catch (IOException e) {
                e.printStackTrace();
            }
    
        }
    }

监听到消息就可以做自己的逻辑处理

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQ1NTQyMzQ0_size_16_color_FFFFFF_t_70]: /images/20220829/69c150a2ef1f4229bc38e034b7ba4305.png