canal搭配RabbitMQ实现数据同步解决方案_canal 同步消息到rabbitmq

目录

1.cannal的使用及原理

2.docker安装cannal

3.在java中使用cannal及RabbitMQ

---

1.cannal的使用及原理

cannal主要支持MySQL数据库，它会将自己伪装成从节点产生与主节点的交互操作，主节点一旦数据发生变更也就是发生增删改的操作就会将这些信息写入到binlog日志中并且推送到从节点也就是cannal，然后从节点去执行信息里增删改的操作，从而实现数据主从节点数据一致。

2.docker安装cannal

1.下载镜像

docker pull docker.io/canal/canal-server

2.容器安装

docker run -p 11111:11111 --name canal -d docker.io/canal/canal-server

进入容器,修改核心配置canal.properties 和instance.properties，canal.properties 是canal自身的配置，instance.properties是需要同步数据的数据库连接配置。

执行代码如下:

docker exec -it canal /bin/bash
cd canal-server/conf/
vi canal.properties
cd example/
vi instance.properties

修改canal.properties的id，不能和mysql的server-id重复，如下图：

修改instance.properties,配置数据库连接地址:   其实这个不做也可以在Java代码中也可以指定监听那个数据库那张表

设置开机启动，并重启cannal

docker update --restart=always canal
docker restart canal

3.在java中使用cannal及RabbitMQ

1.POM.XML

#cannal
<dependency>
<groupId>com.xpand</groupId>
<artifactId>starter-canal</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
 
#RabbitMQ
 <dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

2.application.yml

#canal配置
canal:
  client:
    instances:
    # exmaple
      example:
        host: 127.0.0.1
        port: 11111

3.SpringBoot启动类

@SpringBootApplication
@EnableCanalClient   #声明当前服务是cannal监控数据变化客户端
public class CanalApplication {
 
    public static void main(String[] args) {
        SpringApplication.run(CanalApplication.class,args);
    }
}

4.实例代码cannal将监听到的数据通过MQ进行发送

/**
 * @author JiaWei
 */
@Slf4j
@CanalEventListener //声明当前的类是canal的监听类
public class MycannalListener {
 
    @Autowired
    RabbitTemplate rabbitTemplate;
 
    /**
     * @param eventType 当前操作数据库
     * @param rowData   当前操作数据库数据
     */
    @ListenPoint(schema = "changgou_businness",table = "tb_ad")
    public void adUpdate(CanalEntry.EventType eventType, CanalEntry.RowData rowData) {
        System.out.println("广告数据发生改变");
        //改变之前的数据
        rowData.getBeforeColumnsList().forEach((x) -> System.out.println("改变前的数据:" + x.getName() + "::" + x.getValue()));
        //改变之后的数据
        rowData.getAfterColumnsList().forEach((x) -> System.out.println("改变后的数据:" + x.getName() + "::" + x.getValue()));
 
        for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
            if ("position".equals(column)){
            log.info("cannal监听到的数据为{}",column + "已经发送到MQ队列为{}",RabbitMQConfig.AD_UPDATE_QUEUE);
                rabbitTemplate.convertAndSend("", RabbitMQConfig.AD_UPDATE_QUEUE,column);
            }
        }
 
    }
 
}
 

5.MQ消费者

package com.changgou.business.listener;
import okhttp3.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import java.io.IOException;
/**
 * @author JiaWei
 */
@Component
public class AdListener {
    @RabbitListener(queues = "ad_update_queue")
    public void receviceMessage(String message){
        System.out.println("接收到的消息为" + message);
        
        《《《《业务逻辑》》》》 可以是feign调用
}