首页 > 数据库> > Canal 原理说明和Mysql+Canal+kafaka 按装说明

Canal 原理说明和Mysql+Canal+kafaka 按装说明

2022-08-31 08:01:53 作者：互联网

Canal 原理说明：

主要应用场景 Mysql与Redis 可靠一致性，因为msyql修改变更将数据加到kafka队列可以确保存数据一定会被更新到redis，kafka有重试和可以确保被消费。使用阿里的canal将Mysql的binlog日志采集发送到MQ

【同步缓存redis/全文搜索ES】

【下发任务根据Biglog信息变更】

原理：

l MySQL主备复制原理

1) MySQL master 将数据变更写入二进制日志( binary log, 其中记录叫做二进制日志事件binary log events，可以通过 show binlog events 进行查看)

2) MySQL slave 将 master 的 binary log events 拷贝到它的中继日志(relay log)

注意：中继日志是从服务器I/O线程将主服务器的二进制日志读取过来，记录到从服务器本地文件，然后从服务器SQL线程会读取relay-log日志的内容并应用到从服务器，从而使从服务器和主服务器的数据保持一致。

3) MySQL slave 重放 relay log 中事件，将数据变更反映它自己的数据

l canal 工作原理

1) canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送dump 协议

2) MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )

3) canal 解析 binary log 对象(原始为 byte 流)

注意：mysql-binlog是MySQL数据库的二进制日志，记录了所有的DDL和DML(除了数据查询语句)语句信息。一般来说开启二进制日志大概会有1%的性能损耗。

Mysql+Canal+kafaka 按装说明

一、概述

1、按装Mysql和开启BingLog

2、按装 zookeeper

3、按装 kafka

4、按装 canal

5、进行简要测试

二、按装Mysql

2.1 Centos 7 按装Mysql 8 总有问题后来按装

下载地址：MySQL :: Download MySQL Community Server (Archived Versions)

mysql-community-server-5.7.38-1.el7.x86_64.rpm和mysql-community-client-5.7.38-1.el7.x86_64.rpm

2.2将下载的Mysql 上传到linux

1)新建目录：mkdir software 然后进入

2)rz 命令上传

3）安装mysql所需要的组件

y um install libaio(可选)

4）安装mysql Server

rpm -ivh mysql-community-server-5.7.32-1.el7.x86_64.rpm --nodeps --force

如果不加--nodeps --force

则提示需要安装客户端依赖等错误

5）查看是否按装

rpm -qa | grep mysql

6)启动mysql并查看状态

systemctl start mysqld.service -- 启动

systemctl status mysqld.service -- 查看状态

通过上面我们就安装好了MySQL的Server端，这时候是无法使用MySQL命令登录的，还需要安装一个Client端才行

7）装客户端：rpm -ivh mysql-community-client-5.7.32-1.el7.x86_64.rpm --nodeps --force

2.2 Mysql配置

2.2.1获取原始密码

grep "password" /var/log/mysqld.log

2.2.2 使用原始密码登录mysql服务器
mysql -uroot -p

2.2.3修改密码

ALTER USER 'root'@'localhost' IDENTIFIED BY 'new password'

2.2.3 开启mysql binlog日志

在/etc/my.cnf文件中[mysqld]下写入以下内容：

[mysqld]

# 随机指定一个不能和其他集群中机器重名的字符串，配置 MySQL replaction 需要定#义，不要和 canal 的 slaveId 重复

server-id=123

#配置binlog日志目录，配置后会自动开启binlog日志，并写入该目录

log-bin=/var/lib/mysql/mysql-bin

# 选择 ROW 模式

binlog-format=ROW

2.2.4、vi etc/my.cnf

修改mysql默认语言为utf8mb4

1、重启

修改完成后重启mysql
systemctl restart mysqld.service

　　按装mysql成功后

三、按装zookeeper

解压： tar -zxf zookeeper-3.4.6.tar.gz -C /usr/

进入：　cd　／usr/ zookeeper

复制：cp conf/zoo_sample.cfg conf/zoo.cfg

vi conf/zoo.cfg

只修改：dataDir=/root/zkdata

保存：wq

创建目录：mkdir /root/zkdata

启动zookeeper

cd /usr/zookeeper-3.4.6/

./bin/zkServer.sh

./bin/zkServer.sh start zoo.cfg

所有java进程： jps

查看zookeeper: ./bin/zkServer.sh status zoo.cfg

ps：按装在了/usr/local 目录下

二、kafka 按装

解压： tar -zxf kafka_2...........tgz -C /usr/

进入目录: cd /usr/kafka_w ....

