canal

落日映苍穹つ 2022-05-14 05:28 234阅读 0赞

教程：  
[http://www.importnew.com/25189.html][http_www.importnew.com_25189.html]  
[https://blog.csdn.net/u012758088/article/details/78789616][https_blog.csdn.net_u012758088_article_details_78789616]

基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了mysql

canal装在/work/app/canal可以在bin目录下启动或关闭  
![这里写图片描述][70]

原理：  
\- canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议  
\- mysql master收到dump请求，开始推送binary log给slave(也就是canal)  
\- canal解析binary log对象(原始为byte流)  
![这里写图片描述][70 1]

canal采用protobuff格式：

Entry
        Header
            logfileName [binlog文件名]
            logfileOffset [binlog position]
            executeTime [发生的变更]
            schemaName 
            tableName
            eventType [insert/update/delete类型]
        entryType   [事务头BEGIN/事务尾END/数据ROWDATA]
        storeValue  [byte数据,可展开，对应的类型为RowChange]    
    RowChange
        isDdl       [是否是ddl变更操作，比如create table/drop table]
        sql     [具体的ddl sql]
        rowDatas    [具体insert/update/delete的变更数据，可为多条，1个binlog event事件可对应多条变更，比如批处理]
            beforeColumns [Column类型的数组]
            afterColumns [Column类型的数组]      
    Column 
        index       
        sqlType     [jdbc type]
        name        [column name]
        isKey       [是否为主键]
        updated     [是否发生过变更]
        isNull      [值是否为null]
        value       [具体的内容，注意为文本]

package com.kexin.canal;
    
    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 javax.validation.constraints.NotNull;
    import java.net.InetSocketAddress;
    import java.util.List;
    
    
    /**
     * @Author KeXin
     * @Date 2018/7/18 上午10:09
     **/
    /**
     * Created by hp on 14-7-17.
     */
    
    public class ClientSample {
    
        public static void main(String args[]) {
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(),
                    11111), "example", "", "");
            int batchSize = 1000;
            int emptyCount = 0;
            try {
                connector.connect();
    //            connector.subscribe(".*\\..*");
                connector.rollback();
                int totalEmtryCount = 1200;
                while (emptyCount < totalEmtryCount) {
                    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) {
                            e.printStackTrace();
                        }
                    } 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(@NotNull List<CanalEntry.Entry> entrys) {
            for (CanalEntry.Entry entry : entrys) {
                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("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                            e);
                }
    
                CanalEntry.EventType eventType = rowChange.getEventType();
                //根据binlog的filename和position来定位
                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 : rowChange.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(@NotNull List<CanalEntry.Column> columns) {
            for (CanalEntry.Column column : columns) {
                System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
            }
        }
    }

[http_www.importnew.com_25189.html]: http://www.importnew.com/25189.html
[https_blog.csdn.net_u012758088_article_details_78789616]: https://blog.csdn.net/u012758088/article/details/78789616
[70]: /images/20220514/61c6530d368f4e6a839e308bcbd0f248.png
[70 1]: /images/20220514/4518ea2bf47346809dbfc7eb8fc9cc20.png