因为自己项目中需要用到mysql数据同步到es中，查找了相关资料最后决定用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