Canal 最初只支持将数据从 MySQL 同步到 Kafka，RabbitMQ 等消息队列中，从 1.1.1 版本开始，Canal 实现了一个配套落地的模块 Canal Adapter，实现对 Canal Server 订阅的 binlog 消息进行消费，支持将数据输出至 HBase，MySQL，Elasticsearch等。
机器规划
| IP地址 | 服务 |
|---|---|
| 192.168.1.151:3306 | canal元数据库 |
| 192.168.1.154:3306 | 源数据库 |
| 192.168.1.151:3306 | 目标数据库 |
| 192.168.1.156 | canal、canal server、canal adapter、zookeeper |
| 192.168.1.155 | canal、canal server、canal adapter、canal admin |
准备工作
1、源库需要binlog以及binlog模式为ROW
2、部署zookeeper集群（192.168.1.156）
这里因机器限制，暂时部署zookeeper单节点，版本这里采用zookeeper-3.4.10。
tar -zxvf zookeeper-3.4.10.tar.gz -C /opt/
mv /opt/zookeeper-3.4.10 /opt/zookeeper
mkdir /opt/zookeeper/{data,logs}
cd /opt/zookeeper/conf/
cp zoo_sample.cfg zoo.cfg
编辑zoo.cfg内容
tickTime：ZooKeeper使用的基本时间单位（毫秒）。 它用于做心跳，并且最小会话（session）超时将是tickTime的两倍。
initLimit：用于限制ZooKeeper服务器必须连接到Leader服务器的时间长度，有必要可以设置在一点点的时间长度。
syncLimit：服务器与领导者之间过时的距离。可以通过心跳检测机制，来检测机器的存活状态。如果Leader发出心跳包。在syncLimit之后，还没有从Folloer那里收到响应，那么就认为这个F已经不在线了.所以这个参数不移设置过大。
dataDir：配置存储内存数据库快照的位置以及数据库更新的事务日志。
clientPort：侦听客户端连接的端口
创建myid，注明上面zoo.cfg中的server.x中的x
echo "1" > ../data/myid
配置环境变量
vim /etc/profile
#添加以下代码
export ZOOKEEPER_HOME=/opt/zookeeper
export PATH=$ZOOKEEPER_HOME/bin:$PATH
#执行初始化环境变量系统变量文件
source /etc/profile
启动zookeeper
/opt/zookeeper/bin/zkServer.sh start
部署canal Admin（192.168.1.155）
mkdir /opt/canal/{admin,server,adapter}
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.admin-1.1.5.tar.gz
tar -xzvf canal.admin-1.1.5.tar.gz -C /opt/canal/admin
初始化 Canal Admin 元数据库
将解压后的压缩包conf里面的canal_manager.sql挪到元数据库那台机器后，进行导入。
Canal Admin 配置文件
修改配置文件：vim /opt/canal/admin/conf/application.yml
server:
port: 8089
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
spring.datasource:
address: 192.168.1.151:3306 #元数据库信息
database: canal_manager #库名
username: root #用户
password: 123456 #密码
driver-class-name: com.mysql.jdbc.Driver
url: jdbc:mysql://${spring.datasource.address}/${spring.datasource.database}?useUnicode=true&characterEncoding=UTF-8&useSSL=false
hikari:
maximum-pool-size: 30
minimum-idle: 1
canal:
adminUser: admin
adminPasswd: admin
启动 Canal Admin
/opt/canal/admin/bin/startup.sh
浏览器输入192.168.1.155:8089,访问管理界面，默认账号：admin，密码 ：123456
部署canal server（192.168.1.155、192.168.1.156）
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.deployer-1.1.5.tar.gz
tar -zxvf canal.deployer-1.1.5.tar.gz -C /opt/canal/server
因为是使用canal Admin集群部署的，所以server节点只需要配置Admin的信息即可；统一的配置文件是由Admin界面管理，编辑 /opt/canal/server/conf/canal_local.properties 文件
server 1的配置文件
# register ip
canal.register.ip = 192.168.1.155
# canal admin config
canal.admin.manager = 192.168.1.155:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = canal-cluster-1
canal.admin.register.name = canal-server-1
server 2的配置文件
# register ip
canal.register.ip = 192.168.1.156
# canal admin config
canal.admin.manager = 192.168.1.155:8089
canal.admin.port = 11110
canal.admin.user = admin
canal.admin.passwd = 4ACFE3202A5FF5CF467898FC58AAB1D615029441
# admin auto register
canal.admin.register.auto = true
canal.admin.register.cluster = canal-cluster-1
canal.admin.register.name = canal-server-2
Canal Admin 创建集群
进入admin管理界面，新建集群，配置zookeeper信息
点击主配置
点击载入模板、暂时不用修改配置文件，点击保存即可
启动server
/opt/canal/server/bin/startup.sh local
在 Admin 上查看注册的 Canal Server。
设置mysql同步mysql配置。
server 默认情况下把源库对表结构修改的记录存储在本地 H2 数据库中，当 server 主备切换后会导致新的 server 无法正常同步数据，因此修改 TSDB 的设置将外部 MySQL 数据库作为 server 存储表结构变更信息的库。在admin界面选择相对应的集群，修改配置文件。
修改地方如下：
点击保存。
在元数据库上创建canal_tsdb库。
创建 Instance
在 Instance 管理界面，点击新建 Instance
点击载入模板后，修改以下信息
点击保存，查看Instance 日志，能正常读取到binlog以及偏移量为正常。
部署canal adapter（192.168.1.155、192.168.1.156）
wget https://github.com/alibaba/canal/releases/download/canal-1.1.5/canal.adapter-1.1.5.tar.gz
tar -zxvf canal.adapter-1.1.5.tar.gz -C /opt/canal/adapter
server 只支持将数据输出到消息队列中，因此需要额外部署 Adapter 订阅 server 中的消息，然后写入目标 MySQL。编辑/opt/canal/adapter/conf/application.yml
server:
port: 8081
spring:
jackson:
date-format: yyyy-MM-dd HH:mm:ss
time-zone: GMT+8
default-property-inclusion: non_null
canal.conf:
mode: tcp #tcp kafka rocketMQ rabbitMQ
flatMessage: true
zookeeperHosts:
syncBatchSize: 1000
retries: 0
timeout:
accessKey:
secretKey:
consumerProperties:
# canal tcp consumer
#canal.tcp.server.host: 127.0.0.1:11111
canal.tcp.server.host:
canal.tcp.zookeeper.hosts: 192.168.1.156:2181 #zookeeper地址
canal.tcp.batch.size: 500
canal.tcp.username:
canal.tcp.password:
# kafka consumer
kafka.bootstrap.servers: 127.0.0.1:9092
kafka.enable.auto.commit: false
kafka.auto.commit.interval.ms: 1000
kafka.auto.offset.reset: latest
kafka.request.timeout.ms: 40000
kafka.session.timeout.ms: 30000
kafka.isolation.level: read_committed
kafka.max.poll.records: 1000
# rocketMQ consumer
rocketmq.namespace:
rocketmq.namesrv.addr: 127.0.0.1:9876
rocketmq.batch.size: 1000
rocketmq.enable.message.trace: false
rocketmq.customized.trace.topic:
rocketmq.access.channel:
rocketmq.subscribe.filter:
# rabbitMQ consumer
rabbitmq.host:
rabbitmq.virtual.host:
rabbitmq.username:
rabbitmq.password:
rabbitmq.resource.ownerId:
#源库信息
srcDataSources:
defaultDS:
url: jdbc:mysql://192.168.1.154:3306/data?useUnicode=true
username: root
password: 123456
#目标库信息
canalAdapters:
- instance: data_test01 # canal instance Name or mq topic name
groups:
- groupId: g1
outerAdapters:
- name: rdb
key: mysql1
properties:
jdbc.driverClassName: com.mysql.jdbc.Driver
jdbc.url: jdbc:mysql://192.168.1.151:3306/data?useUnicode=true
jdbc.username: root
jdbc.password: 123456
配置rdb目录里面的文件
mv /opt/canal/adapter/conf/rdb/my_user.yml /opt/canal/adapter/conf/rdb/date_test01.yml
#配置文件如下
dataSourceKey: defaultDS
destination: data_test01
groupId: g1
outerAdapterKey: mysql1
concurrent: true
dbMapping:
database: data
table: test01
targetTable: test01
targetPk:
id: id
mapAll: true
mirrorDb: true
启动 Adapter
/opt/canal/adapter/bin/startup.sh
