canal 客户端api 使用示例

迷南。 2022-10-11 13:35 417阅读 0赞

--------------------

## canal 客户端api 使用示例 ##

## \*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\* ##

**示例**

**CanalTest**

public class CanalTest {
    
        public static void printEntry(List<CanalEntry.Entry> entries){
            for (CanalEntry.Entry entry:entries){
                if (entry.getEntryType()== CanalEntry.EntryType.TRANSACTIONBEGIN
                        || entry.getEntryType()== CanalEntry.EntryType.TRANSACTIONEND){
                    continue;
                }
    
                CanalEntry.RowChange rowChange=null;
                try {
                    rowChange= CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                }catch (Exception e){
                    throw new RuntimeException(e.getMessage());
                }
    
                CanalEntry.EventType eventType=rowChange.getEventType();
                System.out.println("事件类型："+eventType);
                System.out.println("数据库名："+entry.getHeader().getSchemaName());
                System.out.println("表名："+entry.getHeader().getTableName());
                System.out.println("binlog name："+entry.getHeader().getLogfileName());
                System.out.println("binlog position："+entry.getHeader().getLogfileOffset()+"\n");
    
                for (CanalEntry.RowData data:rowChange.getRowDatasList()){
                    if (eventType == CanalEntry.EventType.DELETE){
                        System.out.println(entry.getHeader().getSchemaName()+"."+entry.getHeader().getTableName()+"删除前");
                        printColumns(data.getAfterColumnsList());
                    }else if (eventType == CanalEntry.EventType.INSERT){
                        System.out.println(entry.getHeader().getSchemaName()+"."+entry.getHeader().getTableName()+"插入后");
                        printColumns(data.getAfterColumnsList());
                    }else if (eventType == CanalEntry.EventType.UPDATE){
                        System.out.println(entry.getHeader().getSchemaName()+"."+entry.getHeader().getTableName()+"更新前：");
                        printColumns(data.getBeforeColumnsList());
    
                        System.out.println(entry.getHeader().getSchemaName()+"."+entry.getHeader().getTableName()+"更新后：");
                        printColumns(data.getAfterColumnsList());
                    }
                }
            }
        }
    
        public static void printColumns(List<CanalEntry.Column> columns){
            for (CanalEntry.Column column:columns){
                System.out.print("column index："+column.getIndex()+"、");
                System.out.print("column name："+column.getName()+"、");
                System.out.print("column value："+column.getValue()+"、");
                System.out.println("column update："+column.getUpdated());
            }
    
            System.out.println();
        }
    
        public static void main(String[] args){
            CanalConnector connector= CanalConnectors.newSingleConnector(
                    new InetSocketAddress("192.168.57.120",11111),
                    "example","","");
    
            int batchSize=100;
            int emptyCount=0;
            try {
                connector.connect();
                connector.subscribe("");
    
                int totalEmptyCount=100;
                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("emptyCount："+emptyCount);
    
                        try {
                            Thread.sleep(1000);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }else {
                        emptyCount=0;
    
                        printEntry(message.getEntries());
                    }
                }
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                connector.disconnect();
            }
        }
    }

**connector.subscribe说明**