查看目录： ls /bin

ls config/

配置信息：vi server.properties

打开监听： listeners=PLAINTEXT://localhost:9092

修改配置： log.dirs=/usr/kafka-logs

zookeeper.connect=localhost:2181

启动kafka： ./bin/kafka-server-start.sh -daemon config/server.properties

这个参数：-daemon 以后台形式启动

查看：jps --应该有kafka

需再打开一个 shell窗口

创建一个topic：./bin/kafka-topics.sh --help

./bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic topic01 --partitions 2 --replication-factor 1

消费： ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic01 --group group1

另一个shell窗口：

发送消息： ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic01 this is ma shibing

hello

一、Canal 按装

1、 Mysq 检查mysql配置

mysql> show variables like 'log_%';

检查msyql是否开启 binglog：# 编辑/etc/my.cnf

vim /etc/my.cnf

# 加入下面三行

log-bin=mysql-bin # 开启 binlog

binlog-format=ROW # 选择 ROW 模式

server_id=1 # 不能与 canal配值相关服务ID相同

# 编辑/etc/my.cnf

vim /etc/my.cnf

# 加入下面三行

log-bin=mysql-bin # 开启 binlog

binlog-format=ROW # 选择 ROW 模式

server_id=1

5.1、按装Canal

下载Canal

Cannal下载地址如下：https://github.com/alibaba/canal/releases,这里选择Canal 1.1.4版本下载。

#首先创建目录 “/software/canal”

[root@node3 ~]# mkdir -p /software/canal

#将Canal安装包解压到创建的canal目录中

[root@node3 ~]# tar -zxvf /software/canal.deployer-1.1.4.tar.gz -C /software/canal/

1、配置“canal.properties”

进入“/software/canal/conf”目录下，编辑“canal.properties”文件：

#canal将数据写入Kafka，可配：tcp, kafka, RocketMQ，tcp就是使用canal代码接收

canal.serverMode = kafka

#配置canal写入Kafka地址

canal.mq.servers = node1:9092,node2:9092,node3:9092

关于canal.properties更多参数参照：https://github.com/alibaba/canal/wiki/AdminGuide

2、配置mysql slave的权限

Canal的原理是模拟自己为mysql slave，所以这里一定需要做为mysql slave的相关权限，授权Canal连接MySQL具有作为MySQL slave的权限：

mysql> CREATE USER canal IDENTIFIED BY 'canal';

mysql> GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';

mysql> FLUSH PRIVILEGES;

mysql> show grants for 'canal' ;

3、配置“instance.properties”

进入“/software/canal/conf/example/”下，编辑“instance.properties”文件：

#canal伪装为一个mysql的salve，配置其id，不要和真正mysql server-id冲突，这里也可以不配置，会自动生成

canal.instance.mysql.slaveId=123456

#配置mysql master 节点及端口

canal.instance.master.address=127.0.0.1:3306

#配置连接mysql的用户名和密码，就是前面复制权限的用户名和密码

canal.instance.dbUsername=canal

canal.instance.dbPassword=canal

#配置Canal将数据导入到Kafka topic

canal.mq.topic=canal_topic

配置参照：

https://github.com/alibaba/canal/wiki/Canal-Kafka-RocketMQ-QuickStart

关于“instance.properties”更多参数介绍如下：https://github.com/alibaba/canal/wiki/AdminGuide#instanceproperties%E4%BB%8B%E7%BB%8D

4、启动Canal

进入“/software/canal/bin”,执行“startup.sh”脚本启动Canal。

#启动Canal

[root@node3 ~]# cd /software/canal/bin/

[root@node3 bin]# ./startup.sh

[root@node3 bin]# jps

68675 CanalLauncher #启动成功

如果启动失败可查看 canla/log 相关日志vi logs/canal/canal.log</pre>

打开编辑canal 启动脚本:startup.sh,【查看失败原因如果是jdk版本问题错误】

改成

str=`file -L $JAVA | grep 64-bit`

if [ -n "$str" ];

then

JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -Xss256k"

else

JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m "

sh bin/stop.sh

5、启动zookeeper和Kafka，并监控Kafka中“canal_topic”的数据

注意：“canal_topic”不需要提前创建，默认创建就是1个分区。

[root@node2 bin]# ./kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic canal_topic

6、在MySQL中建表，插入语句

mysql> create database testdb;

mysql> use testdb;

mysql> create table person(id int ,name varchar(255),age int);

mysql> insert into person values (1,"zs",18),(2,"ls",19),(3,"ww",20);

#对应的在Kafka中有对应的数据日志写入

概要说明：