Canal说明

canal [kə’næl]，水道/管道/沟渠， 阿里巴巴 旗下的一款开源项目，纯Java开发。是基于数据库增量日志解析，提供增量数据订阅&消费，主要支持了MySQL（也支持mariaDB）。

MySQL主备复制原理

参考：

Canal工作原理：

依据于MySQL主备复制原理，相对比较简单：

1、canal模拟mysql slave的交互协议，伪装自己未mysql slave，向mysql master发送dump协议。

2、mysql master收到dump请求，开始推送binary log给slave（也就是canal）；

3、canal解析binary log对象（原始为 byte 流）。

Canal环境搭建

环境： windows10 ，mysql-5.7.13-winx64。

MySQL配置（配置文件my.ini中添加）：

 ## 设置server_id，同一局域网中需要唯一
server_id=101 
## 指定不需要同步的数据库名称
binlog-ignore-db=mysql  
## 开启 二进制 日志功能
log-bin=mall-mysql-bin  
## 设置二进制日志使用内存大小（事务）
binlog_cache_size=1M  
## 设置使用的二进制日志格式（mixed, statement ,row）
binlog_format=row  
## 二进制日志过期清理时间。默认值为0，表示不自动清理。
expire_logs_days=7  
## 跳过主从复制中遇到的所有错误或指定类型的错误，避免slave端复制中断。
## 如：1062错误是指一些主键重复，1032错误是因为主从数据库数据不一致
slave_skip_errors=1062   

重启服务后查询：

show variables like ‘%log_bin%’;

show variables like ‘binlog_format%’;

创建Cannl使用的账号：

 创建用户，用户名：canal 密码：canal@123456：
create user 'canal'@'%' identified by 'canal@123456';
授权 *.*表示所有库：
grant SELECT,REPLICATION SLAVE,REPLICATION CLIENT on *.* to 'canal'@'%' identified by 'canal@123456';  

Canal服务部署

canal-1.1.5，下载： canal.deployer-1.1.5.tar.gz ，

配置canal.deployer-1.1.5\conf\example\instance.properties：

 # 需要同步数据的MySQL地址
canal.instance.master.address= 127.0.0.1 :3306
# 用于同步数据的数据库账号
canal.instance.tsdb.dbUsername=canal
# 用于同步数据的数据库密码
#canal.instance.tsdb.dbPassword=canal@123456
# 数据库连接编码
canal.instance.connectionCharset = UTF-8
# 需要订阅binlog的表过滤正则表达式
canal.instance.filter.regex=.*\\..*  

启动：startup.bat，可在logs是下查看日志。

Canal客户端

Java， Maven 工程：

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

有效的验证代码：

 package com.what21.demo.mysql;

import java.net.InetSocketAddress;
import java.util.List;

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.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;

public class SimpleCanalClientExample {

    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 = 120;
            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 (Interrupted Exception  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<Entry> entrys) {
        for (Entry entry : entrys) {
            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                continue;
            }

            RowChange rowChage = null;
            try {
                rowChage = RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(),
                        e);
            }

            EventType eventType = rowChage.getEventType();
            System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                    entry.getHeader().getLog file Name(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (RowData rowData : rowChage.getRowDatasList()) {
                if (eventType == EventType.DELETE) {
                    printColumn(rowData.getBeforeColumnsList());
                } else if (eventType == 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<Column> columns) {
        for (Column column : columns) {
            System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
        }
    }
}  

这时候，修改mysql数据库中的表，增、删、改，这里就会有打印了。