connector.subscribe("")
    若客户端为初次启动，默认监听所有的数据库、表，等同于：.*\\..*
    若为再次启动，则监听上次监听的数据库、表
    
    
    connector.subscribe("database_name\\.table_name")：监听指定的数据库、表
    db.*\\.table.*：监听db开头的数据库下table开头的表
    
    
    # 查看客户端监听的数据库、表
    [root@895debf75f5a example]# cat example.log
    2021-07-04 17:25:34.624 [main] INFO  c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example 
    2021-07-04 17:25:34.714 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^.*\..*$
    2021-07-04 17:25:34.714 [main] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table black filter : ^mysql\.slave_.*$
    2021-07-04 17:25:34.754 [main] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - start successful....
    2021-07-04 17:25:35.352 [destination = example , address = /172.18.0.2:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> begin to find start position, it will be long time for reset or first position
    2021-07-04 17:25:35.372 [destination = example , address = /172.18.0.2:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - prepare to find start position just show master status
    2021-07-04 17:25:40.468 [destination = example , address = /172.18.0.2:3306 , EventParser] WARN  c.a.o.c.p.inbound.mysql.rds.RdsBinlogEventParserProxy - ---> find start position successfully, EntryPosition[included=false,journalName=binlog.000002,position=4,serverId=1,gtid=<null>,timestamp=1625390690000] cost : 5065ms , the next step is binlog dump
    
    # 监听的数据库、表转换为：lihu2\..*
    2021-07-04 17:43:42.227 [New I/O server worker #1-2] INFO  c.a.otter.canal.instance.core.AbstractCanalInstance - subscribe filter change to lihu2\..*
    2021-07-04 17:43:42.228 [New I/O server worker #1-2] WARN  c.a.o.canal.parse.inbound.mysql.dbsync.LogEventConvert - --> init table filter : ^lihu2\..*$

**\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\*\***

**使用测试**

**创建 mysql**

docker run -it -d --net fixed --ip 172.18.0.2 -p 3306:3306 --privileged=true \
    --name mysql -e  MYSQL_ROOT_PASSWORD=123456 mysql
    
    
    # 创建用户、并授权
    mysql> create user canal identified with mysql_native_password by "123456";
    Query OK, 0 rows affected (0.00 sec)
    
    mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
    Query OK, 0 rows affected (0.00 sec)
    
    mysql> flush privileges;
    Query OK, 0 rows affected (0.00 sec)

**创建 canal实例**

docker run -it -d --net fixed --ip 172.18.0.3 \
    -p 11111:11111 --name canal-server \
    -e canal.instance.master.address=172.18.0.2:3306  \
    -e canal.instance.dbUsername=canal \
    -e canal.instance.dbPassword=123456 canal/canal-server

**数据库操作**

mysql> create database lihu;
    Query OK, 1 row affected (0.04 sec)
    
    mysql> create database lihu2;
    Query OK, 1 row affected (0.04 sec)
    
    mysql> use lihu;
    Database changed
    mysql> create table test(id int not null primary key auto_increment,value int not null);
    Query OK, 0 rows affected (0.03 sec)
    
    mysql> use lihu2
    Database changed
    mysql> create table test2(id int not null primary key auto_increment,value int not null);
    Query OK, 0 rows affected (0.03 sec)
    
    mysql> use lihu
    Reading table information for completion of table and column names
    You can turn off this feature to get a quicker startup with -A
    
    Database changed
    mysql> insert into test(id,value) values(1,2),(2,6);
    Query OK, 2 rows affected (0.01 sec)
    Records: 2  Duplicates: 0  Warnings: 0
    
    mysql> use lihu2;
    Reading table information for completion of table and column names
    You can turn off this feature to get a quicker startup with -A
    
    Database changed
    mysql> insert into test2(id,value) values(1,2),(2,6),(3,8);
    Query OK, 3 rows affected (0.00 sec)
    Records: 3  Duplicates: 0  Warnings: 0

**控制台输出**

事件类型：QUERY
    数据库名：lihu
    表名：
    binlog name：binlog.000002
    binlog position：933
    
    事件类型：QUERY
    数据库名：lihu2
    表名：
    binlog name：binlog.000002
    binlog position：1118
    
    事件类型：CREATE
    数据库名：lihu
    表名：test
    binlog name：binlog.000002
    binlog position：1306
    
    事件类型：CREATE
    数据库名：lihu2
    表名：test2
    binlog name：binlog.000002
    binlog position：1551
    
    事件类型：INSERT
    数据库名：lihu
    表名：test
    binlog name：binlog.000002
    binlog position：1927
    
    lihu.test插入后
    column index：0、column name：id、column value：1、column update：true
    column index：1、column name：value、column value：2、column update：true
    
    lihu.test插入后
    column index：0、column name：id、column value：2、column update：true
    column index：1、column name：value、column value：6、column update：true
    
    事件类型：INSERT
    数据库名：lihu2
    表名：test2
    binlog name：binlog.000002
    binlog position：2219
    
    lihu2.test2插入后
    column index：0、column name：id、column value：1、column update：true
    column index：1、column name：value、column value：2、column update：true
    
    lihu2.test2插入后
    column index：0、column name：id、column value：2、column update：true
    column index：1、column name：value、column value：6、column update：true
    
    lihu2.test2插入后
    column index：0、column name：id、column value：3、column update：true
    column index：1、column name：value、column value：8、column update：true