canal與kafka的結合使用

centos7中安裝zk: https://www.cnblogs.com/tdyang/p/13895839.html

centos7中安裝kafka:  https://www.cnblogs.com/tdyang/p/13898004.html

第一步、下載安裝canal

本安裝目錄：/usr/local/soft,創建一個canal目錄，版本1.1.4

cd /usr/local/soft/
mkdir canal
cd canal

 下載解壓，這個路徑可能會有所變化，canal的github地址：https://github.com/alibaba/canal，如果下載速度慢，建議用迅雷下載

迅雷下載這些文件速度還是不錯

wget https://github.com/alibaba/canal/releases/download/canal-1.1.4/canal.deployer-1.1.4.tar.gz
tar -zxvf canal.deployer-1.1.4.tar.gz

 

第二步，修改canal的配置文件

1、conf下的canal.properties文件，修改以下兩項：建議用xftp連接打開文件,快速找到需要修改的配置。

canal.serverMode=kafka

 

kafka的服務器地址：

canal.mq.servers = 192.168.146.102:9092

 

 

2、修改example/instance.properties,mysql的數據庫連接地址

canal.instance.master.address=192.168.146.102:3306

 

 canal的用戶和密碼，這里在mysql數據庫里面設置

canal.instance.dbUsername=canal
canal.instance.dbPassword=123456

# 接着上面，新增一個配置
canal.instance.defaultDatabaseName=canaltest

# 這個topic會自動創建
canal.mq.topic=canal-topic

 

 mq的配置中還有其他的一些設置可自己根據需要設置

通過以上配置，還需要開啟的mysql的binlog配置

 

第三步：修改mysql的配置文件

log-bin=/var/lib/mysql/mysql-bin
binlog-format=ROW

 登錄數據庫以后，查看是否開啟了binlog， ON表示開啟了

show variables like 'log_%';

 

設置和上面instance.properties配置文件中canal的用戶名和密碼：

-- 創建canal用戶
CREATE USER canal IDENTIFIED BY '123456';

-- 給canal用戶分配查詢和復制的權限
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal@'%';

刷新權限：

FLUSH PRIVILEGES;

創建測試數據庫：

ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY '123456';

CREATE DATABASE `canaltest` CHARSET `utf8mb4` COLLATE `utf8mb4_unicode_ci`;

 

第四步：啟動各個服務

1、啟動zk，在zk的bin目錄下

./zkServer.sh start

2、啟動kafka，在kafka的安裝目錄下，后台啟動

nohup ./bin/kafka-server-start.sh ./config/server.properties & 

啟動kafka消費端：這個topic和上面example/instance.properties配置文件里面一致

./kafka-console-consumer.sh --bootstrap-server 192.168.146.102:9092 --topic canal-topic

3、啟動canal，在bin目錄下

sh startup.sh 

4、修改測試數據庫數據，看kafka消費端是否收到數據

測試：在一張dept表中加入一條數據

 kafka消費端收到消息：

 

 

 這樣一個簡單的測試就完成了