具体参考:
https://github.com/alibaba/canal/wiki/QuickStart
准备
对于自建 MySQL , 需要先开启 Binlog 写入功能,配置 binlog-format 为 ROW 模式,my.cnf 中配置如下
[mysqld]
log-bin=mysql-bin # 开启 binlog
binlog-format=ROW # 选择 ROW 模式
server_id=1 # 配置 MySQL replaction 需要定义,不要和 canal 的 slaveId 重复
注意:针对阿里云 RDS for MySQL , 默认打开了 binlog , 并且账号默认具有 binlog dump 权限 , 不需要任何权限或者 binlog 设置,可以直接跳过这一步
授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限, 如果已有账户可直接 grant
CREATE USER canal IDENTIFIED BY 'canal';
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;
FLUSH PRIVILEGES;
启动
官方指导的配置是有问题的,启动canal-server会报以下错误:
^@2020-06-29 08:27:52.320 [destination = example , address = /127.0.0.1:3306 , EventParser] ERROR c.a.otter.canal.parse.inbound.mysql.MysqlEventParser - dump address /127.0.0.1:3306 has an error, retrying. caused by
java.io.IOException: connect /127.0.0.1:3306 failure:java.io.IOException: Unexpected End Stream
将
log-bin=【有权限的绝对路径】 # 开启 binlog
重启mysql后生效
下载 canal, 访问 release 页面 , 选择需要的包下载, 如以 1.0.17 版本为例
wget https://github.com/alibaba/canal/releases/download/canal-1.0.17/canal.deployer-1.0.17.tar.gz
解压缩
mkdir /tmp/canal
tar zxvf canal.deployer-$version.tar.gz -C /tmp/canal
解压完成后,进入 /tmp/canal 目录,可以看到如下结构
drwxr-xr-x 2 jianghang jianghang 136 2013-02-05 21:51 bin
drwxr-xr-x 4 jianghang jianghang 160 2013-02-05 21:51 conf
drwxr-xr-x 2 jianghang jianghang 1.3K 2013-02-05 21:51 lib
drwxr-xr-x 2 jianghang jianghang 48 2013-02-05 21:29 logs
配置修改
vi conf/example/instance.properties
## mysql serverId
canal.instance.mysql.slaveId = 1234
#position info,需要改成自己的数据库信息
canal.instance.master.address = 127.0.0.1:3306
canal.instance.master.journal.name =
canal.instance.master.position =
canal.instance.master.timestamp =
#canal.instance.standby.address =
#canal.instance.standby.journal.name =
#canal.instance.standby.position =
#canal.instance.standby.timestamp =
#username/password,需要改成自己的数据库信息
canal.instance.dbUsername = canal
canal.instance.dbPassword = canal
canal.instance.defaultDatabaseName =
canal.instance.connectionCharset = UTF-8
#table regex
canal.instance.filter.regex = .\*\\\\..\*
canal.instance.connectionCharset 代表数据库的编码方式对应到 java 中的编码类型,比如 UTF-8,GBK , ISO-8859-1
如果系统是1个 cpu,需要将 canal.instance.parser.parallel 设置为 false
启动
sh bin/startup.sh
查看 server 日志
vi logs/canal/canal.log</pre>
2013-02-05 22:45:27.967 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## start the canal server.
2013-02-05 22:45:28.113 [main] INFO com.alibaba.otter.canal.deployer.CanalController - ## start the canal server[10.1.29.120:11111]
2013-02-05 22:45:28.210 [main] INFO com.alibaba.otter.canal.deployer.CanalLauncher - ## the canal server is running now ......
查看 instance 的日志
vi logs/example/example.log
2013-02-05 22:50:45.636 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [canal.properties]
2013-02-05 22:50:45.641 [main] INFO c.a.o.c.i.spring.support.PropertyPlaceholderConfigurer - Loading properties file from class path resource [example/instance.properties]
2013-02-05 22:50:45.803 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start CannalInstance for 1-example
2013-02-05 22:50:45.810 [main] INFO c.a.otter.canal.instance.spring.CanalInstanceWithSpring - start successful....
关闭
sh bin/stop.sh
客户端:
package com.alibaba.test.steed.utils;
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", "", "");
// 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());
}
}
}
客户端消费:
empty count : 113
empty count : 114
================> binlog[mysql-bin.000001:382] , name[example,tb_user] , eventType : INSERT
id : 1 update=false
aid : 1 update=false
name : 1111 update=false
passwd : update=false
telephone : update=false
icon : update=false
email : update=false
title : update=false
status : 0 update=false
create_time : 2000-01-01 00:00:00 update=false
update_time : 2020-06-29 08:32:52 update=false
empty count : 1
empty count : 2
分享到:
相关推荐
《深入理解Canal客户端:canal.deployer-1.1.7-SNAPSHOT.tar.gz剖析》 Canal,由阿里开源的一款高效、稳定的数据库增量日志订阅与消费组件,广泛应用于数据同步、实时数据接入等场景。在本篇文章中,我们将深入探讨...
Linux 下 Canal 安装和使用 Canal 是一个基于 MySQL 数据库增量日志解析的工具,提供增量数据订阅和消费。 Canal 的主要用途包括数据库镜像、数据库实时备份、索引构建和实时维护、业务 Cache 刷新、带业务逻辑的...
《Canal Deployer 1.1.6:MySQL数据同步利器详解》 Canal Deployer是阿里巴巴开源的数据同步工具Canal的重要组成部分,主要用于部署和管理Canal实例。在1.1.6版本中,它提供了稳定高效的数据变更捕获与传输服务,...
canal-server(canal-deploy):可以直接监听MySQL的binlog,把自己伪装成MySQL的从库,只负责接收数据,并不做处理。 canal-adapter:相当于canal的客户端,会从canal-server中获取数据,然后对数据进行同步,可以...
阿里Canal是一款由阿里巴巴开源的、轻量级的数据变更订阅与推送组件,主要用来实现数据库的增量数据同步。在IT行业中,特别是在互联网公司,数据一致性是至关重要的,而Canal就是为了满足这种需求而诞生的。它能有效...
【标题】"Canal-1.1.6" 是一个重要的数据库同步工具,主要用于实现MySQL数据的实时增量复制。在IT行业中,数据同步是确保多系统间数据一致性和实时性的关键环节,尤其在分布式系统、大数据处理和云计算环境中。Canal...
canal.deployer-1.1.4.tar.gz canal.admin-1.1.4.tar.gz canal.adapter-1.1.4.tar.gz canal.example-1.1.4.tar.gz canal-canal-1.1.4.zip canal 全量工具包集合 https://juejin.cn/post/6864585112994971662
SpringBoot整合Canal是将阿里巴巴开源的数据库实时变更数据订阅与推送组件Canal与SpringBoot框架结合,以实现高效的数据同步。Canal主要用于数据库的实时增量数据同步,它能够监听MySQL数据库的binlog事件,当数据库...
【标题】"canal.adapter-1.1.6.zip" 提供的是Canal适配器的1.1.6版本,这是一个与数据库变更日志相关的软件组件。 【描述】"canal.adapter-1.1.6.zip" 指出这个压缩包包含的是Canal适配器的特定版本1.1.6,它是一个...
【标题】"canal-1.1.6-SNAPSHOT.rar" 提供的是一个名为 Canal 的数据同步工具的特定版本,即 1.1.6-SNAPSHOT 版本。这个压缩包是针对 Java 开发环境,特别是 JDK 17 的安装文件。这意味着在使用前,你需要确保你的...
《Canal Deployer 1.1.6:MySQL 数据库实时同步工具详解》 Canal Deployer 是阿里巴巴开源的一款高效、稳定、轻量级的 MySQL 数据库增量日志解析组件,其主要功能是实现 MySQL 数据库到其他系统(如 Elasticsearch...
Canal包含三个主要组件:Canal Server、Canal Client和Canal Adapter。Canal Server负责监听MySQL的Binlog事件,并将这些事件转化为结构化的数据。Canal Client可以订阅并消费这些数据,而Canal Adapter则扮演着数据...
【canal-canal-1.0.22_源码】是阿里巴巴开源的数据库实时增量数据订阅与同步工具,主要用于实现MySQL到其他数据库或者数据存储系统的实时数据迁移、同步,常用于构建数据仓库、日志收集等场景。Canal的核心功能是...
《Canal Deployer 1.1.4:MySQL数据库监听利器》 Canal Deployer 是阿里巴巴开源的一个数据库增量日志订阅与消费组件,主要用于数据库的实时数据同步。在这个场景下,`canal.deployer-1.1.4.tar.gz` 文件扮演了关键...
包含canal全套资源包含以下 canal.adapter-1.1.5.tar.gz canal.admin-1.1.5.tar.gz canal.deployer-1.1.5.tar.gz canal.example-1.1.5.tar.gz
【标题】"Canal Adapter 1.1.4版本" Canal是阿里巴巴开源的一个数据库增量日志订阅与消费组件,主要用于实现数据库的实时数据同步。Canal Adapter是其配套的适配器模块,用于将Canal捕获到的数据库变更事件转化为...
MySQL的Canal是一款开源的数据库实时同步工具,它主要用于数据复制和数据变更事件的捕获。在本场景中,我们关注的是Canal与特定MySQL版本——5.6.19的兼容性问题。通常,数据库版本升级或降级可能会导致与现有同步...
《阿里巴巴开源项目Canal详解与应用》 Canal,由阿里巴巴开源的一款高性能、轻量级的MySQL数据库实时增量数据订阅和消费组件,主要用于实现数据库的数据变更实时同步到其他系统,如大数据处理、缓存更新等场景。在...
【标题】"Canal安装包免费下载:MySQL同步ES工具" 【描述】"这款工具无需复杂的配置,只需在服务器上解压即可使用,对于Windows用户同样便捷,无需积分即可获取。" 【知识点详解】 1. **Canal**: Canal是阿里...
阿里Canal是一款由阿里巴巴开源的、轻量级的数据变更订阅与推送组件,主要用来解决数据库的增量数据同步问题。在大数据处理、实时计算以及数据仓库等场景中,Canal发挥着重要的作用。Canal-deployer是Canal的部署...