【Canal】之 搭建环境 和 客户端运行

前言

在搭建时候，发现几个易错点（坑点），并一一填过 1. mysql 版本尽量5.7以下（官方有支持的版本） 2. canal运行方式

一、准备

---

平台：Ubuntu 16.04 MySQL： 5.7.22

MySQL 安装（指定版本）

# 指定版本：apt-get install mysql-client-core-* （* 为 版本号） apt-get install mysql-client-core-5.7 apt-get install mysql-client-5.7 apt-get install mysql-server-5.7

MySQL 相关命令

# 服务开启 service mysql start # 服务关闭 service mysql stop # 登录 mysql -uxxx -pyyy （xxx 为账号， yyy为密码）

MySQL 对应目录

数据库目录：/var/lib/mysql配置文件: /usr/share/mysql , /etc/mysql相关命令: /usr/bin , /usr/sbin启动脚本: etc/init.d/mysql

二、配置MySQL

---

（1）在/etc/mysql/my.cnf 中添加

[mysqld] log-bin=mysql-bin #添加这一行就ok binlog-format=ROW #选择row模式 server_id=1 #配置mysql replaction需要定义，不能和canal的slaveId重复

（2）在mysql中配置canal数据库管理用户，配置相应权限

CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES;

三、搭建Canal

---

（1）下载

wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.deployer-1.0.24.tar.gz

（2）解压

在 /tmp 目录下的文件在重启后会被自动删除

mkdir /tmp/canal tar zxvf canal.deployer-$version.tar.gz -C /tmp/canal

（3）修改配置

sudo vi /tmp/canal/conf/example/instance.properties

（4）启动

./bin/startup.sh

（5）查看日志

vi /tmp/canal/logs/canal/canal.log

vi /tmp/canal/logs/example/example.log

（6）关闭

./bin/stop.sh

四、搭建canal-client

---

Maven 配置

<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.0.12</version> </dependency>

CanalDemo.java

package com.donaldy.platform.canalyoung.demo; import java.net.InetSocketAddress; import java.util.List; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.client.CanalConnector; 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 CanalDemo { public static void main(String args[]) { // 创建链接 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "", ""); int batchSize = 1000; int emptyCount = 0; try { connector.connect(); connector.subscribe(".*\\..*"); connector.rollback(); int totalEmptyCount = 120; while (emptyCount < totalEmptyCount) { Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { emptyCount++; System.out.println("empty count : " + emptyCount); try { Thread.sleep(1000); } catch (InterruptedException e) { } } else { emptyCount = 0; // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size); printEntry(message.getEntries()); } connector.ack(batchId); // 提交确认 // connector.rollback(batchId); // 处理失败, 回滚数据 } System.out.println("empty too many times, exit"); } finally { connector.disconnect(); } } private static void printEntry(List<Entry> entrys) { for (Entry entry : entrys) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { continue; } RowChange rowChage = null; try { rowChage = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } EventType eventType = rowChage.getEventType(); System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType)); for (RowData rowData : rowChage.getRowDatasList()) { if (eventType == EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); } else if (eventType == EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); } else { System.out.println("-------> before"); printColumn(rowData.getBeforeColumnsList()); System.out.println("-------> after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List<Column> columns) { for (Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } }

运行

首先启动Canal Server，即之前的 ./bin/startup.sh

创建数据库，表，插入记录

# 创建数据库 create database test; use test; # 创建表 create table test ( uid int (4) primary key not null auto_increment, name varchar(10) not null); # 插入记录 insert into test (name) values('10');

五、参考资料

---

https://github.com/alibaba/canal/wiki/QuickStart https://github.com/alibaba/canal/wiki/ClientExample https://blog.csdn.net/hackerwin7/article/details/37923607 https://blog.csdn.net/zjerryj/article/details/77152226