canal的使用

• 2020 年 12 月 4 日
• 筆記

• mysql開啟binlog模式
• 查看mysql是否開啟binlog模式

SHOW VARIABLES LIKE 『%log_bin%『 </DI< div>

• 修改/etc/my.cnf 需要開啟binlog模式

[mysqld]

log-bin=mysql-bin

binlog-format=ROW

server_id=1 </DI< div>

• 修改完成之後，重啟mysqld的服務。
• canal服務端安裝配置
• 下載地址canal
• //github.com/chenqian56131/spring-boot-starter-canal
• 下載好之後安裝本地倉庫，導入依賴

<dependency>

<groupId>com.xpand</groupId>

<artifactId>starter-canal</artifactId>

<version>0.0.1-SNAPSHOT</version>

</dependency>

canal使用

• 啟動類添加

@EnableCanalClient

• 使用

package com.changgou.canal.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.changgou.canal.config.RabbitMQConfig;
import com.xpand.starter.canal.annotation.CanalEventListener;
import com.xpand.starter.canal.annotation.ListenPoint;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;

@CanalEventListener //聲明當前的類是canal的監聽類
public class BusinessListener {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    /**
     *
     * @param eventType 當前操作資料庫的類型
     * @param rowData 當前操作資料庫的數據
     */
    @ListenPoint(schema = "changgou_business",table = "tb_ad")
    public void adUpdate(CanalEntry.EventType eventType,CanalEntry.RowData rowData){
        System.out.println("廣告表數據發生改變");
        //獲取改變之前的數據
        //rowData.getBeforeColumnsList().forEach((c)-> System.out.println("改變前的數據:"+c.getName()+"::"+c.getValue()));

        //獲取改變之後的數據
        //rowData.getAfterColumnsList().forEach((c)-> System.out.println("改變之後的數據:"+c.getName()+"::"+c.getValue()));

        for (CanalEntry.Column column : rowData.getAfterColumnsList()) {
            if ("position".equals(column.getName())){
                System.out.println("發送最新的數據到MQ:"+column.getValue());

                //發送消息
                rabbitTemplate.convertAndSend("", RabbitMQConfig.AD_UPDATE_QUEUE,column.getValue());
            }
        }
    }
}