一. MySQL的准备
MySQL版本 5.7.31
1.1 binlog格式
vim /etc/my.cnf
server-id=1
log-bin=mysql-bin
binlog_format=row
binlog-do-db=canal_test
重启mysql服务
测试记录:
mysql> show variables like '%log_bin%';
+---------------------------------+-------------------------------------+
| Variable_name | Value |
+---------------------------------+-------------------------------------+
| log_bin | ON |
| log_bin_basename | /home/mysql/data/3306/hp8-bin |
| log_bin_index | /home/mysql/data/3306/hp8-bin.index |
| log_bin_trust_function_creators | OFF |
| log_bin_use_v1_row_events | OFF |
| sql_log_bin | ON |
+---------------------------------+-------------------------------------+
6 rows in set (0.01 sec)
mysql> show variables like '%binlog_format%';
+---------------+-------+
| Variable_name | Value |
+---------------+-------+
| binlog_format | ROW |
+---------------+-------+
1 row in set (0.00 sec)
mysql>
1.2 创建库表
代码:
create database canal_test default character set utf8;
show create database canal_test\G
use canal_test;
CREATE TABLE user_info(
`id` VARCHAR(255),
`name` VARCHAR(255),
`sex` VARCHAR(255)
);
测试记录:
mysql> create database canal_test default character set utf8;
Query OK, 1 row affected (0.00 sec)
mysql> show create database canal_test\G
*************************** 1. row ***************************
Database: canal_test
Create Database: CREATE DATABASE `canal_test` /*!40100 DEFAULT CHARACTER SET utf8 */
1 row in set (0.00 sec)
mysql>
mysql> use canal_test;
Database changed
mysql> CREATE TABLE user_info(
-> `id` VARCHAR(255),
-> `name` VARCHAR(255),
-> `sex` VARCHAR(255)
-> );
Query OK, 0 rows affected (0.01 sec)
mysql> show create table user_info\G
*************************** 1. row ***************************
Table: user_info
Create Table: CREATE TABLE `user_info` (
`id` varchar(255) DEFAULT NULL,
`name` varchar(255) DEFAULT NULL,
`sex` varchar(255) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8
1 row in set (0.00 sec)
mysql>
1.3 赋权限
set global validate_password_length=4;
set global validate_password_policy=0;
GRANT SELECT,REPLICATION SLAVE,REPLICATION CLIENT ON *.* TO
'canal'@'%' IDENTIFIED BY 'canal' ;
二. Canal安装及配置
2.1 Canal下载及安装
下载地址:
https://github.com/alibaba/canal/releases
cd /usr/local/src
gunzip canal.deployer-1.1.2.tar.gz
tar -xvf canal.deployer-1.1.2.tar
2.2 修改 canal.properties 的配置
cd /usr/local/src/conf
vim canal.properties
canal.id = 1
canal.ip =
canal.port = 11111
canal.metrics.pull.port = 11112
canal.zkServers =
# flush data to zk canal.zookeeper.flush.period = 1000
canal.withoutNetty = false
# tcp, kafka, RocketMQ
canal.serverMode = kafka
# flush meta cursor/parse position to file
canal.mq.servers = hp2:9092,hp3:9092,hp4:9092
说明：
这个文件是 canal 的基本通用配置，canal 端口号默认就是 11111，修改 canal 的 输出 model，默认 tcp，改为输出到 kafka多实例配置如果创建多个实例，通过前面 canal 架构，我们可以知道，一个 canal 服务 中可以有多个 instance，conf/下的每一个 example 即是一个实例，每个实例下面都有独立的 配置文件。默认只有一个实例 example，如果需要多个实例处理不同的 MySQL 数据的话，直 接拷贝出多个 example，并对其重新命名，命名和配置文件中指定的名称一致，然后修改 canal.properties 中的 canal.destinations=实例 1，实例 2，实例 3。
2.3 修改 instance.properties
我们这里只读取一个 MySQL 数据，所以只有一个实例，这个实例的配置文件在conf/example 目录下
cd /usr/local/src/conf/example
vim instance.properties
canal.instance.master.address=127.0.0.1:3306
canal.instance.dbUsername=canal
canal.instance.dbPassword=canal
canal.instance.connectionCharset = UTF-8
canal.instance.defaultDatabaseName =canal_test
# mq config
canal.mq.topic=canal_test
canal.mq.partitionsNum=2
canal.mq.partition=2
# hash partition config
#canal.mq.partitionHash=mytest.person:id,mytest.role:id
备注：
canal.mq.partitionsNum 不能操作kafka主题的分区数，不然会报错
ERROR com.alibaba.otter.canal.kafka.CanalKafkaProducer - Invalid partition given with record: 3 is not in the range [0...1).
kafka默认的分区数是1，此时需要修改
kafka-topics.sh --zookeeper localhost:2181 --alter --partitions 3 --topic canal_test
2.4 启动 Canal
cd /usr/local/src/
bin/startup.sh
image.png
2.5 看到 CanalLauncher 你表示启动成功，同时会创建 canal_test 主题
[root@hp8 ~]# jps
11498 BrokerBootstrap
15662 Jps
[root@hp8 ~]#
2.6 启动 Kafka 消费客户端测试，查看消费情况
cd /opt/cloudera/parcels/CDH-6.3.1-1.cdh6.3.1.p0.1470567/lib/kafka/bin
./kafka-console-consumer.sh --bootstrap-server hp2:9092 --topic canal_test
2.7 向 MySQL 中插入数据后查看消费者控制台
MySQL
INSERT INTO user_info VALUES('1001','zhangsan','male');
INSERT INTO user_info VALUES('1002','lishi','male');
Kafka
{"data":[{"id":"1002","name":"lishi","sex":"male"}],"database":"canal_test","es":1646041639000,"id":1,"isDdl":false,"mysqlType":{"id":"varchar(255)","name":"varchar(255)","sex":"varchar(255)"},"old":null,"sql":"","sqlType":{"id":12,"name":12,"sex":12},"table":"user_info","ts":1646042284004,"type":"INSERT"}
image.png
网友评论