為了實現 實時同步數據，在mac環境搭建了canal，mysql，kafka的一套流程

使用canal加mysql加kafka的方式傳遞數據

mysql 數據源頭

canal模仿slave沖mysql取數據。。是一個管道

kafka 將canal獲取的數據放入kafka 。然后消費(程序獲取kafka的隊列。消費數據)

1.mysql安裝

這個就不詳述了。。不要安裝最新版的mysql。本人親測8.0版本和canal不一定能很好的兼容

所以。。安裝了mysql5.7

安裝命令(這樣安裝 安裝的是8.0版本)

brew install mysql

2.java安裝

不管是canal還是kafka都是需要java環境的。。

也不能用最新版的java，推薦使用java8也就是jdk8

最好是從官網oracle下載

安裝方法：

https://blog.csdn.net/oumuv/article/details/84064169

3.canal安裝

在裝calal之前 確保mysql 及 java8 環境安裝好了

下載地址： https://github.com/alibaba/canal/releases

下載 canal.deployer-1.1.4.tar.gz

在家目錄下面新建一個canal目錄.直接解壓就行

官方文件已經編譯好了，不需再編譯

需要添加賬戶(canal@%及canal@local)

在canal/bin目錄下有幾個腳本文件，startup.sh 啟動服務用的，stop.sh 停止服務用的

在canal/logs目錄下放的是日志文件。

在canal/conf目錄下放的是配置文件。

實例配置文件

canal/conf/example/instance.properties

實例的配置文件。。決定了 你連接哪個實例的數據庫 可以精確到表

instance.properties中比較重要的參數

#實例

canal.instance.master.address = 127.0.0.1:3306

#db

canal.instance.dbUsername = canal

canal.instance.dbPassword = canal

canal.instance.defaultDatabaseName =test

canal.instance.connectionCharset = UTF-8

#table

# table regex

canal.instance.filter.regex = test.ttt

#mq topic

canal.mq.topic=test12345

上面這樣配置 就代表了。。通過canal連接127.0.0.1中的test庫的ttt表，放到一個叫做test12345的topic里面

驗證canal是否和mysql連接 只需要在mysql的進程里面查看是否有一個復制連接(因為canal就是模仿了一個slave)

還有一個全局的文件

表示了需要連接的kafka zookeeper等

canal/conf/canal.properties

列舉canal連接kafka的重要參數

canal.id= 1#如果有多個canal 這個值和集群中的canal不能沖突

canal.ip=172.17.61.113#canal的ip

canal.port= 11111

canal.zkServers=172.17.61.113:2181#zookeeper的ip:port

canal.serverMode = kafka

canal.destinations = example

#mq

canal.mq.servers = 172.17.61.113:9092#kafka的ip:port

上面有些關于kafka的參數 是 需要kafka安裝好之后才能陪得(因為已經裝好了 所以先列舉出來)

4.kafka安裝

kafka安裝時需要zookeeper的。。

但是一般zookeeper會集成在kafka里面 所以不需要特別單獨安裝(也可以單獨安裝)

mac環境安裝命令如下

brew install kafka

一直等安裝完成就行

運行kafka是需要依賴于zookeeper的，所以安裝kafka的時候也會包含zookeeper。

kafka的安裝目錄：/usr/local/Cellar/kafka/2.0.0/bin

kafka的配置文件目錄：/usr/local/Cellar/kafka/2.0.0/bin

kafka服務的配置文件：/usr/local/etc/kafka/server.properties

zookeeper配置文件： /usr/local/etc/kafka/zookeeper.properties

# server.properties中的重要配置

需要修改的參數listeners=PLAINTEXT://172.17.61.113:9092advertised.listeners=PLAINTEXT://172.17.61.113:9092

zookeeper 配置文件我在安裝的時候 沒有做任何修改

然后 通過上面配置canal.properties 就連接上kafka

kafka的基本命令

首先，啟動zookeeper：

zookeeper-server- start /usr/ local /etc/kafka/zookeeper.properties

然后，啟動kafka

kafka- server -start /usr/local/etc/kafka/ server .properties

創建一個“使用單個分區”、“只有一個副本”、名為“test”的主題的topic

kafka-topics

--create

--zookeeper

localhost

:2181

--replication-factor 1

--partitions 1

--topic

test

(創建topic，這個topic。我們在canal中已經有了配置)

查看創建的topic，運行list topic命令：

kafka-topics --list --zookeeper 172.17.61.113:2181

生產消息

kafka-console-producer --broker-list 172.17.61.113:9092 --topic test

(消息數據由canal自動發送。所以這里的命令我們只有明白就行)

消費消息(這個地方就可以最后來查看mysql+canal+kafka是否已經聯動起來了)

kafka-console-consumer --bootstrap-server 172.17.61.113:9092 --topic

test12345 --from-beginning

我在mysql里面的test庫的ttt表里加了一條記錄

然后反應在kafka就是如下結果

{"data":[{"id":"13","var":"ded"}],"database":"test","es":1575448571000,"id":8,"isDdl":false,"mysqlType":{"id":"int(11)","var":"varchar(5)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"var":12},"table":"ttt","ts":1575448571758,"type":"INSERT"}

這就表示整個mysql+canal+kafka已經成功了

總結

以上是生活随笔為你收集整理的kafka mysql安装与配置_Mac环境canal+mysql+kafka的安装及使用的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。