canal demo搭建全记录

系统管理员 2021-11-09 14:22 318阅读 0赞

### 一、环境介绍 ###

canal是阿里开源的中间件，主要用于同步mysql数据库变更。具体参见：[https://github.com/alibaba/canal/releases][https_github.com_alibaba_canal_releases]

搭建环境：

vmware centos7 部署mysql和canal

windows开发canal client，自动捕获mysql数据库变更

### 二、Centos安装Mysql ###

**1、尝试用yum安装mysql**

wget http://repo.mysql.com/mysql57-community-release-el7-10.noarch.rpm

返回：2018-07-13 16:04:42 (63.9 KB/s) - ‘mysql57-community-release-el7-10.noarch.rpm’ saved \[25548/25548\]

sudo rpm -Uvh mysql57-community-release-el7-10.noarch.rpm
    sudo yum install -y mysql-community-server

如果执行顺利，安装mysql server 成功。

**2.改用阿里源安装**

可是官方的yum源在国内访问效果不佳，我下载mysql server的速度太慢了，决定改用阿里源

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

#下载wget
    yum install wget -y
    
    #备份当前的yum源
    mv /etc/yum.repos.d /etc/yum.repos.d.backup4comex
    
    #新建空的yum源设置目录
    mkdir /etc/yum.repos.d
    
    #下载阿里云的yum源配置
    wget -O /etc/yum.repos.d/CentOS-Base.repo http://mirrors.aliyun.com/repo/Centos-7.repo
    
    #最后重建缓存
    yum clean all
    yum makecache

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

**3.安装MariaDB**

MariaDB数据库管理系统是MySQL的一个分支，主要由开源社区在维护，采用GPL授权许可。开发这个分支的原因之一是：甲骨文公司收购了MySQL后，有将MySQL闭源的潜在风险，因此社区采用分支的方式来避开这个风险。MariaDB的目的是完全兼容MySQL，包括API和命令行，使之能轻松成为MySQL的代替品。

安装mariadb，大小59 M。

[root@yl-web yl]# yum install mariadb-server mariadb

其它几条常用的mariadb命令：

systemctl start mariadb  \#启动MariaDB

systemctl stop mariadb  \#停止MariaDB

systemctl restart mariadb  \#重启MariaDB

systemctl enable mariadb  \#设置开机启动

运行systemctl start mariadb，然后就可以正常使用mysql了

**4.设置数据库密码：**

set password for 'root'@'localhost' =password('root');

**5.遇到的几个问题**

①从windows访问centos mysql失败

解决方案：设置mysql允许远程连接

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

mysql -u root;
    //赋予任何主机访问数据的权限
    mysql>GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' WITH GRANT OPTION;
    
    //使修改生效
    mysql>FLUSH PRIVILEGES;

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

②进行上述操作之后，发现仍然连接失败，返回错误

Can't connect to MySQL server on '10.168.12.43' (10060)

解决方案：从windows连接vmware里面的mysql失败，关闭windows防火墙后成功。

### 三、部署canal server ###

（参考：[https://github.com/alibaba/canal/wiki/QuickStart][https_github.com_alibaba_canal_wiki_QuickStart]）

**1.下载canal server**

[https://github.com/alibaba/canal/releases][https_github.com_alibaba_canal_releases]

[![Image][]][Image 1]

我下载的是canal.exaple-1.0.24.gar.gz，下载完成后解压缩：

mkdir /tmp/canal
    tar zxvf canal.deployer-1.0.24.tar.gz  -C /tmp/canal

**2.查看binlog相关数据库命令：**

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

是否启用了日志
    mysql>show variables like 'log_bin';
    
    怎样知道当前的日志
    mysql> show master status;
    
    查看mysql binlog模式
    show variables like 'binlog_format';
    
    获取binlog文件列表
    show binary logs;
    
    查看当前正在写入的binlog文件
    show master status\G
    
    查看指定binlog文件的内容
    show binlog events in 'mysql-bin.000002';

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

**3.开启binlog**

如果log\_bin关闭，需要在etc下面找到my.cnf，开启binlog：

server-id=1
    log-bin=/var/lib/mysql/mysql-bin

然后重启mysql服务。

**4.添加canal mysql数据库账号**

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

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

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

**5.配置canal实例，设置本地数据库信息**

vi conf/example/instance.properties

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

## mysql serverId
    canal.instance.mysql.slaveId = 1234
    # position info
    canal.instance.master.address = 10.168.12.43:3306
    canal.instance.master.journal.name =mysql-bin.000003
    canal.instance.master.position =
    canal.instance.master.timestamp =
    
    ……
    
    canal.instance.dbUsername = canal
    canal.instance.dbPassword = canal
    canal.instance.defaultDatabaseName =testcanal
    canal.instance.connectionCharset = UTF-8
    
    # table regex
    canal.instance.filter.regex = .*\\..*

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

**6.启动canal**

sh bin/startup.sh

**7.查看日志**

vi logs/canal/canal.log
    vi logs/example/example.log

### 四、canal lient demo ###

**1.引入pom依赖**

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

**2.客户端代码**

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

public class ClientTest {
        public static void main(String args[]) {
            // 创建链接
            CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.168.12.43",
                    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;
                        printEntry(message.getEntries());
                    }
    
                    connector.ack(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().getLogfileName(), 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());
            }
        }
    }

![复制代码][aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]

**3.建立数据库连接，进行insert，delete等数据库操作**

[![Image(1)][Image_1]][Image_1_Image_1]

[![Image(2)][Image_2]][Image_2_Image_2]

### 五、遇到的问题 ###

1.canal建立连接失败

解决方案：用telnet命令测试建立连接仍然失败，关闭linux防火墙。

systemctl stop firewalld.service

其他centos7防火墙相关命令：

firewall-cmd --list-ports\#查看已经开放的端口：

firewall-cmd --reload \#重启firewall

systemctl stop firewalld.service \#停止firewall

systemctl disable firewalld.service \#禁止firewall开机启动

firewall-cmd --state \#查看默认防火墙状态（关闭后显示notrunning，开启后显示running）

来源：[https://www.cnblogs.com/janes/p/9318576.html][https_www.cnblogs.com_janes_p_9318576.html]

[https_github.com_alibaba_canal_releases]: https://github.com/alibaba/canal/releases
[aHR0cHM6Ly9jb21tb24uY25ibG9ncy5jb20vaW1hZ2VzL2NvcHljb2RlLmdpZg]: /images/20211109/44110c1344214d31a25ed8d974c24bde.png
[https_github.com_alibaba_canal_wiki_QuickStart]: https://github.com/alibaba/canal/wiki/QuickStart
[Image]: /images/20211109/cc98a43d7a2c45518c2949b86399cd97.png
[Image 1]: https://images2018.cnblogs.com/blog/37001/201807/37001-20180716165506844-661872852.png
[Image_1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWFnZXMyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMzcwMDEvMjAxODA3LzM3MDAxLTIwMTgwNzE2MTY1NTEzOTI3LTQyNDc2Njg1My5wbmc
[Image_1_Image_1]: /images/20211109/64adfa3b723241acbb030bf1808bba0a.png
[Image_2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9pbWFnZXMyMDE4LmNuYmxvZ3MuY29tL2Jsb2cvMzcwMDEvMjAxODA3LzM3MDAxLTIwMTgwNzE2MTY1NTE4ODUzLTEyODUyMDQ1NjgucG5n
[Image_2_Image_2]: /images/20211109/bab7d6abe2074fd5a0121a4b86cc5b0a.png
[https_www.cnblogs.com_janes_p_9318576.html]: https://www.cnblogs.com/janes/p/9318576.html