2 监听mysql表内容变化，使用canal_tianyaleixiaowu的专栏-CSDN博客_canal监听mysql变化

安宇雨 - 随手采集
2021-05-25 14:03:48
随手采集
0000-未整理-等待研究

mysql本身是支持主从的（master slave），原理就是master产生的binlog日志记录了所有的增删改语句，将binlog发送到slave节点进行执行即可完成数据的同步。

canal是阿里开源的一个中间件，它就是通过解析binlog来完成数据变更的监听的。

可以看到，canal是这样工作的：canal有一个server工程，该server自己伪装为mysql的一个slave节点，然后向master请求所有的变更日志。同时canal有一个client工程，通过添加client的sdk，我们就可以在项目里监听到server端传来的数据变更信息，从而达到监听数据变化的目的。

安装配置canal

在这里https://github.com/alibaba/canal/releases，下载canal releases包，如https://github.com/alibaba/canal/releases/download/canal-1.0.22/canal.deployer-1.0.22.tar.gz。

这个就是server端，下载后解压，如图

打开conf/example下的instance.properties文件，编辑

################################################### mysql serverIdcanal.instance.mysql.slaveId = 1234 # position infocanal.instance.master.address = 127.0.0.1:3306canal.instance.master.journal.name =  canal.instance.master.position = canal.instance.master.timestamp =  #canal.instance.standby.address = #canal.instance.standby.journal.name =#canal.instance.standby.position = #canal.instance.standby.timestamp =  # username/passwordcanal.instance.dbUsername = canalcanal.instance.dbPassword = canalcanal.instance.defaultDatabaseName = testcanal.instance.connectionCharset = UTF-8 # table regexcanal.instance.filter.regex = .*\\..*# table black regexcanal.instance.filter.black.regex =   #################################################

address设置为mysql的连接地址，defaultDatabaseName设置为自己要监听的库名，如test。

在mysql命令行，创建一个新用户，作为slave

CREATE USER canal IDENTIFIED BY 'canal';  GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;FLUSH PRIVILEGES;

对应配置文件里的canal用户。到此配置完毕。

执行bin目录下的startup.sh

启动后可以在logs目录下查看日志。在example目录下的example.log，如果没有报错，说明启动成功。

canal客户端编写

服务端启动完毕后，在客户端即可监听test库的变化。

新建一个java maven项目，pom.xml里添加依赖

<dependency>      <groupId>com.alibaba.otter</groupId>      <artifactId>canal.client</artifactId>      <version>1.0.12</version>    </dependency>

package com.mindata; import com.alibaba.otter.canal.client.CanalConnector;import com.alibaba.otter.canal.client.CanalConnectors;import com.alibaba.otter.canal.common.utils.AddressUtils;import com.alibaba.otter.canal.protocol.CanalEntry;import com.alibaba.otter.canal.protocol.Message; import java.net.InetSocketAddress;import java.util.List; /** * A Camel Application */public class MainApp {     /**     * A main() so we can easily run these routing rules in our IDE     */    public static void main(String... args) throws Exception {         // 创建链接        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),                11111), "example", "", "");        int batchSize = 1000;        int emptyCount = 0;        try {            connector.connect();            connector.subscribe(".*\\..*");            connector.rollback();            int totalEmptyCount = 120;            while (emptyCount < totalEmptyCount) {                Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据                long batchId = message.getId();                int size = message.getEntries().size();                if (batchId == -1 || size == 0) {                    emptyCount++;                    System.out.println("empty count : " + emptyCount);                    try {                        Thread.sleep(1000);                    } catch (InterruptedException e) {                    }                } else {                    emptyCount = 0;                    // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size);                    printEntry(message.getEntries());                }                 connector.ack(batchId); // 提交确认                // connector.rollback(batchId); // 处理失败, 回滚数据            }             System.out.println("empty too many times, exit");        } finally {            connector.disconnect();        }    }     private static void printEntry(List<CanalEntry.Entry> entrys) {        for (CanalEntry.Entry entry : entrys) {            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry                    .EntryType                    .TRANSACTIONEND) {                continue;            }             CanalEntry.RowChange rowChage = null;            try {                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());            } catch (Exception e) {                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),                        e);            }             CanalEntry.EventType eventType = rowChage.getEventType();            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),                    eventType));             for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {                if (eventType == CanalEntry.EventType.DELETE) {                    printColumn(rowData.getBeforeColumnsList());                } else if (eventType == CanalEntry.EventType.INSERT) {                    printColumn(rowData.getAfterColumnsList());                } else {                    System.out.println("-------> before");                    printColumn(rowData.getBeforeColumnsList());                    System.out.println("-------> after");                    printColumn(rowData.getAfterColumnsList());                }            }        }    }     private static void printColumn(List<CanalEntry.Column> columns) {        for (CanalEntry.Column column : columns) {            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());        }    }}

注意newSingleConnector那里，设置canal server的ip和端口，端口默认为11111。example是和conf目录下的相对应。