因為自己項目中需要用到mysql數據同步到es中，查找了相關資料最后決定用canal來做，所以便有了本文，下面一起來看如何使用canal吧

canal教程

根據 https://github.com/alibaba/canal 上的原理解釋，我們知道 canal 會模擬 mysql slave 的交互協議，偽裝自己為 mysql slave，然后向 mysql master 發送 dump 協議。

mysql master 收到 dump 請求，開始推送 binary log 給 slave(也就是 canal)，然后 canal 解析 binary log 對象(原始為 byte流)。

經 canal 解析過的對象，我們使用起來就非常的方便了。

再根據 https://github.com/alibaba/canal/releases 提供的版本信息，你會發現 canal 其實相當于一個中間件，專門用來解析 MySQL 的 binlog 日志。canal 解析好了之后，會封裝成一個數據對象，通過 protobuf3.0 協議進行交互，讓 canal 客戶端進行消費。

根據上面的解釋，以及 canal 提供的版本信息，我們在使用 canal 的時候，首選要安裝一個 canal.deployer-1.1.4.tar.gz 進行解析 MySQL 的 binlog 日志。

下載后，復制 canal.deployer-1.1.4.tar.gz 到 MySQL 主機上，比如放在 /usr/local/soft/目錄下。然后依次執行下面的命令：

mkdir canalcd canaltar -zxvf ../canal.deployer-1.1.4-SNAPSHOT.tar.gz

然后修改 canal 的配置文件 vim conf/example/instance.properties

這三項改成你自己的，比如我的配置如下：

canal.instance.dbUsername=canalcanal.instance.dbPassword=canalcanal.instance.connectionCharset = UTF-8canal.instance.defaultDatabaseName =canal_test

然后保存并退出。(VI 模式下，按 Esc 輸入 :wq 回車退出。)

接著，我們檢查一下 MySQL 的配置。確定版本和是否開啟了 binlog 日志，以及日志格式。

show variables like 'binlog_format';show variables like 'log_bin';select version();

canal 支持 binlog 格式為 ROW 的模式。如果你沒開啟 binlog，并且格式是非 row 的，建議修改一下 mysql 的配置文件。

執行 mysql –help | grep my.cnf 找到 mysql 的 my.cnf 文件。

執行 vim /etc/my.cnf 命令。添加下面 3 個配置。

log-bin=mysql-bin #添加這一行就okbinlog-format=ROW #選擇row模式server_id=1 #配置mysql replaction需要定義，不能和canal的slaveId重復

然后保存并退出。

接著執行 sudo service mysqld restart 重啟 MySQL。

需要注意的是你的 mysql 用戶，必須要有 REPLICATION SLAVE 權限。該權限授予 slave 服務器以該賬戶連接 master 后可以執行 replicate 操作的權利。

如果沒有權限，則使用 root 賬戶登錄進 MySQL，執行下面的語句，創建用戶，分配權限。

CREATE USER canal IDENTIFIED BY ‘canal’;GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON . TO ‘‘canal’’@’%’;FLUSH PRIVILEGES;

MySQL 啟動后，就可以開啟 canal 服務了。

/usr/local/soft/canal/bin/startup.sh

開啟后，觀察 canal 服務的日志，確保服務正常。

tail 300f /usr/local/soft/canal/logs/canal/canal.log

查看 canal 的日志

確定沒有問題后，開始編寫我們的測試程序。

pom.xml 中導入下面的依賴。

com.alibaba.otter canal.client 1.1.4

使用JAVA進行測試

import java.net.InetSocketAddress;import java.util.List;import com.alibaba.otter.canal.client.CanalConnectors;import com.alibaba.otter.canal.client.CanalConnector;import com.alibaba.otter.canal.common.utils.AddressUtils;import com.alibaba.otter.canal.protocol.Message;import com.alibaba.otter.canal.protocol.CanalEntry.Column;import com.alibaba.otter.canal.protocol.CanalEntry.Entry;import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;import com.alibaba.otter.canal.protocol.CanalEntry.EventType;import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;import com.alibaba.otter.canal.protocol.CanalEntry.RowData;public class SimpleCanalClientExample {public static void main(String args[]) { // 創建鏈接 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example

總結

以上是生活随笔為你收集整理的es 全量同步mysql_使用canal将mysql同步到es中的全部內容，希望文章能夠幫你解決所遇到的問題。

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